103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import datetime
|
|
import database as db
|
|
|
|
# TODO: Ugly ass vibe coded shit. Someone please rewrite.
|
|
|
|
# TODO: This belonds in app
|
|
DRIVES_ROOT = None
|
|
|
|
async def init_drives(data_directory):
|
|
"""Initialize DRIVES_ROOT global from app startup."""
|
|
global DRIVES_ROOT
|
|
DRIVES_ROOT = os.path.join(data_directory, "drives")
|
|
os.makedirs(DRIVES_ROOT, exist_ok=True)
|
|
|
|
def _generate_file_id():
|
|
"""Generate a random hex string for filesystem-level filenames."""
|
|
return secrets.token_hex(8)
|
|
|
|
async def upload_file(user_id, drive_name, filename, stream, mime_type=None):
|
|
"""Write an incoming stream to disk and register it in the DB. Returns file ID."""
|
|
# 1. Locate or create Drive
|
|
drive = await db.Drive.get_or_none(user_id=user_id, name=drive_name)
|
|
if not drive:
|
|
drive = await db.Drive.create(user_id=user_id, name=drive_name)
|
|
|
|
# Check collision
|
|
if await db.DriveFile.get_or_none(drive_id=drive.id, filename=filename):
|
|
raise Exception("Duplicated file")
|
|
|
|
# 2. Prepare disk path
|
|
file_id = _generate_file_id()
|
|
drive_dir = os.path.join(DRIVES_ROOT, str(drive.id))
|
|
os.makedirs(drive_dir, exist_ok=True)
|
|
real_path = os.path.join(drive_dir, file_id)
|
|
|
|
# 3. Write stream and calculate size
|
|
size = 0
|
|
with open(real_path, "wb") as f:
|
|
while chunk := stream.read(8192):
|
|
f.write(chunk)
|
|
size += len(chunk)
|
|
|
|
# 4. Register in database
|
|
# TODO: catch id conflict and retry
|
|
record = await db.DriveFile.create(
|
|
id=file_id,
|
|
drive_id=drive.id,
|
|
filename=filename,
|
|
mime_type=mime_type or "application/octet-stream",
|
|
size_bytes=size,
|
|
)
|
|
return record.id
|
|
|
|
async def get_metadata(file_id):
|
|
"""Return JSON metadata for frontend display."""
|
|
file = await db.DriveFile.get_or_none(id=file_id)
|
|
if not file:
|
|
raise Exception("File not found")
|
|
return {
|
|
"id": file.id,
|
|
"name": file.filename,
|
|
"size": file.size_bytes,
|
|
"mime": file.mime_type,
|
|
"created": file.created_at.isoformat(),
|
|
"accessed": file.accessed_at.isoformat() if file.accessed_at else None
|
|
}
|
|
|
|
async def delete_file(file_id):
|
|
file = await db.DriveFile.get_or_none(id=file_id)
|
|
if not file:
|
|
raise Exception("File not found")
|
|
|
|
await file.delete()
|
|
|
|
abs_path = os.path.join(DRIVES_ROOT, str(file.drive_id), file.id)
|
|
if os.path.exists(abs_path):
|
|
os.remove(abs_path)
|
|
|
|
return abs_path
|
|
|
|
async def resolve_file(user_id, drive_name, filename):
|
|
"""Resolve file to absolute disk path after permission check. Returns (abs_path, file.id)."""
|
|
# Eager load drive and verify ownership
|
|
drive = await db.Drive.get_or_none(user_id=user_id, name=drive_name)
|
|
if drive.user_id != user_id:
|
|
raise PermissionError("Access denied")
|
|
|
|
file = await db.DriveFile.get_or_none(filename=filename, drive_id=drive.id)
|
|
if not file:
|
|
raise FileNotFoundError("File not found")
|
|
|
|
# can do this asynchronously?
|
|
file.accessed_at = datetime.datetime.now(datetime.UTC)
|
|
await file.save()
|
|
|
|
abs_path = os.path.join(DRIVES_ROOT, str(drive.id), file.id)
|
|
if not os.path.exists(abs_path):
|
|
raise FileNotFoundError("File missing on disk")
|
|
|
|
return abs_path, file.id |