Initial import: bare-metal stacks, Server UI, GPU fan, tutorials.
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>
This commit is contained in:
tomasz-syn-grzegorza
@@ -0,0 +1,248 @@
|
||||
"""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)}
|
||||
Reference in New Issue
Block a user