mirror of
https://git.srhtnin.garden/sarah/SaraView.git
synced 2026-10-13 18:28:14 +02:00
470 lines
20 KiB
Python
470 lines
20 KiB
Python
"""Where files come from: the local disk (one root folder, set in Settings) or
|
|
a SaraView server's HTTP API. Both speak the same terms as the server: a
|
|
Root name plus a "/"-separated Path inside it ("" is the root itself), and
|
|
entries shaped like the server's /api/tree items, so the rest of the client
|
|
never needs to know which one it's talking to."""
|
|
|
|
import fnmatch
|
|
import shutil
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional
|
|
|
|
import httpx
|
|
import send2trash
|
|
|
|
from Core import AppConfig, CacheDir
|
|
|
|
SessionCookieName = "SaraViewSession" # Backend/Auth.py
|
|
|
|
# Backend/Explorer.py's FileTypeMap and Backend/Preview.py's TextFileTypes,
|
|
# so local files get the same types (and so the same viewer) as server ones.
|
|
FileTypeMap = {
|
|
".md": "markdown", ".markdown": "markdown", ".txt": "text",
|
|
".toml": "config", ".yaml": "config", ".yml": "config", ".json": "config", ".cfg": "config", ".ini": "config",
|
|
".html": "html", ".htm": "html",
|
|
".png": "image", ".jpg": "image", ".jpeg": "image", ".gif": "image", ".webp": "image", ".svg": "image",
|
|
".mp4": "video", ".webm": "video", ".mov": "video", ".mkv": "video",
|
|
}
|
|
TextFileTypes = {"markdown", "text", "config", "html"}
|
|
|
|
NoteFilePrefix = ".sv_" # a directory's own note: <Dir>/.sv_<Dir>.md
|
|
DefaultNoteExtension = ".md"
|
|
|
|
# Backend/Explorer.py's chunked upload limits (and Frontend/src/Api.js's).
|
|
UploadChunkSizeBytes = 20 * 1024 * 1024
|
|
ChunkedUploadThresholdBytes = 40 * 1024 * 1024
|
|
|
|
|
|
def ClassifyFile(Name: str) -> str:
|
|
return FileTypeMap.get(Path(Name).suffix.lower(), "unknown")
|
|
|
|
|
|
def JoinPath(DirectoryPath: str, Name: str) -> str:
|
|
return f"{DirectoryPath}/{Name}" if DirectoryPath else Name
|
|
|
|
|
|
def ParentPath(RelativePath: str) -> str:
|
|
return RelativePath.rsplit("/", 1)[0] if "/" in RelativePath else ""
|
|
|
|
|
|
def BaseName(RelativePath: str) -> str:
|
|
return RelativePath.rsplit("/", 1)[-1]
|
|
|
|
|
|
def NormalizeRelativePath(BaseDirectory: str, RawPath: str) -> str:
|
|
"""Backend/Core.py's: joins and folds "." and ".." without leaving the root."""
|
|
Parts: list[str] = []
|
|
for Part in JoinPath(BaseDirectory, RawPath).split("/"):
|
|
if Part in ("", "."):
|
|
continue
|
|
if Part == "..":
|
|
if Parts:
|
|
Parts.pop()
|
|
continue
|
|
Parts.append(Part)
|
|
return "/".join(Parts)
|
|
|
|
|
|
def MoveDestination(RelativePath: str, NewName: str) -> str:
|
|
"""Where Rename sends an entry: a plain name renames it in place, a name
|
|
with "/" in it moves it (relative to its directory), like SaraView's web
|
|
Rename."""
|
|
return NormalizeRelativePath(ParentPath(RelativePath), NewName)
|
|
|
|
|
|
@dataclass
|
|
class Entry:
|
|
Name: str
|
|
IsDirectory: bool
|
|
FileType: Optional[str] = None
|
|
Size: Optional[int] = None
|
|
Note: Optional[dict] = None # a directory's .sv_ note: {"Name", "FileType"}
|
|
|
|
@classmethod
|
|
def FromDict(cls, Data: dict[str, Any]) -> "Entry":
|
|
return cls(Data["Name"], bool(Data.get("IsDirectory")), Data.get("FileType"), Data.get("Size"), Data.get("Note"))
|
|
|
|
|
|
class SourceError(Exception):
|
|
"""Status is the server's HTTP status (0 for local or network errors);
|
|
401 means "log in first"."""
|
|
|
|
def __init__(self, Message: str, Status: int = 0):
|
|
super().__init__(Message)
|
|
self.Status = Status
|
|
|
|
|
|
class Source:
|
|
"""What the browser needs from a place files live."""
|
|
|
|
Kind = ""
|
|
|
|
def Describe(self) -> str: ...
|
|
def Roots(self) -> list[str]: ...
|
|
def List(self, Root: str, RelativePath: str) -> list[Entry]: ...
|
|
def ReadText(self, Root: str, RelativePath: str) -> str: ...
|
|
def ReadBytes(self, Root: str, RelativePath: str) -> bytes: ...
|
|
def WriteText(self, Root: str, RelativePath: str, Content: str) -> None: ...
|
|
def Create(self, Root: str, RelativePath: str, IsDirectory: bool) -> None: ...
|
|
def Move(self, Root: str, RelativePath: str, Destination: str) -> None: ...
|
|
def Delete(self, Root: str, RelativePath: str) -> None: ...
|
|
def Upload(self, Root: str, DirectoryPath: str, LocalFile: Path, OnProgress: Optional[Callable[[float], None]] = None) -> str: ...
|
|
def LocalFile(self, Root: str, RelativePath: str) -> Path: ...
|
|
def Book(self, Root: str, RelativePath: str) -> dict: ...
|
|
def Graph(self, Root: str, RelativePath: str) -> dict: ...
|
|
|
|
@property
|
|
def CanWrite(self) -> bool:
|
|
return True
|
|
|
|
|
|
class LocalSource(Source):
|
|
"""One folder on this computer, shown as a single root named after it.
|
|
Follows the server's rules (directory notes, hidden SaraTag files) so a
|
|
folder looks the same whether SaraView serves it or not."""
|
|
|
|
Kind = "Local"
|
|
|
|
def __init__(self, Config: AppConfig):
|
|
self.Config = Config
|
|
self.RootPath = Config.LocalRootPath
|
|
self.RootName = self.RootPath.name or str(self.RootPath)
|
|
|
|
def Describe(self) -> str:
|
|
return f"Local: {self.RootPath}"
|
|
|
|
def Roots(self) -> list[str]:
|
|
if not self.RootPath.is_dir():
|
|
raise SourceError(f"The root directory {self.RootPath} doesn't exist. Pick another in Settings.")
|
|
return [self.RootName]
|
|
|
|
def _Resolve(self, Root: str, RelativePath: str) -> Path:
|
|
if Root != self.RootName:
|
|
raise SourceError(f"Unknown root: {Root}")
|
|
Candidate = (self.RootPath / RelativePath.lstrip("/")).resolve()
|
|
if Candidate != self.RootPath and self.RootPath not in Candidate.parents:
|
|
raise SourceError("Path escapes the root directory.")
|
|
return Candidate
|
|
|
|
def _IsInternal(self, Name: str) -> bool:
|
|
return self.Config.HideInternalFiles and any(fnmatch.fnmatch(Name, Pattern) for Pattern in self.Config.InternalPatterns)
|
|
|
|
@staticmethod
|
|
def _FindNote(DirectoryPath: Path) -> Optional[Path]:
|
|
Prefix = f"{NoteFilePrefix}{DirectoryPath.name}."
|
|
try:
|
|
return next((C for C in DirectoryPath.iterdir() if C.is_file() and C.name.startswith(Prefix)), None)
|
|
except OSError:
|
|
return None
|
|
|
|
def List(self, Root: str, RelativePath: str) -> list[Entry]:
|
|
DirectoryPath = self._Resolve(Root, RelativePath)
|
|
try:
|
|
Children = sorted(DirectoryPath.iterdir(), key=lambda P: (P.is_file(), P.name.casefold()))
|
|
except NotADirectoryError:
|
|
raise SourceError(f"{DirectoryPath.name} is not a directory.")
|
|
except OSError as Error:
|
|
raise SourceError(f"Can't read {DirectoryPath}: {Error.strerror or Error}")
|
|
Entries = []
|
|
for Child in Children:
|
|
if self._IsInternal(Child.name):
|
|
continue
|
|
if Child.name.startswith(NoteFilePrefix):
|
|
if not self.Config.ShowSvFiles:
|
|
continue
|
|
elif Child.name.startswith(".") and not self.Config.ShowHiddenFolders:
|
|
continue
|
|
try:
|
|
IsDirectory = Child.is_dir()
|
|
Size = None if IsDirectory else Child.stat().st_size
|
|
except OSError:
|
|
continue # a broken link or a file that vanished mid-listing
|
|
Note = None
|
|
if IsDirectory and (NotePath := self._FindNote(Child)):
|
|
Note = {"Name": NotePath.name, "FileType": ClassifyFile(NotePath.name)}
|
|
Entries.append(Entry(Child.name, IsDirectory, None if IsDirectory else ClassifyFile(Child.name), Size, Note))
|
|
return Entries
|
|
|
|
def ReadText(self, Root: str, RelativePath: str) -> str:
|
|
try:
|
|
return self._Resolve(Root, RelativePath).read_text(encoding="utf-8", errors="replace")
|
|
except OSError as Error:
|
|
raise SourceError(f"Can't read it: {Error.strerror or Error}")
|
|
|
|
def ReadBytes(self, Root: str, RelativePath: str) -> bytes:
|
|
try:
|
|
return self._Resolve(Root, RelativePath).read_bytes()
|
|
except OSError as Error:
|
|
raise SourceError(f"Can't read it: {Error.strerror or Error}")
|
|
|
|
def WriteText(self, Root: str, RelativePath: str, Content: str) -> None:
|
|
try:
|
|
self._Resolve(Root, RelativePath).write_text(Content, encoding="utf-8")
|
|
except OSError as Error:
|
|
raise SourceError(f"Couldn't save: {Error.strerror or Error}")
|
|
|
|
def Create(self, Root: str, RelativePath: str, IsDirectory: bool) -> None:
|
|
# Backend/Explorer.py's CreateEntry: a new directory gets its note.
|
|
EntryPath = self._Resolve(Root, RelativePath)
|
|
if EntryPath.exists():
|
|
raise SourceError(f"{EntryPath.name} already exists.")
|
|
try:
|
|
if IsDirectory:
|
|
EntryPath.mkdir()
|
|
(EntryPath / f"{NoteFilePrefix}{EntryPath.name}{DefaultNoteExtension}").touch()
|
|
else:
|
|
EntryPath.parent.mkdir(parents=True, exist_ok=True)
|
|
EntryPath.touch()
|
|
except OSError as Error:
|
|
raise SourceError(f"Couldn't create {EntryPath.name}: {Error.strerror or Error}")
|
|
|
|
def Move(self, Root: str, RelativePath: str, Destination: str) -> None:
|
|
# Backend/Explorer.py's MoveEntry: a directory's note follows its new name.
|
|
SourcePath = self._Resolve(Root, RelativePath)
|
|
DestinationPath = self._Resolve(Root, Destination)
|
|
if not SourcePath.exists():
|
|
raise SourceError(f"{SourcePath.name} does not exist.")
|
|
if DestinationPath.exists():
|
|
raise SourceError(f"{DestinationPath.name} already exists.")
|
|
if SourcePath.is_dir() and SourcePath in DestinationPath.parents:
|
|
raise SourceError(f"Can't move {SourcePath.name} into itself.")
|
|
OldName = SourcePath.name
|
|
try:
|
|
DestinationPath.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(SourcePath), str(DestinationPath))
|
|
if DestinationPath.is_dir():
|
|
OldNote = next((C for C in DestinationPath.iterdir()
|
|
if C.is_file() and C.name.startswith(f"{NoteFilePrefix}{OldName}.")), None)
|
|
if OldNote is not None:
|
|
OldNote.rename(DestinationPath / f"{NoteFilePrefix}{DestinationPath.name}{OldNote.suffix}")
|
|
except OSError as Error:
|
|
raise SourceError(f"Couldn't move {OldName}: {Error.strerror or Error}")
|
|
|
|
def Delete(self, Root: str, RelativePath: str) -> None:
|
|
# To the trash, unlike the server (which deletes for good): it's your own disk.
|
|
EntryPath = self._Resolve(Root, RelativePath)
|
|
if EntryPath == self.RootPath:
|
|
raise SourceError("Can't delete the root directory.")
|
|
try:
|
|
send2trash.send2trash(EntryPath)
|
|
except OSError as Error:
|
|
raise SourceError(f"Couldn't delete {EntryPath.name}: {Error}")
|
|
|
|
def Upload(self, Root: str, DirectoryPath: str, LocalFile: Path, OnProgress=None) -> str:
|
|
"""Copies a file in from elsewhere on this computer."""
|
|
Target = self._Resolve(Root, JoinPath(DirectoryPath, LocalFile.name))
|
|
if Target.exists():
|
|
raise SourceError(f"{LocalFile.name} already exists.")
|
|
try:
|
|
shutil.copy2(LocalFile, Target)
|
|
except OSError as Error:
|
|
raise SourceError(f"Couldn't copy {LocalFile.name}: {Error.strerror or Error}")
|
|
return Target.name
|
|
|
|
def LocalFile(self, Root: str, RelativePath: str) -> Path:
|
|
return self._Resolve(Root, RelativePath)
|
|
|
|
def Book(self, Root: str, RelativePath: str) -> dict:
|
|
return {} # SaraTag tags are only read through a server for now
|
|
|
|
def Graph(self, Root: str, RelativePath: str) -> dict:
|
|
"""The graph view's data, built here the way the server builds it."""
|
|
from Graph import BuildLocalGraph # Graph.py imports this module
|
|
|
|
return BuildLocalGraph(self, Root, RelativePath)
|
|
|
|
|
|
class ServerSource(Source):
|
|
"""A SaraView server, through the same /api endpoints the web frontend
|
|
uses (Frontend/src/Api.js). Logged out, it browses as Guest like the web
|
|
view does; anything that changes files needs a login."""
|
|
|
|
Kind = "Server"
|
|
|
|
def __init__(self, Config: AppConfig, Token: str = ""):
|
|
self.Config = Config
|
|
self.Url = Config.ServerUrl.strip().rstrip("/")
|
|
if self.Url and "://" not in self.Url:
|
|
self.Url = "http://" + self.Url
|
|
self.Client = httpx.Client(base_url=self.Url, timeout=httpx.Timeout(30, read=120), follow_redirects=True)
|
|
if Token:
|
|
self.Client.cookies.set(SessionCookieName, Token)
|
|
self.User: Optional[dict] = None # the server's /api/auth/me, None as Guest
|
|
|
|
@property
|
|
def Token(self) -> str:
|
|
return self.Client.cookies.get(SessionCookieName) or ""
|
|
|
|
@property
|
|
def CanWrite(self) -> bool:
|
|
return self.User is not None
|
|
|
|
def Describe(self) -> str:
|
|
Who = self.User["Username"] if self.User else "Guest"
|
|
return f"{self.Url} as {Who}"
|
|
|
|
def _Request(self, Method: str, Endpoint: str, **Arguments) -> httpx.Response:
|
|
if not self.Url:
|
|
raise SourceError("No server address yet. Set it in Settings > Connection.")
|
|
try:
|
|
Response = self.Client.request(Method, f"/api{Endpoint}", **Arguments)
|
|
except httpx.HTTPError as Error:
|
|
raise SourceError(f"Can't reach {self.Url}: {Error}")
|
|
if Response.is_success:
|
|
return Response
|
|
try:
|
|
Detail = Response.json().get("detail") or Response.reason_phrase
|
|
except (ValueError, AttributeError):
|
|
Detail = Response.reason_phrase
|
|
if Response.status_code == 401:
|
|
Detail = "Log in to do that." if self.User is None else "Your session ended. Log in again."
|
|
raise SourceError(str(Detail), Response.status_code)
|
|
|
|
def _Json(self, Method: str, Endpoint: str, **Arguments) -> Any:
|
|
return self._Request(Method, Endpoint, **Arguments).json()
|
|
|
|
@staticmethod
|
|
def _At(Root: str, RelativePath: str) -> dict[str, str]:
|
|
return {"Root": Root, "Path": RelativePath}
|
|
|
|
# --- Sessions ---------------------------------------------------------
|
|
|
|
def Connect(self) -> None:
|
|
"""Checks the server is there and whether the saved session still
|
|
works (sessions live in the server's memory, so a restart ends them)."""
|
|
self._Json("GET", "/auth/status")
|
|
self.User = None
|
|
if self.Token:
|
|
try:
|
|
self.User = self._Json("GET", "/auth/me")
|
|
except SourceError as Error:
|
|
if Error.Status != 401:
|
|
raise
|
|
self.Client.cookies.clear()
|
|
|
|
def Login(self, Username: str, Password: str) -> None:
|
|
self.Client.cookies.clear()
|
|
try:
|
|
self.User = self._Json("POST", "/auth/login", json={"Username": Username, "Password": Password})
|
|
except SourceError as Error:
|
|
if Error.Status == 401:
|
|
raise SourceError("Wrong username or password.", 401)
|
|
raise
|
|
if not self.Token:
|
|
self.User = None
|
|
raise SourceError("The server didn't hand out a session.")
|
|
|
|
def Logout(self) -> None:
|
|
try:
|
|
self._Request("POST", "/auth/logout")
|
|
except SourceError:
|
|
pass # logged out locally either way
|
|
self.Client.cookies.clear()
|
|
self.User = None
|
|
|
|
def Close(self) -> None:
|
|
self.Client.close()
|
|
|
|
# --- Files ------------------------------------------------------------
|
|
|
|
def Roots(self) -> list[str]:
|
|
return [Root["Name"] for Root in self._Json("GET", "/roots")]
|
|
|
|
def List(self, Root: str, RelativePath: str) -> list[Entry]:
|
|
return [Entry.FromDict(Item) for Item in self._Json("GET", "/tree", params=self._At(Root, RelativePath))]
|
|
|
|
def ReadText(self, Root: str, RelativePath: str) -> str:
|
|
return self._Request("GET", "/file", params=self._At(Root, RelativePath)).text
|
|
|
|
def ReadBytes(self, Root: str, RelativePath: str) -> bytes:
|
|
return self._Request("GET", "/file", params=self._At(Root, RelativePath)).content
|
|
|
|
def WriteText(self, Root: str, RelativePath: str, Content: str) -> None:
|
|
# The server records line blame for whoever saves.
|
|
self._Request("PUT", "/file", params=self._At(Root, RelativePath), content=Content.encode("utf-8"))
|
|
|
|
def Create(self, Root: str, RelativePath: str, IsDirectory: bool) -> None:
|
|
self._Request("POST", "/tree", params=self._At(Root, RelativePath), json={"IsDirectory": IsDirectory})
|
|
|
|
def Move(self, Root: str, RelativePath: str, Destination: str) -> None:
|
|
# Destination is root-relative here, but the server reads
|
|
# DestinationPath relative to the entry's own directory, so walk up
|
|
# from there to the root first.
|
|
Parent = ParentPath(RelativePath)
|
|
if ParentPath(Destination) == Parent:
|
|
Body = {"NewName": BaseName(Destination)}
|
|
else:
|
|
Depth = len(Parent.split("/")) if Parent else 0
|
|
Body = {"DestinationPath": "/".join([".."] * Depth + [Destination])}
|
|
self._Request("PATCH", "/tree", params=self._At(Root, RelativePath), json=Body)
|
|
|
|
def Delete(self, Root: str, RelativePath: str) -> None:
|
|
self._Request("DELETE", "/tree", params=self._At(Root, RelativePath))
|
|
|
|
def Upload(self, Root: str, DirectoryPath: str, LocalFile: Path, OnProgress=None) -> str:
|
|
Size = LocalFile.stat().st_size
|
|
Params = self._At(Root, DirectoryPath)
|
|
if Size <= ChunkedUploadThresholdBytes:
|
|
with LocalFile.open("rb") as Handle:
|
|
Result = self._Json("POST", "/upload", params=Params, files=[("Files", (LocalFile.name, Handle))])
|
|
if Result.get("Errors"):
|
|
raise SourceError("; ".join(Result["Errors"]))
|
|
return Result["Saved"][0]
|
|
# Large files go up in pieces, like the web view's UploadFileChunked.
|
|
UploadId = str(uuid.uuid4())
|
|
TotalChunks = max(1, -(-Size // UploadChunkSizeBytes))
|
|
with LocalFile.open("rb") as Handle:
|
|
for ChunkIndex in range(TotalChunks):
|
|
ChunkParams = {**Params, "UploadId": UploadId, "ChunkIndex": ChunkIndex,
|
|
"TotalChunks": TotalChunks, "FileName": LocalFile.name}
|
|
self._Request("POST", "/upload/chunk", params=ChunkParams,
|
|
files=[("Chunk", ("chunk", Handle.read(UploadChunkSizeBytes)))])
|
|
if OnProgress:
|
|
OnProgress((ChunkIndex + 1) / TotalChunks)
|
|
Done = {**Params, "UploadId": UploadId, "TotalChunks": TotalChunks, "FileName": LocalFile.name}
|
|
return self._Json("POST", "/upload/complete", params=Done)["Saved"]
|
|
|
|
def Download(self, Root: str, RelativePath: str, Destination: Path) -> None:
|
|
"""A file as-is, a directory as a .zip (the server builds it)."""
|
|
if not self.Url:
|
|
raise SourceError("No server address yet.")
|
|
Partial = Destination.with_name(Destination.name + ".part")
|
|
try:
|
|
with self.Client.stream("GET", "/api/download", params=self._At(Root, RelativePath)) as Response:
|
|
if not Response.is_success:
|
|
Response.read()
|
|
raise SourceError(f"Download failed: {Response.status_code} {Response.reason_phrase}", Response.status_code)
|
|
with Partial.open("wb") as Handle:
|
|
for Chunk in Response.iter_bytes(1024 * 1024):
|
|
Handle.write(Chunk)
|
|
Partial.replace(Destination)
|
|
except httpx.HTTPError as Error:
|
|
Partial.unlink(missing_ok=True)
|
|
raise SourceError(f"Download failed: {Error}")
|
|
except OSError as Error:
|
|
Partial.unlink(missing_ok=True)
|
|
raise SourceError(f"Couldn't write {Destination}: {Error.strerror or Error}")
|
|
|
|
def LocalFile(self, Root: str, RelativePath: str) -> Path:
|
|
"""A fresh copy in ~/.cache/SaraView, for opening in another program."""
|
|
Target = CacheDir / "Server" / Root / RelativePath
|
|
Target.parent.mkdir(parents=True, exist_ok=True)
|
|
self.Download(Root, RelativePath, Target)
|
|
return Target
|
|
|
|
def Book(self, Root: str, RelativePath: str) -> dict:
|
|
try:
|
|
return self._Json("GET", "/book", params=self._At(Root, RelativePath)) or {}
|
|
except SourceError:
|
|
return {} # tags are extra; a file without them still opens
|
|
|
|
def Graph(self, Root: str, RelativePath: str) -> dict:
|
|
# A big root takes the server a while to read through.
|
|
return self._Json("GET", "/graph", params=self._At(Root, RelativePath), timeout=120)
|
|
|
|
|
|
def MakeSource(Config: AppConfig, Token: str = "") -> Source:
|
|
return ServerSource(Config, Token) if Config.Mode == "Server" else LocalSource(Config)
|