SaraView/Backend/Main.py

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:
    sys.path.insert(0, str(ApplicationDirectory))

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,
    DeleteSession,
    GetCurrentUser,
    GetCurrentUserOrGuest,
    RequireAdmin,
    SessionCookieName,
)
from BookReader import ReadBookForFile
from Blame import LoadBlame, RenameUserInBlame, UpdateBlameOnSave
from Core import NormalizeRelativePath, ResolvePath
from Diagrams import DiagramError, DiagramsUnavailable, RenderDiagram
from Embeds import BuildOgTags
from Editor import WriteTextFile
from Explorer import (
    ClassifyFile,
    CreateEntry,
    DeleteEntry,
    EnsureDirectoryNote,
    FinishChunkedUpload,
    ListDirectory,
    ListRoots,
    MoveEntry,
    SaveUploadChunk,
    SaveUploadedFile,
)
from Groups import (
    CreateGroup,
    DeleteGroup,
    GuestGroupName,
    ListGroups,
    RenameUserInGroups,
    SetGroupMembers,
)
from Locks import (
    GetInheritedLock,
    GetLockForPath,
    IsPathAccessible,
    LoadLocks,
    RemoveLock,
    RequirePathAccessible,
    SetLock,
)
from Graph import BuildGraph
from Preview import GetMimeType, ReadTextFile, StreamBinaryFile, TextFileTypes
from Settings import (
    DefaultSettings,
    DeleteTheme,
    ListThemeNames,
    LoadSettings,
    LoadTheme,
    PaletteRoles,
    RenameUserSettings,
    SaveSettings,
    SaveTheme,
)
from Users import (
    AnyUsersExist,
    ChangePassword,
    CreateUser,
    GetAvatarPath,
    ListUsers,
    RenameUser,
    RenameUserAvatar,
    SaveAvatar,
    UpdateUserColour,
    UserColourOptions,
    VerifyLogin,
)
from VarFunctions import ResolvePortals
from pydantic import BaseModel


Application = FastAPI(title="SaraView")

Application.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True,
)


class LoginRequest(BaseModel):
    Username: str
    Password: str


@Application.get("/api/auth/status")
def AuthStatusEndpoint() -> dict:
    return {"SetupComplete": AnyUsersExist()}


@Application.post("/api/auth/bootstrap")
def BootstrapEndpoint(Body: LoginRequest) -> dict:
    if AnyUsersExist():
        raise HTTPException(status_code=403, detail="Setup already completed.")

    try:
        CreateUser(Body.Username, Body.Password, True)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Created": True}


@Application.post("/api/auth/login")
def LoginEndpoint(Body: LoginRequest, response: Response) -> dict:
    User = VerifyLogin(Body.Username, Body.Password)

    if User is None:
        raise HTTPException(status_code=401, detail="Invalid username or password.")

    Token = CreateSession(Body.Username)
    response.set_cookie(
        key=SessionCookieName,
        value=Token,
        httponly=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 30,
    )
    return User


@Application.post("/api/auth/logout")
def LogoutEndpoint(request: Request, response: Response) -> dict:
    Token = request.cookies.get(SessionCookieName)
    if Token:
        DeleteSession(Token)
    response.delete_cookie(SessionCookieName)
    return {"LoggedOut": True}


@Application.get("/api/auth/me")
def MeEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    return CurrentUser


class CreateUserRequest(BaseModel):
    Username: str
    Password: str
    IsAdmin: bool = False
    Colour: str = "Purple"


@Application.get("/api/users")
def GetUsersEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
    return ListUsers()


