mirror of
https://git.srhtnin.garden/sarah/SaraView.git
synced 2026-10-13 18:28:14 +02:00
682 lines
20 KiB
Python
682 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
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 PlainTextResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from Auth import CreateSession, DeleteSession, GetCurrentUser, RequireAdmin, SessionCookieName
|
|
from BookReader import ReadBookForFile
|
|
from Blame import LoadBlame, RenameUserInBlame, UpdateBlameOnSave
|
|
from Core import ResolvePath
|
|
from Editor import WriteTextFile
|
|
from Explorer import (
|
|
ClassifyFile,
|
|
CreateEntry,
|
|
DeleteEntry,
|
|
EnsureDirectoryNote,
|
|
ListDirectory,
|
|
ListRoots,
|
|
MoveEntry,
|
|
)
|
|
from Groups import (
|
|
CreateGroup,
|
|
DeleteGroup,
|
|
ListGroups,
|
|
RenameUserInGroups,
|
|
SetGroupMembers,
|
|
)
|
|
from Locks import (
|
|
GetLockForPath,
|
|
IsPathAccessible,
|
|
LoadLocks,
|
|
RemoveLock,
|
|
RequirePathAccessible,
|
|
SetLock,
|
|
)
|
|
from Preview import GetFileSize, GetMimeType, ParseRangeHeader, ReadTextFile, TextFileTypes
|
|
from Settings import (
|
|
DeleteTheme,
|
|
ListThemeNames,
|
|
LoadSettings,
|
|
LoadTheme,
|
|
PaletteRoles,
|
|
RenameUserSettings,
|
|
SaveSettings,
|
|
SaveTheme,
|
|
)
|
|
from Users import (
|
|
AnyUsersExist,
|
|
ChangePassword,
|
|
CreateUser,
|
|
GetAvatarPath,
|
|
ListUsers,
|
|
RenameUser,
|
|
RenameUserAvatar,
|
|
SaveAvatar,
|
|
UpdateUserColour,
|
|
UserColourOptions,
|
|
VerifyLogin,
|
|
)
|
|
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(GetCurrentUser)):
|
|
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 {}
|
|
|
|
|
|
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(GetCurrentUser)) -> list[dict]:
|
|
return ListRoots()
|
|
|
|
|
|
@Application.get("/api/tree")
|
|
def GetTreeEndpoint(
|
|
Root: str, Path: str = "", CurrentUser: dict = Depends(GetCurrentUser)
|
|
) -> list[dict]:
|
|
RequirePathAccessible(Root, Path, CurrentUser)
|
|
|
|
try:
|
|
UserSettings = 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.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}
|
|
|
|
|
|
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)
|
|
|
|
if Body.DestinationPath:
|
|
Destination = Body.DestinationPath
|
|
elif Body.NewName:
|
|
ParentPath = Path.rsplit("/", 1)[0] if "/" in Path else ""
|
|
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(GetCurrentUser)
|
|
):
|
|
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))
|
|
|
|
FileSize = GetFileSize(FilePath)
|
|
MimeType = GetMimeType(FilePath)
|
|
RangeHeader = request.headers.get("range")
|
|
|
|
if RangeHeader:
|
|
Start, End = ParseRangeHeader(RangeHeader, FileSize)
|
|
ChunkLength = End - Start + 1
|
|
|
|
def IterateChunk():
|
|
with FilePath.open("rb") as SourceFile:
|
|
SourceFile.seek(Start)
|
|
yield SourceFile.read(ChunkLength)
|
|
|
|
Headers = {
|
|
"Content-Range": f"bytes {Start}-{End}/{FileSize}",
|
|
"Accept-Ranges": "bytes",
|
|
"Content-Length": str(ChunkLength),
|
|
}
|
|
return StreamingResponse(
|
|
IterateChunk(), status_code=206, media_type=MimeType, headers=Headers
|
|
)
|
|
|
|
def IterateFile():
|
|
with FilePath.open("rb") as SourceFile:
|
|
while Chunk := SourceFile.read(1024 * 1024):
|
|
yield Chunk
|
|
|
|
Headers = {"Accept-Ranges": "bytes", "Content-Length": str(FileSize)}
|
|
return StreamingResponse(IterateFile(), media_type=MimeType, headers=Headers)
|
|
|
|
|
|
@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(GetCurrentUser)) -> 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(GetCurrentUser)
|
|
):
|
|
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)
|
|
|
|
|
|
@Application.get("/api/book")
|
|
def GetBookEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> 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(GetCurrentUser)) -> dict:
|
|
return LoadSettings(CurrentUser["Username"])
|
|
|
|
|
|
@Application.put("/api/settings")
|
|
def PutSettingsEndpoint(NewSettings: dict, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
|
|
return SaveSettings(CurrentUser["Username"], NewSettings)
|
|
|
|
|
|
@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(GetCurrentUser)) -> 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}
|
|
|
|
|
|
FrontendDist = ApplicationDirectory.parent / "Frontend" / "dist"
|
|
if FrontendDist.is_dir():
|
|
Application.mount("/", StaticFiles(directory=str(FrontendDist), html=True), name="Frontend")
|