Add chunked upload for large files

Backend/Explorer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import shutil
+import tempfile
 
 from pathlib import Path
 from typing import Any
@@ -8,6 +9,9 @@ from typing import Any
 from Core import GetRoots, IsInternalPath, ResolvePath
 
 
+UploadTempRoot = Path(tempfile.gettempdir()) / "saraview_uploads"
+
+
 FileTypeMap = {
     ".md": "markdown",
     ".markdown": "markdown",
@@ -124,9 +128,7 @@ def MoveEntry(RootName: str, SourceRelativePath: str, DestinationRelativePath: s
         RenameDirectoryNote(DestinationPath, OldName)
 
 
-def SaveUploadedFile(
-    RootName: str, DirectoryRelativePath: str, FileName: str, FileBytes: bytes
-) -> str:
+def _ResolveUploadTarget(RootName: str, DirectoryRelativePath: str, FileName: str) -> Path:
     DirectoryPath = ResolvePath(RootName, DirectoryRelativePath)
 
     if not DirectoryPath.is_dir():
@@ -142,8 +144,53 @@ def SaveUploadedFile(
     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 SafeName
+    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: