SaraView/Backend/Explorer.py

from __future__ import annotations

import shutil
import tempfile

from pathlib import Path
from typing import Any

from Core import GetRoots, IsInternalPath, ResolvePath


UploadTempRoot = Path(tempfile.gettempdir()) / "saraview_uploads"


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


def ClassifyFile(EntryPath: Path) -> str:
    return FileTypeMap.get(EntryPath.suffix.lower(), "unknown")


NoteFilePrefix = ".sv_"
DefaultNoteExtension = ".md"


def IsNoteFile(EntryName: str) -> bool:
    return EntryName.startswith(NoteFilePrefix)


def FindDirectoryNote(DirectoryPath: Path) -> Path | None:
    Prefix = f"{NoteFilePrefix}{DirectoryPath.name}."

    for Candidate in DirectoryPath.iterdir():
        if Candidate.is_file() and Candidate.name.startswith(Prefix):
            return Candidate

    return None


def EnsureDirectoryNote(DirectoryPath: Path) -> Path:
    Existing = FindDirectoryNote(DirectoryPath)
    if Existing is not None:
        return Existing

    NotePath = DirectoryPath / f"{NoteFilePrefix}{DirectoryPath.name}{DefaultNoteExtension}"
    NotePath.touch()
    return NotePath


def GetDirectoryNoteInfo(DirectoryPath: Path) -> dict[str, Any] | None:
    NotePath = FindDirectoryNote(DirectoryPath)

    if NotePath is None:
        return None

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


def RenameDirectoryNote(DirectoryPath: Path, OldDirectoryName: str) -> None:
    OldPrefix = f"{NoteFilePrefix}{OldDirectoryName}."

    for Candidate in DirectoryPath.iterdir():
        if Candidate.is_file() and Candidate.name.startswith(OldPrefix):
            NewName = f"{NoteFilePrefix}{DirectoryPath.name}{Candidate.suffix}"
            Candidate.rename(DirectoryPath / NewName)
            return


def ListRoots() -> list[dict[str, Any]]:
    return [{"Name": Name} for Name in GetRoots()]


def CreateEntry(RootName: str, RelativePath: str, IsDirectory: bool) -> None:
    EntryPath = ResolvePath(RootName, RelativePath)

    if EntryPath.exists():
        raise FileExistsError(f"{EntryPath.name} already exists.")

    if IsDirectory:
        EntryPath.mkdir(parents=False)
        EnsureDirectoryNote(EntryPath)
    else:
        EntryPath.parent.mkdir(parents=True, exist_ok=True)
        EntryPath.touch()


def MoveEntry(RootName: str, SourceRelativePath: str, DestinationRelativePath: str) -> None:
    SourcePath = ResolvePath(RootName, SourceRelativePath)
    DestinationPath = ResolvePath(RootName, DestinationRelativePath)

    if not SourcePath.exists():
        raise FileNotFoundError(f"{SourcePath.name} does not exist.")

    if DestinationPath.exists():
        raise FileExistsError(f"{DestinationPath.name} already exists.")

    OldName = SourcePath.name
    DestinationPath.parent.mkdir(parents=True, exist_ok=True)

    try:
        SourcePath.rename(DestinationPath)
    except OSError:
        shutil.move(str(SourcePath), str(DestinationPath))

    if DestinationPath.is_dir():
        RenameDirectoryNote(DestinationPath, OldName)


def _ResolveUploadTarget(RootName: str, DirectoryRelativePath: str, FileName: str) -> Path:
    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.")

    return TargetPath


def SaveUploadedFile(
    RootName: str, DirectoryRelativePath: str, FileName: str, FileBytes: bytes
) -> str:
    TargetPath = _ResolveUploadTarget(RootName, DirectoryRelativePath, FileName)
    TargetPath.write_bytes(FileBytes)
    return TargetPath.name


def _GetUploadChunkDir(UploadId: str) -> Path:
    SafeId = "".join(Char for Char in UploadId if Char.isalnum() or Char in ("-", "_"))

    if not SafeId:
        raise ValueError("Invalid upload id.")

    return UploadTempRoot / SafeId


def SaveUploadChunk(UploadId: str, ChunkIndex: int, ChunkBytes: bytes) -> None:
    ChunkDir = _GetUploadChunkDir(UploadId)
    ChunkDir.mkdir(parents=True, exist_ok=True)
    (ChunkDir / f"{ChunkIndex:08d}.part").write_bytes(ChunkBytes)


def FinishChunkedUpload(
    RootName: str,
    DirectoryRelativePath: str,
    UploadId: str,
    TotalChunks: int,
    FileName: str,
) -> str:
    TargetPath = _ResolveUploadTarget(RootName, DirectoryRelativePath, FileName)
    ChunkDir = _GetUploadChunkDir(UploadId)
    ChunkPaths = [ChunkDir / f"{Index:08d}.part" for Index in range(TotalChunks)]

    if not all(ChunkPath.exists() for ChunkPath in ChunkPaths):
        raise ValueError("Missing upload chunks.")

    with TargetPath.open("wb") as OutputFile:
        for ChunkPath in ChunkPaths:
            with ChunkPath.open("rb") as ChunkFile:
                shutil.copyfileobj(ChunkFile, OutputFile)

    shutil.rmtree(ChunkDir)
    return TargetPath.name


def DeleteEntry(RootName: str, RelativePath: str) -> None:
    EntryPath = ResolvePath(RootName, RelativePath)

    if not EntryPath.exists():
        raise FileNotFoundError(f"{EntryPath.name} does not exist.")

    if EntryPath.is_dir():
        shutil.rmtree(EntryPath)
    else:
        EntryPath.unlink()


def ListDirectory(
    RootName: str, RelativePath: str, ShowHidden: bool = False, ShowSvFiles: bool = False
) -> list[dict[str, Any]]:
    DirectoryPath = ResolvePath(RootName, RelativePath)

    if not DirectoryPath.is_dir():
        raise NotADirectoryError(f"{DirectoryPath} is not a directory.")

    Entries: list[dict[str, Any]] = []

    SortedChildren = sorted(
        DirectoryPath.iterdir(),
        key=lambda EntryPath: (EntryPath.is_file(), EntryPath.name.casefold()),
    )

    for EntryPath in SortedChildren:
        if IsInternalPath(EntryPath):
            continue

        if IsNoteFile(EntryPath.name):
            if not ShowSvFiles:
                continue
        elif not ShowHidden and EntryPath.name.startswith("."):
            continue

        IsDirectory = EntryPath.is_dir()
        Entries.append(
            {
                "Name": EntryPath.name,
                "IsDirectory": IsDirectory,
                "FileType": None if IsDirectory else ClassifyFile(EntryPath),
                "Size": None if IsDirectory else EntryPath.stat().st_size,
                "Note": GetDirectoryNoteInfo(EntryPath) if IsDirectory else None,
            }
        )

    return Entries