"""Persistent storage for draft attachments (proforma files).""" from datetime import datetime, timezone import logging import os from pathlib import Path import re from uuid import UUID, uuid4 logger = logging.getLogger("bot.storage") def sanitize_filename(filename: str) -> str: """Sanitize filename preserving Persian/Arabic/English letters, numbers, and extensions.""" raw = (filename or "proforma.pdf").strip() name = Path(raw).name # Remove control characters and path traversal clean = re.sub(r'[\/\\:\*\?"<>\|\r\n\t]', "_", name) clean = clean.strip(". ") return clean or f"proforma_{uuid4().hex[:8]}.pdf" async def save_proforma_file( draft_id: str | UUID, file_name: str, content: bytes, mime_type: str | None = None, storage_dir: str = "data/attachments/proformas", ) -> dict: """Save proforma bytes permanently to disk. Returns dict metadata for persisting in draft payload. """ clean_name = sanitize_filename(file_name) target_dir = Path(storage_dir) / str(draft_id) target_dir.mkdir(parents=True, exist_ok=True) unique_prefix = uuid4().hex[:8] saved_filename = f"{unique_prefix}_{clean_name}" target_path = target_dir / saved_filename # Write bytes to disk target_path.write_bytes(content) logger.info("Persisted proforma (%d bytes) to %s", len(content), target_path) return { "file_name": clean_name, "saved_filename": saved_filename, "saved_path": str(target_path.resolve()), "relative_path": str(target_path.as_posix()), "size_bytes": len(content), "mime_type": mime_type or "application/octet-stream", "saved_at": datetime.now(timezone.utc).isoformat(), }