Add file upload and download (zip for directories)

Backend/Explorer.py

@@ -124,6 +124,28 @@ def MoveEntry(RootName: str, SourceRelativePath: str, DestinationRelativePath: s
         RenameDirectoryNote(DestinationPath, OldName)
 
 
+def SaveUploadedFile(
+    RootName: str, DirectoryRelativePath: str, FileName: str, FileBytes: bytes
+) -> str:
+    DirectoryPath = ResolvePath(RootName, DirectoryRelativePath)
+
+    if not DirectoryPath.is_dir():
+        raise NotADirectoryError(f"{DirectoryPath} is not a directory.")
+
+    SafeName = Path(FileName).name
+
+    if not SafeName or SafeName in (".", ".."):
+        raise ValueError("Invalid file name.")
+
+    TargetPath = DirectoryPath / SafeName
+
+    if TargetPath.exists():
+        raise FileExistsError(f"{SafeName} already exists.")
+
+    TargetPath.write_bytes(FileBytes)
+    return SafeName
+
+
 def DeleteEntry(RootName: str, RelativePath: str) -> None:
     EntryPath = ResolvePath(RootName, RelativePath)
 

Backend/Main.py

@@ -2,8 +2,12 @@ from __future__ import annotations
 
 import json
 import mimetypes
+import os
 import sys
+import tempfile
+import zipfile
 from pathlib import Path
+from urllib.parse import quote
 
 ApplicationDirectory = Path(__file__).resolve().parent
 if str(ApplicationDirectory) not in sys.path:
@@ -12,6 +16,7 @@ if str(ApplicationDirectory) not in sys.path:
 from fastapi import Depends, FastAPI, File, HTTPException, Request, Response, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse, StreamingResponse
+from starlette.background import BackgroundTask
 
 from Auth import (
     CreateSession,
@@ -34,6 +39,7 @@ from Explorer import (
     ListDirectory,
     ListRoots,
     MoveEntry,
+    SaveUploadedFile,
 )
 from Groups import (
     CreateGroup,
@@ -429,6 +435,36 @@ def PostTreeEndpoint(
     return {"Created": True}
 
 
+@Application.post("/api/upload")
+async def PostUploadEndpoint(
+    Root: str,
+    Path: str,
+    Files: list[UploadFile] = File(...),
+    CurrentUser: dict = Depends(GetCurrentUser),
+) -> dict:
+    RequirePathAccessible(Root, Path, CurrentUser)
+
+    SavedNames: list[str] = []
+    Errors: list[str] = []
+
+    for UploadedFile in Files:
+        DisplayName = UploadedFile.filename or "upload"
+
+        try:
+            FileBytes = await UploadedFile.read()
+            SavedName = SaveUploadedFile(Root, Path, DisplayName, FileBytes)
+            SavedNames.append(SavedName)
+        except (ValueError, FileExistsError, NotADirectoryError) as Error:
+            Errors.append(f"{DisplayName}: {Error}")
+        except OSError as Error:
+            Errors.append(f"{DisplayName}: Could not save file ({Error})")
+
+    if Errors and not SavedNames:
+        raise HTTPException(status_code=400, detail="; ".join(Errors))
+
+    return {"Saved": SavedNames, "Errors": Errors}
+
+
 class MoveEntryRequest(BaseModel):
     NewName: str | None = None
     DestinationPath: str | None = None
@@ -622,6 +658,81 @@ def GetRawFileEndpoint(
     return StreamingResponse(IterateFile(), media_type=MimeType)
 
 
+def _AddDirectoryToZip(
+    ZipHandle: zipfile.ZipFile,
+    RootName: str,
+    RelativePath: str,
+    Username: str,
+    UserGroups: list[str],
+    ArcPrefix: str,
+) -> None:
+    try:
+        Entries = ListDirectory(RootName, RelativePath, False, False)
+    except (ValueError, NotADirectoryError, FileNotFoundError):
+        return
+
+    for Entry in Entries:
+        EntryRelativePath = f"{RelativePath}/{Entry['Name']}" if RelativePath else Entry["Name"]
+
+        if not IsPathAccessible(RootName, EntryRelativePath, Username, UserGroups):
+            continue
+
+        ArcName = f"{ArcPrefix}/{Entry['Name']}" if ArcPrefix else Entry["Name"]
+
+        if Entry["IsDirectory"]:
+            _AddDirectoryToZip(ZipHandle, RootName, EntryRelativePath, Username, UserGroups, ArcName)
+        else:
+            ZipHandle.write(ResolvePath(RootName, EntryRelativePath), ArcName)
+
+
+@Application.get("/api/download")
+def GetDownloadEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)):
+    RequirePathAccessible(Root, Path, CurrentUser)
+
+    try:
+        EntryPath = ResolvePath(Root, Path)
+    except ValueError as Error:
+        raise HTTPException(status_code=400, detail=str(Error))
+
+    if not EntryPath.exists():
+        raise HTTPException(status_code=404, detail="Not found.")
+
+    Username = CurrentUser["Username"]
+    UserGroups = CurrentUser.get("Groups", [])
+    DownloadName = EntryPath.name or Root
+
+    if EntryPath.is_dir():
+        TempZip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False)
+        TempZip.close()
+
+        with zipfile.ZipFile(TempZip.name, "w", zipfile.ZIP_DEFLATED) as ZipHandle:
+            _AddDirectoryToZip(ZipHandle, Root, Path, Username, UserGroups, "")
+
+        EncodedName = quote(f"{DownloadName}.zip")
+        Headers = {"Content-Disposition": f"attachment; filename*=UTF-8''{EncodedName}"}
+
+        return FileResponse(
+            TempZip.name,
+            media_type="application/zip",
+            headers=Headers,
+            background=BackgroundTask(os.remove, TempZip.name),
+        )
+
+    if not EntryPath.is_file():
+        raise HTTPException(status_code=400, detail="Not a file or directory.")
+
+    MimeType = mimetypes.guess_type(str(EntryPath))[0] or "application/octet-stream"
+    EncodedName = quote(DownloadName)
+    Headers = {"Content-Disposition": f"attachment; filename*=UTF-8''{EncodedName}"}
+
+    def IterateDownload():
+        with EntryPath.open("rb") as SourceFile:
+            while Chunk := SourceFile.read(1024 * 1024):
+                yield Chunk
+
+    return StreamingResponse(IterateDownload(), media_type=MimeType, headers=Headers)
+
+
 @Application.get("/api/book")
 def GetBookEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> dict:
     RequirePathAccessible(Root, Path, CurrentUser)