tomasz-syn-grzegorza
359afb3a59
Infrastructure configs for GMKtec K11 (Docker, vLLM, LocalAI, ComfyUI, control-plane, gpu-fan agent, Server UI with CLI/file explorer/GPU fan curve). Co-authored-by: Cursor <cursoragent@cursor.com>
249 lines
8.5 KiB
Python
249 lines
8.5 KiB
Python
"""Server filesystem browser — list, read, write, CRUD."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
import stat
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class FileExplorerError(Exception):
|
|
"""Raised for filesystem operations; map to HTTP in app.py."""
|
|
|
|
|
|
def _posix_error(exc: OSError, action: str) -> FileExplorerError:
|
|
if exc.errno in (13, 1):
|
|
return FileExplorerError(f"Brak uprawnień: {action}")
|
|
if exc.errno == 2:
|
|
return FileExplorerError("Nie znaleziono pliku lub katalogu")
|
|
if exc.errno == 17:
|
|
return FileExplorerError("Plik lub katalog już istnieje")
|
|
if exc.errno == 39:
|
|
return FileExplorerError("Katalog nie jest pusty")
|
|
if exc.errno == 20:
|
|
return FileExplorerError("To nie jest katalog")
|
|
return FileExplorerError(f"{action}: {exc.strerror or exc}")
|
|
|
|
|
|
class FileExplorer:
|
|
def __init__(self, root: str = "/", max_bytes: int = 2_097_152) -> None:
|
|
self.root = Path(root).resolve()
|
|
self.max_bytes = max_bytes
|
|
|
|
def resolve_path(self, raw: str) -> Path:
|
|
raw = (raw or "/").strip()
|
|
if not raw:
|
|
raw = "/"
|
|
if raw.startswith("/"):
|
|
target = Path(raw).resolve()
|
|
else:
|
|
target = (self.root / raw).resolve()
|
|
|
|
if self.root == Path("/"):
|
|
return target
|
|
|
|
root_s = str(self.root)
|
|
target_s = str(target)
|
|
if target != self.root and not target_s.startswith(root_s + os.sep):
|
|
raise FileExplorerError(f"Ścieżka poza dozwolonym katalogiem: {self.root}")
|
|
return target
|
|
|
|
def _access_flags(self, path: Path) -> tuple[bool, bool]:
|
|
readable = os.access(path, os.R_OK)
|
|
writable = os.access(path, os.W_OK)
|
|
return readable, writable
|
|
|
|
def list_directory(self, raw_path: str) -> dict[str, Any]:
|
|
path = self.resolve_path(raw_path)
|
|
if not path.exists():
|
|
raise FileExplorerError("Nie znaleziono pliku lub katalogu")
|
|
if not path.is_dir():
|
|
raise FileExplorerError("To nie jest katalog")
|
|
|
|
readable, writable = self._access_flags(path)
|
|
if not readable:
|
|
raise FileExplorerError("Brak uprawnień: odczyt katalogu")
|
|
|
|
entries: list[dict[str, Any]] = []
|
|
try:
|
|
names = sorted(path.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower()))
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "odczyt katalogu") from exc
|
|
|
|
for child in names:
|
|
try:
|
|
st = child.stat()
|
|
except OSError:
|
|
continue
|
|
is_dir = stat.S_ISDIR(st.st_mode)
|
|
c_readable, c_writable = self._access_flags(child)
|
|
entries.append(
|
|
{
|
|
"name": child.name,
|
|
"path": str(child),
|
|
"is_dir": is_dir,
|
|
"size": st.st_size if not is_dir else None,
|
|
"mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(),
|
|
"readable": c_readable,
|
|
"writable": c_writable,
|
|
}
|
|
)
|
|
|
|
parent = str(path.parent) if path != self.root else None
|
|
return {
|
|
"path": str(path),
|
|
"parent": parent,
|
|
"readable": readable,
|
|
"writable": writable,
|
|
"entries": entries,
|
|
}
|
|
|
|
def read_file(self, raw_path: str) -> dict[str, Any]:
|
|
path = self.resolve_path(raw_path)
|
|
if not path.exists():
|
|
raise FileExplorerError("Nie znaleziono pliku lub katalogu")
|
|
if path.is_dir():
|
|
raise FileExplorerError("To jest katalog, nie plik")
|
|
|
|
readable, writable = self._access_flags(path)
|
|
if not readable:
|
|
raise FileExplorerError("Brak uprawnień: odczyt pliku")
|
|
|
|
try:
|
|
size = path.stat().st_size
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "odczyt pliku") from exc
|
|
|
|
if size > self.max_bytes:
|
|
raise FileExplorerError(
|
|
f"Plik za duży ({size} B, limit {self.max_bytes} B)"
|
|
)
|
|
|
|
try:
|
|
data = path.read_bytes()
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "odczyt pliku") from exc
|
|
|
|
if b"\x00" in data:
|
|
return {
|
|
"path": str(path),
|
|
"binary": True,
|
|
"size": size,
|
|
"content_base64": base64.b64encode(data).decode("ascii"),
|
|
"readable": readable,
|
|
"writable": writable,
|
|
}
|
|
|
|
try:
|
|
content = data.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return {
|
|
"path": str(path),
|
|
"binary": True,
|
|
"size": size,
|
|
"content_base64": base64.b64encode(data).decode("ascii"),
|
|
"readable": readable,
|
|
"writable": writable,
|
|
}
|
|
|
|
return {
|
|
"path": str(path),
|
|
"binary": False,
|
|
"size": size,
|
|
"content": content,
|
|
"readable": readable,
|
|
"writable": writable,
|
|
}
|
|
|
|
def write_file(self, raw_path: str, content: str) -> dict[str, Any]:
|
|
path = self.resolve_path(raw_path)
|
|
encoded = content.encode("utf-8")
|
|
if len(encoded) > self.max_bytes:
|
|
raise FileExplorerError(
|
|
f"Zawartość za duża ({len(encoded)} B, limit {self.max_bytes} B)"
|
|
)
|
|
|
|
parent = path.parent
|
|
if not parent.exists():
|
|
raise FileExplorerError("Katalog nadrzędny nie istnieje")
|
|
if not os.access(parent, os.W_OK):
|
|
raise FileExplorerError("Brak uprawnień: zapis pliku")
|
|
|
|
if path.exists() and path.is_dir():
|
|
raise FileExplorerError("To jest katalog, nie plik")
|
|
if path.exists() and not os.access(path, os.W_OK):
|
|
raise FileExplorerError("Brak uprawnień: zapis pliku")
|
|
|
|
try:
|
|
path.write_text(content, encoding="utf-8")
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "zapis pliku") from exc
|
|
|
|
return {"ok": True, "path": str(path), "size": len(encoded)}
|
|
|
|
def mkdir(self, raw_path: str) -> dict[str, Any]:
|
|
path = self.resolve_path(raw_path)
|
|
parent = path.parent
|
|
if not parent.exists():
|
|
raise FileExplorerError("Katalog nadrzędny nie istnieje")
|
|
if not os.access(parent, os.W_OK):
|
|
raise FileExplorerError("Brak uprawnień: tworzenie katalogu")
|
|
if path.exists():
|
|
raise FileExplorerError("Plik lub katalog już istnieje")
|
|
|
|
try:
|
|
path.mkdir(parents=False, exist_ok=False)
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "tworzenie katalogu") from exc
|
|
|
|
return {"ok": True, "path": str(path)}
|
|
|
|
def delete_path(self, raw_path: str) -> dict[str, Any]:
|
|
path = self.resolve_path(raw_path)
|
|
if path == self.root:
|
|
raise FileExplorerError("Nie można usunąć katalogu głównego")
|
|
if not path.exists():
|
|
raise FileExplorerError("Nie znaleziono pliku lub katalogu")
|
|
if not os.access(path, os.W_OK):
|
|
raise FileExplorerError("Brak uprawnień: usuwanie")
|
|
|
|
if path.is_dir():
|
|
try:
|
|
if any(path.iterdir()):
|
|
raise FileExplorerError("Katalog nie jest pusty")
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "odczyt katalogu") from exc
|
|
|
|
try:
|
|
if path.is_dir():
|
|
path.rmdir()
|
|
else:
|
|
path.unlink()
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "usuwanie") from exc
|
|
|
|
return {"ok": True, "path": str(path)}
|
|
|
|
def rename_path(self, raw_from: str, raw_to: str) -> dict[str, Any]:
|
|
src = self.resolve_path(raw_from)
|
|
dst = self.resolve_path(raw_to)
|
|
if not src.exists():
|
|
raise FileExplorerError("Nie znaleziono pliku lub katalogu")
|
|
if dst.exists():
|
|
raise FileExplorerError("Docelowa ścieżka już istnieje")
|
|
if not os.access(src, os.W_OK):
|
|
raise FileExplorerError("Brak uprawnień: zmiana nazwy")
|
|
parent = dst.parent
|
|
if not parent.exists() or not os.access(parent, os.W_OK):
|
|
raise FileExplorerError("Brak uprawnień: zapis w katalogu docelowym")
|
|
|
|
try:
|
|
src.rename(dst)
|
|
except OSError as exc:
|
|
raise _posix_error(exc, "zmiana nazwy") from exc
|
|
|
|
return {"ok": True, "from": str(src), "to": str(dst)}
|