@Application.post("/api/users")
def PostUsersEndpoint(Body: CreateUserRequest, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
    try:
        CreateUser(Body.Username, Body.Password, Body.IsAdmin, Body.Colour)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Created": True}


@Application.get("/api/profile/colours")
def GetProfileColourOptionsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[str]:
    return UserColourOptions


class UpdateColourRequest(BaseModel):
    Colour: str


@Application.put("/api/profile/colour")
def PutProfileColourEndpoint(
    Body: UpdateColourRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    try:
        UpdateUserColour(CurrentUser["Username"], Body.Colour)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Username": CurrentUser["Username"], "IsAdmin": CurrentUser["IsAdmin"], "Colour": Body.Colour}


@Application.post("/api/profile/avatar")
async def PostProfileAvatarEndpoint(
    Avatar: UploadFile = File(...), CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    Extension = Path(Avatar.filename or "").suffix.lower() or ".png"
    FileBytes = await Avatar.read()

    try:
        SaveAvatar(CurrentUser["Username"], FileBytes, Extension)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Saved": True}


@Application.get("/api/profile/avatar")
def GetProfileAvatarEndpoint(Username: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)):
    AvatarPath = GetAvatarPath(Username)

    if AvatarPath is None:
        raise HTTPException(status_code=404, detail="No avatar set.")

    MimeType = mimetypes.guess_type(str(AvatarPath))[0] or "application/octet-stream"
    return StreamingResponse(iter([AvatarPath.read_bytes()]), media_type=MimeType)


class ChangeUsernameRequest(BaseModel):
    NewUsername: str


@Application.put("/api/profile/username")
def PutProfileUsernameEndpoint(
    Body: ChangeUsernameRequest,
    request: Request,
    response: Response,
    CurrentUser: dict = Depends(GetCurrentUser),
) -> dict:
    OldUsername = CurrentUser["Username"]
    NewUsername = Body.NewUsername.strip()

    try:
        RenameUser(OldUsername, NewUsername)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if NewUsername != OldUsername:
        RenameUserSettings(OldUsername, NewUsername)
        RenameUserAvatar(OldUsername, NewUsername)
        RenameUserInBlame(OldUsername, NewUsername)
        RenameUserInGroups(OldUsername, NewUsername)

        OldToken = request.cookies.get(SessionCookieName)
        if OldToken:
            DeleteSession(OldToken)

        NewToken = CreateSession(NewUsername)
        response.set_cookie(
            key=SessionCookieName,
            value=NewToken,
            httponly=True,
            samesite="lax",
            max_age=60 * 60 * 24 * 30,
        )

    return {"Username": NewUsername, "IsAdmin": CurrentUser["IsAdmin"], "Colour": CurrentUser["Colour"]}


class ChangePasswordRequest(BaseModel):
    CurrentPassword: str
    NewPassword: str


@Application.put("/api/profile/password")
def PutProfilePasswordEndpoint(
    Body: ChangePasswordRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    try:
        ChangePassword(CurrentUser["Username"], Body.CurrentPassword, Body.NewPassword)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Changed": True}


@Application.get("/api/groups")
def GetGroupsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
    return ListGroups()


class CreateGroupRequest(BaseModel):
    Name: str


@Application.post("/api/groups")
def PostGroupsEndpoint(Body: CreateGroupRequest, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
    try:
        CreateGroup(Body.Name)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Created": True}


@Application.delete("/api/groups")
def DeleteGroupsEndpoint(Name: str, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
    try:
        DeleteGroup(Name)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Deleted": True}


class SetGroupMembersRequest(BaseModel):
    Members: list[str]


@Application.put("/api/groups/members")
def PutGroupMembersEndpoint(
    Name: str, Body: SetGroupMembersRequest, CurrentUser: dict = Depends(RequireAdmin)
) -> dict:
    try:
        SetGroupMembers(Name, Body.Members)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Saved": True}


@Application.get("/api/lock")
def GetLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)
    return GetLockForPath(Root, Path) or {}


@Application.get("/api/lock/inherited")
def GetInheritedLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)
    return GetInheritedLock(Root, Path) or {}


class SetLockRequest(BaseModel):
    Users: list[str] = []
    Groups: list[str] = []


@Application.put("/api/lock")
def PutLockEndpoint(
    Root: str, Path: str, Body: SetLockRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        SetLock(Root, Path, Body.Users, Body.Groups)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Locked": True}


@Application.delete("/api/lock")
def DeleteLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)
    RemoveLock(Root, Path)
    return {"Unlocked": True}


@Application.get("/api/locks")
def GetLocksEndpoint(CurrentUser: dict = Depends(RequireAdmin)) -> list[dict]:
    return LoadLocks()


@Application.get("/api/roots")
def GetRootsEndpoint(CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> list[dict]:
    return ListRoots()


@Application.get("/api/tree")
def GetTreeEndpoint(
    Root: str, Path: str = "", CurrentUser: dict = Depends(GetCurrentUserOrGuest)
) -> list[dict]:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        UserSettings = (
            DefaultSettings if CurrentUser["Username"] is None else LoadSettings(CurrentUser["Username"])
        )
        ShowHidden = UserSettings.get("ShowHiddenFolders", False)
        ShowSvFiles = UserSettings.get("ShowSvFiles", False)
        Entries = ListDirectory(Root, Path, ShowHidden, ShowSvFiles)
    except (ValueError, NotADirectoryError, FileNotFoundError) as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    Username = CurrentUser["Username"]
    UserGroups = CurrentUser.get("Groups", [])

    return [
        Entry
        for Entry in Entries
        if IsPathAccessible(
            Root,
            f"{Path}/{Entry['Name']}" if Path else Entry["Name"],
            Username,
            UserGroups,
        )
    ]


@Application.get("/api/graph")
def GetGraphEndpoint(
    Root: str, Path: str = "", CurrentUser: dict = Depends(GetCurrentUserOrGuest)
) -> dict:
    """Directories, files and the links between them under Path (see Graph.py)."""
    RequirePathAccessible(Root, Path, CurrentUser)
    UserSettings = DefaultSettings if CurrentUser["Username"] is None else LoadSettings(CurrentUser["Username"])

    try:
        return BuildGraph(
            Root,
            Path,
            UserSettings.get("ShowHiddenFolders", False),
            UserSettings.get("ShowSvFiles", False),
            CurrentUser["Username"],
            CurrentUser.get("Groups", []),
        )
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))


@Application.post("/api/note")
def PostNoteEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        DirectoryPath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not DirectoryPath.is_dir():
        raise HTTPException(status_code=400, detail="Not a directory.")

    NotePath = EnsureDirectoryNote(DirectoryPath)
    return {"Name": NotePath.name, "FileType": ClassifyFile(NotePath)}


class CreateEntryRequest(BaseModel):
    IsDirectory: bool


@Application.post("/api/tree")
def PostTreeEndpoint(
    Root: str, Path: str, Body: CreateEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        CreateEntry(Root, Path, Body.IsDirectory)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileExistsError as Error:
        raise HTTPException(status_code=409, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not create entry: {Error}")

    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}


@Application.post("/api/upload/chunk")
async def PostUploadChunkEndpoint(
    Root: str,
    Path: str,
    UploadId: str,
    ChunkIndex: int,
    TotalChunks: int,
    FileName: str,
    Chunk: UploadFile = File(...),
    CurrentUser: dict = Depends(GetCurrentUser),
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        ChunkBytes = await Chunk.read()
        SaveUploadChunk(UploadId, ChunkIndex, ChunkBytes)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Received": ChunkIndex}


@Application.post("/api/upload/complete")
def PostUploadCompleteEndpoint(
    Root: str,
    Path: str,
    UploadId: str,
    TotalChunks: int,
    FileName: str,
    CurrentUser: dict = Depends(GetCurrentUser),
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        SavedName = FinishChunkedUpload(Root, Path, UploadId, TotalChunks, FileName)
    except (ValueError, FileExistsError, NotADirectoryError) as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=400, detail=f"Could not save file ({Error})")

    return {"Saved": SavedName}


class MoveEntryRequest(BaseModel):
    NewName: str | None = None
    DestinationPath: str | None = None


@Application.patch("/api/tree")
def PatchTreeEndpoint(
    Root: str, Path: str, Body: MoveEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    ParentPath = Path.rsplit("/", 1)[0] if "/" in Path else ""

    if Body.DestinationPath:
        Destination = NormalizeRelativePath(ParentPath, Body.DestinationPath)
    elif Body.NewName:
        Destination = f"{ParentPath}/{Body.NewName}" if ParentPath else Body.NewName
    else:
        raise HTTPException(status_code=400, detail="Provide NewName or DestinationPath.")

    RequirePathAccessible(Root, Destination, CurrentUser)

    try:
        MoveEntry(Root, Path, Destination)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))
    except FileExistsError as Error:
        raise HTTPException(status_code=409, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not move entry: {Error}")

    return {"Moved": True, "NewPath": Destination}


@Application.delete("/api/tree")
def DeleteTreeEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        DeleteEntry(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not delete entry: {Error}")

    return {"Deleted": True}


@Application.get("/api/file")
def GetFileEndpoint(
    Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUserOrGuest)
):
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not FilePath.is_file():
        raise HTTPException(status_code=404, detail="File not found.")

    FileType = ClassifyFile(FilePath)

    if FileType in TextFileTypes:
        return PlainTextResponse(ReadTextFile(FilePath))

    return StreamBinaryFile(FilePath, request.headers.get("range"))


@Application.get("/api/file/resolved")
def GetResolvedFileEndpoint(
    Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not FilePath.is_file():
        raise HTTPException(status_code=404, detail="File not found.")

    FileType = ClassifyFile(FilePath)

    if FileType not in TextFileTypes:
        raise HTTPException(status_code=400, detail="Not a text file.")

    Content = ReadTextFile(FilePath)
    Resolved = ResolvePortals(
        Root, Path, Content, CurrentUser["Username"], CurrentUser.get("Groups", [])
    )
    return {"Content": Resolved}


@Application.put("/api/file")
async def PutFileEndpoint(
    Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
):
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    OldContent = ""
    if FilePath.is_file():
        try:
            OldContent = FilePath.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError):
            OldContent = ""

    Content = (await request.body()).decode("utf-8")

    try:
        WriteTextFile(FilePath, Content)
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not write file: {Error}")

    UpdateBlameOnSave(Root, Path, OldContent, Content, CurrentUser["Username"])

    return {"Saved": True}


@Application.get("/api/blame")
def GetBlameEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    RawBlame = LoadBlame(Root, Path)
    AllUsers = {User["Username"]: User for User in ListUsers()}
    ViewerUsername = CurrentUser["Username"]
    Enriched: dict[str, dict] = {}

    for LineNumber, Entry in RawBlame.items():
        AuthorUsername = Entry.get("Username")
        Timestamp = Entry.get("Timestamp")

        if AuthorUsername == ViewerUsername:
            IsVisible = True
        elif AuthorUsername in AllUsers:
            IsVisible = LoadSettings(AuthorUsername).get("VisibleInBlame", True)
        else:
            IsVisible = True

        if IsVisible:
            AuthorRecord = AllUsers.get(AuthorUsername, {})
            Enriched[LineNumber] = {
                "Username": AuthorUsername,
                "Timestamp": Timestamp,
                "Colour": AuthorRecord.get("Colour", "Purple"),
                "HasAvatar": AuthorUsername is not None and GetAvatarPath(AuthorUsername) is not None,
            }
        else:
            Enriched[LineNumber] = {"Username": None, "Hidden": True, "Timestamp": Timestamp}

    return Enriched


@Application.get("/api/raw/{RootName}/{FilePath:path}")
def GetRawFileEndpoint(
    RootName: str, FilePath: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)
):
    RequirePathAccessible(RootName, FilePath, CurrentUser)

    try:
        ResolvedPath = ResolvePath(RootName, FilePath)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not ResolvedPath.is_file():
        raise HTTPException(status_code=404, detail="File not found.")

    MimeType = mimetypes.guess_type(str(ResolvedPath))[0] or "application/octet-stream"

    def IterateFile():
        with ResolvedPath.open("rb") as SourceFile:
            while Chunk := SourceFile.read(1024 * 1024):
                yield Chunk

    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}"}

    # FileResponse sends Content-Length, so browsers can show the size and
    # time left (a plain stream showed them as unknown).
    return FileResponse(EntryPath, media_type=MimeType, headers=Headers)


@Application.get("/api/book")
def GetBookEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    Book = ReadBookForFile(FilePath)
    return Book or {}


@Application.get("/api/settings")
def GetSettingsEndpoint(CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> dict:
    if CurrentUser["Username"] is None:
        return dict(DefaultSettings)

    return LoadSettings(CurrentUser["Username"])


@Application.put("/api/settings")
def PutSettingsEndpoint(NewSettings: dict, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    return SaveSettings(CurrentUser["Username"], NewSettings)


class DiagramRequest(BaseModel):
    Source: str
    Palette: dict[str, str] = {}
    FontSize: int = 14


# A ```mermaid block as a PNG, for the desktop client (the web view draws its
# own). Palette is the client's theme by role, so it matches the client.
@Application.post("/api/diagram")
async def PostDiagramEndpoint(Body: DiagramRequest, CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> Response:
    try:
        Png = await RenderDiagram(Body.Source, Body.Palette, Body.FontSize)
    except DiagramError as Error:
        raise HTTPException(status_code=422, detail=str(Error))
    except DiagramsUnavailable as Error:
        raise HTTPException(status_code=503, detail=str(Error))
    return Response(content=Png, media_type="image/png")


@Application.get("/api/themes")
def GetThemesEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[str]:
    return ListThemeNames()


@Application.get("/api/theme")
def GetThemeEndpoint(Name: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> dict:
    try:
        return LoadTheme(Name)
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))


class SaveThemeRequest(BaseModel):
    Palette: dict[str, str]


@Application.put("/api/theme")
def PutThemeEndpoint(
    Name: str, Body: SaveThemeRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    try:
        SaveTheme(Name, Body.Palette)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Saved": True, "Roles": PaletteRoles}


@Application.delete("/api/theme")
def DeleteThemeEndpoint(Name: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    try:
        DeleteTheme(Name)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))

    return {"Deleted": True}


@Application.get("/public/media/{RootName}/{FilePath:path}")
def GetPublicMediaEndpoint(RootName: str, FilePath: str, request: Request):
    if not IsPathAccessible(RootName, FilePath, "", [GuestGroupName]):
        raise HTTPException(status_code=404, detail="Not found.")

    try:
        ResolvedPath = ResolvePath(RootName, FilePath)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not ResolvedPath.is_file():
        raise HTTPException(status_code=404, detail="File not found.")

    return StreamBinaryFile(ResolvedPath, request.headers.get("range"))


def GetForwardedScheme(request: Request) -> str:
    ForwardedProto = request.headers.get("x-forwarded-proto")
    if ForwardedProto:
        return ForwardedProto.split(",")[0].strip()

    CfVisitor = request.headers.get("cf-visitor")
    if CfVisitor:
        try:
            ParsedScheme = json.loads(CfVisitor).get("scheme")
            if ParsedScheme:
                return ParsedScheme
        except (json.JSONDecodeError, AttributeError):
            pass

    return request.url.scheme


FrontendDist = ApplicationDirectory.parent / "Frontend" / "dist"

if FrontendDist.is_dir():
    IndexHtmlTemplate = (FrontendDist / "index.html").read_text(encoding="utf-8")

    @Application.get("/{FullPath:path}")
    def ServeFrontendEndpoint(FullPath: str, request: Request):
        CandidatePath = (FrontendDist / FullPath).resolve()
        IsWithinDist = CandidatePath == FrontendDist or FrontendDist in CandidatePath.parents

        if FullPath and IsWithinDist and CandidatePath.is_file():
            return FileResponse(CandidatePath)

        OgTags = ""
        Segments = [Segment for Segment in FullPath.split("/") if Segment]

        if Segments:
            Scheme = GetForwardedScheme(request)
            Host = request.headers.get("x-forwarded-host", request.url.netloc)
            BaseUrl = f"{Scheme}://{Host}"
            OgTags = (
                BuildOgTags(Segments[0], "/".join(Segments[1:]), FullPath.endswith("/"), BaseUrl)
                or ""
            )

        return HTMLResponse(IndexHtmlTemplate.replace("<!--OgTags-->", OgTags))