audio2text/app/services/cache_cleaner.py

"""缓存清理：删除超期任务产物 + 孤儿目录，并删除对应 DB 记录。

「缓存」指每个任务落盘的产物：
    <output_dir>/task_<id>/        双语 / 英文 / 中文字幕
    <work_dir>/task_<id>.wav       中间音频（若 keep_audio=true 未被管线删掉）
    <upload_dir>/yyyy/mm/<uuid>.<ext>  保留的原始视频（若 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<id>` 但无对应 Task 的孤儿条目。

    output_dir 下是目录（task_<id>）；work_dir 下可能是 task_<id>.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_<id> 命名，跳过
        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))