from __future__ import annotations from pathlib import Path from fastapi.responses import StreamingResponse TextFileTypes = {"markdown", "text", "config", "html"} MimeTypeMap = { ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp", ".svg": "image/svg+xml", ".mp4": "video/mp4", ".webm": "video/webm", ".mov": "video/quicktime", ".mkv": "video/x-matroska", } def ReadTextFile(FilePath: Path) -> str: return FilePath.read_text(encoding="utf-8") def GetFileSize(FilePath: Path) -> int: return FilePath.stat().st_size def GetMimeType(FilePath: Path) -> str: return MimeTypeMap.get(FilePath.suffix.lower(), "application/octet-stream") def ParseRangeHeader(RangeHeader: str, FileSize: int) -> tuple[int, int]: RangeValue = RangeHeader.replace("bytes=", "").strip() StartText, _, EndText = RangeValue.partition("-") Start = int(StartText) if StartText else 0 End = int(EndText) if EndText else FileSize - 1 End = min(End, FileSize - 1) return Start, End def StreamBinaryFile(FilePath: Path, RangeHeader: str | None) -> StreamingResponse: FileSize = GetFileSize(FilePath) MimeType = GetMimeType(FilePath) if RangeHeader: Start, End = ParseRangeHeader(RangeHeader, FileSize) ChunkLength = End - Start + 1 def IterateChunk(): with FilePath.open("rb") as SourceFile: SourceFile.seek(Start) Remaining = ChunkLength while Remaining > 0: Chunk = SourceFile.read(min(1024 * 1024, Remaining)) if not Chunk: break Remaining -= len(Chunk) yield Chunk 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)