"""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 io 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: /.sv_.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: ... def Diagram(self, Source: str, ThemePalette: dict[str, str], FontSize: int) -> bytes: ... @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) def Diagram(self, Source: str, ThemePalette: dict[str, str], FontSize: int) -> bytes: raise SourceError("Diagrams are drawn by a SaraView server.") 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: Sending = _CountingReader(Handle, Size, OnProgress) if OnProgress else Handle Result = self._Json("POST", "/upload", params=Params, files=[("Files", (LocalFile.name, Sending))]) 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} Chunk = Handle.read(UploadChunkSizeBytes) Sending = io.BytesIO(Chunk) if OnProgress: Before = ChunkIndex * UploadChunkSizeBytes Sending = _CountingReader(Sending, len(Chunk), lambda Fraction, B=Before, L=len(Chunk): OnProgress((B + L * Fraction) / Size)) self._Request("POST", "/upload/chunk", params=ChunkParams, files=[("Chunk", ("chunk", Sending))]) 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, OnProgress: Optional[Callable[[int, int], None]] = None) -> None: """A file as-is, a directory as a .zip (the server builds it). OnProgress(BytesReceived, TotalBytes) as it arrives (Total 0 if unknown).""" 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) Total = int(Response.headers.get("Content-Length") or 0) Received = 0 with Partial.open("wb") as Handle: for Chunk in Response.iter_bytes(1024 * 1024): Handle.write(Chunk) Received += len(Chunk) if OnProgress: OnProgress(Received, Total) 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) # --- Your account and settings on the server (ServerSettings.py) ------ # The same /api endpoints as the web view's Settings panel. def UserSettings(self) -> dict: return self._Json("GET", "/settings") def SaveUserSettings(self, Settings: dict) -> dict: return self._Json("PUT", "/settings", json=Settings) def ThemeNames(self) -> list[str]: return self._Json("GET", "/themes") def Theme(self, Name: str) -> dict: return self._Json("GET", "/theme", params={"Name": Name}) def SaveTheme(self, Name: str, Palette: dict) -> None: self._Json("PUT", "/theme", params={"Name": Name}, json={"Palette": Palette}) def DeleteTheme(self, Name: str) -> None: self._Json("DELETE", "/theme", params={"Name": Name}) def ColourOptions(self) -> list[str]: return self._Json("GET", "/profile/colours") def SetColour(self, Colour: str) -> dict: self.User = {**(self.User or {}), **self._Json("PUT", "/profile/colour", json={"Colour": Colour})} return self.User def UploadAvatar(self, LocalFile: Path) -> None: with LocalFile.open("rb") as Handle: self._Json("POST", "/profile/avatar", files=[("Avatar", (LocalFile.name, Handle))]) def ChangeUsername(self, NewUsername: str) -> dict: self.User = {**(self.User or {}), **self._Json("PUT", "/profile/username", json={"NewUsername": NewUsername})} return self.User def ChangePassword(self, CurrentPassword: str, NewPassword: str) -> None: self._Json("PUT", "/profile/password", json={"CurrentPassword": CurrentPassword, "NewPassword": NewPassword}) def Users(self) -> list[dict]: return self._Json("GET", "/users") def CreateUser(self, Username: str, Password: str, IsAdmin: bool, Colour: str) -> None: self._Json("POST", "/users", json={"Username": Username, "Password": Password, "IsAdmin": IsAdmin, "Colour": Colour}) def Groups(self) -> list[dict]: return self._Json("GET", "/groups") def CreateGroup(self, Name: str) -> None: self._Json("POST", "/groups", json={"Name": Name}) def DeleteGroup(self, Name: str) -> None: self._Json("DELETE", "/groups", params={"Name": Name}) def SetGroupMembers(self, Name: str, Members: list[str]) -> None: self._Json("PUT", "/groups/members", params={"Name": Name}, json={"Members": Members}) def Locks(self) -> list[dict]: return self._Json("GET", "/locks") def RemoveLock(self, Root: str, RelativePath: str) -> None: self._Json("DELETE", "/lock", params=self._At(Root, RelativePath)) def Spans(self, Root: str, RelativePath: str, Content: str) -> list[dict]: """Where Content's variables are and what they resolve to (Variables.py).""" return self._Json("POST", "/file/spans", params=self._At(Root, RelativePath), json={"Content": Content})["Spans"] def Diagram(self, Source: str, ThemePalette: dict[str, str], FontSize: int) -> bytes: """A ```mermaid block as a PNG, drawn by the server (Diagrams.py).""" Body = {"Source": Source, "Palette": ThemePalette, "FontSize": FontSize} return self._Request("POST", "/diagram", json=Body).content class _CountingReader: """A file handle that reports how much of it has been read (sent): httpx reads an upload's file in pieces, so this is its progress.""" def __init__(self, Handle, Size: int, OnProgress: Callable[[float], None]): self._Handle = Handle self._Size = Size self._OnProgress = OnProgress self._Sent = 0 def read(self, Count: int = -1) -> bytes: Data = self._Handle.read(Count) self._Sent += len(Data) if self._Size: self._OnProgress(min(1.0, self._Sent / self._Size)) return Data def __getattr__(self, Name: str): return getattr(self._Handle, Name) # seek, tell, fileno: for the length def MakeSource(Config: AppConfig, Token: str = "") -> Source: return ServerSource(Config, Token) if Config.Mode == "Server" else LocalSource(Config)