mirror of
https://git.srhtnin.garden/sarah/SaraView.git
synced 2026-10-14 18:38:11 +02:00
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
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)
|