""" =============================================================================== FILE: modules/web_api/routers/files.py ROLE: Раздача сформированных отчетов и выгрузок с сохранением оригинальных имен. =============================================================================== """ import os import time import shutil import urllib.parse from typing import Optional from fastapi import APIRouter, HTTPException from fastapi.responses import FileResponse router = APIRouter(prefix="/api/v1/files", tags=["Files"]) BASE_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) WEB_OUTPUT_DIR = os.path.join(BASE_ROOT, "output", "web") REPORTS_DIR = os.path.join(BASE_ROOT, "output", "reports") TEMP_REPORTS_DIR = "/tmp/scud_reports" os.makedirs(WEB_OUTPUT_DIR, exist_ok=True) os.makedirs(REPORTS_DIR, exist_ok=True) os.makedirs(TEMP_REPORTS_DIR, exist_ok=True) SESSION_TTL_HOURS = 24 def purge_old_tool_sessions(tool_dir_path: str): if not os.path.exists(tool_dir_path): return now = time.time() cutoff = now - (SESSION_TTL_HOURS * 3600) try: for entry in os.listdir(tool_dir_path): subpath = os.path.join(tool_dir_path, entry) if os.path.isdir(subpath): if os.path.getmtime(subpath) < cutoff: shutil.rmtree(subpath, ignore_errors=True) except Exception: pass @router.get("/download/{tool_name}/{session_uuid}/{filename}") async def download_file(tool_name: str, session_uuid: str, filename: str): safe_tool = os.path.basename(tool_name) safe_uuid = os.path.basename(session_uuid) safe_filename = os.path.basename(filename) file_path = os.path.join(WEB_OUTPUT_DIR, safe_tool, safe_uuid, safe_filename) if not os.path.exists(file_path) or not os.path.isfile(file_path): raise HTTPException(status_code=404, detail="Файл не найден или срок его действия истек") media_type = "application/octet-stream" if safe_filename.endswith((".md", ".txt")): media_type = "text/markdown; charset=utf-8" elif safe_filename.endswith(".xlsx"): media_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" elif safe_filename.endswith(".pdf"): media_type = "application/pdf" encoded_filename = urllib.parse.quote(safe_filename) return FileResponse( path=file_path, media_type=media_type, headers={ "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}" } ) def _find_report_recursively(target_filename: str) -> Optional[str]: """Рекурсивный поиск файла отчета по имени в /tmp и во всех подпапках output/reports/.""" # 1. Проверяем /tmp/scud_reports tmp_path = os.path.join(TEMP_REPORTS_DIR, target_filename) if os.path.exists(tmp_path) and os.path.isfile(tmp_path): return tmp_path # 2. Проверяем прямой путь в output/reports direct_path = os.path.join(REPORTS_DIR, target_filename) if os.path.exists(direct_path) and os.path.isfile(direct_path): return direct_path # 3. Рекурсивный поиск по подкаталогам (год/месяц) for root, _, files in os.walk(REPORTS_DIR): if target_filename in files: return os.path.join(root, target_filename) return None @router.get("/download/reports/{filename:path}") async def download_report_direct(filename: str): """ Прямое скачивание отчетов: - Декодирует UTF-8 URL (%20 -> пробел). - Ищет файл в output/reports/{YEAR}/{MONTH} и во временном буфере /tmp/scud_reports. """ decoded_name = urllib.parse.unquote(filename).strip() safe_filename = os.path.basename(decoded_name) target_path = _find_report_recursively(safe_filename) if not target_path or not os.path.exists(target_path): raise HTTPException(status_code=404, detail="Отчет не найден") encoded_filename = urllib.parse.quote(safe_filename) return FileResponse( path=target_path, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"} )