"""File storage: uploads and tool artifacts on local disk (S3 later), TTL, signed URLs.""" from __future__ import annotations import base64 import hashlib import hmac import re import time from datetime import timedelta from pathlib import Path from sqlalchemy import delete, select from app.core.config import get_settings from app.db import SessionLocal from app.models import StoredFile, new_id, utcnow ALLOWED_UPLOAD = { "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "xls": "application/vnd.ms-excel", "csv": "text/csv", "pdf": "application/pdf", "png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "webp": "image/webp", "txt": "text/plain", "md": "text/markdown", "json": "application/json", "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", } MIME_BY_TYPE = { "xlsx": ALLOWED_UPLOAD["xlsx"], "image": "image/png", "csv": "text/csv", "pdf": "application/pdf", "file": "application/octet-stream", "docx": ALLOWED_UPLOAD["docx"], } def safe_name(name: str) -> str: name = re.sub(r"[^\w.\-() àâäçéèêëîïôöùûüÿœæÀÂÄÇÉÈÊËÎÏÔÖÙÛÜŸŒÆ]", "_", name).strip() return name[:120] or "fichier" def _root() -> Path: return get_settings().DATA_DIR / "files" def _path(storage_key: str) -> Path: return _root() / storage_key def ext_of(filename: str) -> str: return filename.rsplit(".", 1)[-1].lower() if "." in filename else "" async def store_upload(user_id: str, conversation_id: str | None, filename: str, data: bytes) -> StoredFile: s = get_settings() filename = safe_name(filename) ext = ext_of(filename) if ext not in ALLOWED_UPLOAD: raise ValueError("Type de fichier non autorisé.") if len(data) > s.UPLOAD_MAX_MB * 1024 * 1024: raise ValueError(f"Fichier trop volumineux (max {s.UPLOAD_MAX_MB} Mo).") fid = new_id() key = f"{user_id[:8]}/{fid}.{ext}" p = _path(key) p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(data) ftype = "image" if ext in {"png", "jpg", "jpeg", "webp"} else ext rec = StoredFile(id=fid, user_id=user_id, conversation_id=conversation_id, kind="upload", type=ftype, filename=filename, mime=ALLOWED_UPLOAD[ext], storage_key=key, size_bytes=len(data), expires_at=utcnow() + timedelta(days=s.FILE_TTL_DAYS)) async with SessionLocal() as session: session.add(rec) await session.commit() return rec async def store_artifact(user_id: str, conversation_id: str | None, filename: str, data: bytes, ftype: str = "file", tool_call_id: str | None = None) -> StoredFile: s = get_settings() filename = safe_name(filename) ext = ext_of(filename) or "bin" fid = new_id() key = f"{user_id[:8]}/{fid}.{ext}" p = _path(key) p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(data) mime = ALLOWED_UPLOAD.get(ext, MIME_BY_TYPE.get(ftype, "application/octet-stream")) rec = StoredFile(id=fid, user_id=user_id, conversation_id=conversation_id, kind="artifact", tool_call_id=tool_call_id, type=ftype, filename=filename, mime=mime, storage_key=key, size_bytes=len(data), expires_at=utcnow() + timedelta(days=s.FILE_TTL_DAYS)) async with SessionLocal() as session: session.add(rec) await session.commit() return rec async def get_file(file_id: str) -> StoredFile | None: async with SessionLocal() as session: return await session.get(StoredFile, file_id) def read_bytes(rec: StoredFile) -> bytes: return _path(rec.storage_key).read_bytes() async def list_for_conversation(conversation_id: str) -> list[StoredFile]: async with SessionLocal() as session: rows = await session.execute( select(StoredFile).where(StoredFile.conversation_id == conversation_id) .order_by(StoredFile.created_at)) return list(rows.scalars()) async def sandbox_inputs(user_id: str, file_ids: list[str]) -> list[dict[str, str]]: out: list[dict[str, str]] = [] for fid in file_ids[:10]: rec = await get_file(fid) if rec and rec.user_id == user_id and _path(rec.storage_key).exists(): out.append({"name": rec.filename, "content_b64": base64.b64encode(read_bytes(rec)).decode()}) return out async def set_pinned(file_id: str, user_id: str, pinned: bool) -> bool: s = get_settings() async with SessionLocal() as session: rec = await session.get(StoredFile, file_id) if not rec or rec.user_id != user_id: return False rec.pinned = pinned days = s.FILE_PINNED_TTL_DAYS if pinned else s.FILE_TTL_DAYS rec.expires_at = rec.created_at + timedelta(days=days) await session.commit() return True async def purge_expired() -> int: n = 0 async with SessionLocal() as session: rows = await session.execute(select(StoredFile).where(StoredFile.expires_at < utcnow())) for rec in rows.scalars(): _path(rec.storage_key).unlink(missing_ok=True) n += 1 await session.execute(delete(StoredFile).where(StoredFile.expires_at < utcnow())) await session.commit() return n async def purge_user(user_id: str) -> int: n = 0 async with SessionLocal() as session: rows = await session.execute(select(StoredFile).where(StoredFile.user_id == user_id)) for rec in rows.scalars(): _path(rec.storage_key).unlink(missing_ok=True) n += 1 await session.execute(delete(StoredFile).where(StoredFile.user_id == user_id)) await session.commit() return n # --- signed download links (10 min) -------------------------------------------------------- def sign(file_id: str, ttl_s: int = 600) -> str: s = get_settings() exp = int(time.time()) + ttl_s mac = hmac.new(s.JWT_SECRET.get_secret_value().encode(), f"{file_id}:{exp}".encode(), hashlib.sha256).hexdigest()[:32] return f"{exp}.{mac}" def verify_signature(file_id: str, sig: str) -> bool: try: exp_s, mac = sig.split(".", 1) exp = int(exp_s) except ValueError: return False if exp < time.time(): return False s = get_settings() expected = hmac.new(s.JWT_SECRET.get_secret_value().encode(), f"{file_id}:{exp}".encode(), hashlib.sha256).hexdigest()[:32] return hmac.compare_digest(expected, mac)