"""缓存清理:删除超期任务产物 + 孤儿目录,并删除对应 DB 记录。 「缓存」指每个任务落盘的产物: /task_/ 双语 / 英文 / 中文字幕 /task_.wav 中间音频(若 keep_audio=true 未被管线删掉) /yyyy/mm/. 保留的原始视频(若 delete_original_after_extract=false) 超期 = `created_at` 早于 `now - cache_retention_days`。超期任务的全部产物连同 Task / UploadSession 行一起删除,避免历史页出现指向已删文件的死链接。 另做一次孤儿扫描:output_dir / work_dir 下存在但无对应 Task 的目录(进程崩溃残留), 按目录 mtime 判超期后删除,让磁盘不被异常退出留下的碎片占满。 """ from __future__ import annotations import logging import shutil import time from datetime import datetime, timedelta, timezone from pathlib import Path from sqlalchemy.orm import Session from ..config import get_settings from ..database import get_session_local from ..models.task import Task from ..models.upload_session import UploadSession logger = logging.getLogger("audio2text.cache") def _now_utc() -> datetime: return datetime.now(timezone.utc) def _aware(dt: datetime | None) -> datetime | None: """SQLite 存 naive datetime,统一补 UTC。""" if dt is None: return None return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc) def _rm_tree(path: Path) -> bool: """删目录或文件,失败仅警告不抛。返回是否实际删除。""" try: if path.is_dir(): shutil.rmtree(path, ignore_errors=False) return True if path.is_file(): path.unlink(missing_ok=True) return True except Exception as exc: # pragma: no cover logger.warning("清理 %s 失败:%s", path, exc) return False def purge_expired_cache(db: Session | None = None) -> dict: """执行一次清理,返回统计 dict。 可传入已有 Session(如复用请求会话),不传则自建一个并关闭。 """ own_db = db is None if own_db: db = get_session_local()() try: return _purge(db) finally: if own_db: db.close() def _purge(db: Session) -> dict: s = get_settings() retention = max(0, s.storage.cache_retention_days) cutoff = _now_utc() - timedelta(days=retention) upload_root = s.upload_dir() work_root = s.work_dir() output_root = s.output_dir() stats = {"tasks": 0, "outputs": 0, "audio": 0, "videos": 0, "orphans": 0, "retention_days": retention} if retention <= 0: logger.info("缓存清理已禁用(cache_retention_days=%s)。", retention) return stats # ---------- 1. 超期任务 ---------- tasks = db.query(Task).all() for task in tasks: created = _aware(task.created_at) if created is None or created >= cutoff: continue # 输出目录 out_dir = output_root / f"task_{task.id}" if out_dir.is_dir() and _rm_tree(out_dir): stats["outputs"] += 1 # 中间音频 wav = work_root / f"task_{task.id}.wav" if wav.is_file() and _rm_tree(wav): stats["audio"] += 1 # 保留的原始视频(若未在提取后删除) if task.source_path: src = upload_root / task.source_path if src.is_file() and _rm_tree(src): stats["videos"] += 1 # 删 DB 记录(先删关联的 UploadSession,再删 Task) db.query(UploadSession).filter(UploadSession.task_id == task.id).delete() db.delete(task) stats["tasks"] += 1 logger.info( "清理超期任务 id=%s file=%s created=%s", task.id, task.filename, created.isoformat(), ) if stats["tasks"]: db.commit() # ---------- 2. 孤儿目录扫描 ---------- existing_ids = {row[0] for row in db.query(Task.id).all()} stats["orphans"] += _purge_orphans(output_root, "task_", cutoff, existing_ids) stats["orphans"] += _purge_orphans(work_root, "task_", cutoff, existing_ids, suffix=".wav") logger.info( "缓存清理完成:任务 %d(输出 %d / 音频 %d / 视频 %d)+ 孤儿 %d,保留期 %d 天", stats["tasks"], stats["outputs"], stats["audio"], stats["videos"], stats["orphans"], retention, ) return stats def _purge_orphans(root: Path, prefix: str, cutoff: datetime, existing_ids: set[int], suffix: str | None = None) -> int: """删 root 下名为 `prefix` 但无对应 Task 的孤儿条目。 output_dir 下是目录(task_);work_dir 下可能是 task_.wav 文件。 用 mtime 判超期,避免误删刚产生的中间产物。 """ if not root.is_dir(): return 0 n = 0 for entry in root.iterdir(): name = entry.name if not name.startswith(prefix): continue rest = name[len(prefix):] if suffix: if not rest.endswith(suffix): continue rest = rest[: -len(suffix)] try: tid = int(rest) except ValueError: continue # 不是 task_ 命名,跳过 if tid in existing_ids: continue # 孤儿:按 mtime 判超期 try: mtime = datetime.fromtimestamp(entry.stat().st_mtime, tz=timezone.utc) except OSError: continue if mtime >= cutoff: continue if _rm_tree(entry): n += 1 logger.info("清理孤儿缓存 %s(mtime=%s)", entry, mtime.isoformat()) return n def run_forever(interval_hours: int) -> None: """后台线程入口:先跑一次,再按间隔循环。供 main.py lifespan 拉起。""" interval = max(1, interval_hours) * 3600 logger.info("缓存清理调度启动:间隔 %d 小时,保留期 %d 天。", interval_hours, get_settings().storage.cache_retention_days) # 启动时先跑一次 try: purge_expired_cache() except Exception as exc: # pragma: no cover logger.warning("启动缓存清理失败:%s", exc) while True: time.sleep(interval) try: purge_expired_cache() except Exception as exc: # pragma: no cover logger.warning("定时缓存清理失败:%s", exc) if __name__ == "__main__": # pragma: no cover # 容器内手动触发:python -m app.services.cache_cleaner import json logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") result = purge_expired_cache() print(json.dumps(result, ensure_ascii=False))