live-hub-py/scripts/cleanup_orphan_file_assets.py

#!/usr/bin/env python
"""清理协作台账的孤儿图（`scene=collab` 且已经没有任何一行在引用）。

## 为什么需要它

换图时旧 key 只是从行里移除，**不立即删对象**（可能有人正在看、误删不可恢复，设计文档 §6.4）。
攒久了对象存储里全是没人引用的图，所以补一个**人工跑**的清理脚本：

- 只扫 `file_assets.scene = 'collab'`（协作台账专用的 scene，不碰别的业务）；
- 只删**不被任何 `huya_external_orders` 行引用**的对象（bind_images / home_images 两个槽位）；
- 只删 `created_at` 早于 N 天（默认 30）的（给"刚上传还没提交"的图留足时间）；
- **默认 `--dry-run`**，先出清单，确认后再 `--apply`。

## 用法

```bash
# 1) 先看会删什么（不碰库、不碰对象存储）
python scripts/cleanup_orphan_file_assets.py --dry-run

# 2) 真删（务必先备份库；对象删了不可恢复）
python scripts/cleanup_orphan_file_assets.py --apply
python scripts/cleanup_orphan_file_assets.py --apply --days 60
```

## 安全约束（别删掉正在用的图）

1. **删对象成功才删台账行**：对象存储报错时保留 `file_assets` 行，下次还能再清一遍；
   反过来先删行再删对象，一旦对象没删掉就真的没人知道它存在了。
2. 变体（thumb / medium）按上传时记录在台账里的 key 删，不靠猜扩展名。
3. 引用判断读的是**当前库里**的行；脚本运行期间有人在编辑 → 可能误判，所以默认留 30 天，
   并要求先 `--dry-run`。
"""

from __future__ import annotations

import argparse
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from sqlalchemy.orm import Session

from web.backend.database import SessionLocal
from web.backend.models import FileAsset, HuyaExternalOrder

# 协作台账的 scene（web/backend/services/file_service.normalize_scene 会把 scene 转小写）。
COLLAB_SCENE = "collab"
DEFAULT_RETENTION_DAYS = 30


def referenced_keys(db: Session) -> set[str]:
    """当前被协作台账行引用的 object_key（两个图片槽位）。"""
    keys: set[str] = set()
    rows = db.query(HuyaExternalOrder.bind_images, HuyaExternalOrder.home_images).all()
    for bind_images, home_images in rows:
        for value in (bind_images, home_images):
            if isinstance(value, list):
                keys.update(str(item) for item in value if item)
    return keys


def find_orphans(
    db: Session,
    *,
    days: int = DEFAULT_RETENTION_DAYS,
    scene: str = COLLAB_SCENE,
    now: datetime | None = None,
) -> list[FileAsset]:
    """找出「超过保留期 + 没人引用」的图（按最老的排前面）。"""
    cutoff = (now or datetime.now(UTC)) - timedelta(days=max(0, int(days)))
    referenced = referenced_keys(db)
    candidates = (
        db.query(FileAsset)
        .filter(
            FileAsset.scene == scene,
            FileAsset.status == "active",
            FileAsset.created_at < cutoff,
        )
        .order_by(FileAsset.created_at.asc(), FileAsset.id.asc())
        .all()
    )
    return [asset for asset in candidates if asset.object_key not in referenced]


def _keys_of(asset: FileAsset) -> list[str]:
    """原图 + 两个变体的 key（变体为空说明当时生成失败）。"""
    return [
        key
        for key in (
            asset.object_key,
            asset.thumbnail_object_key,
            asset.medium_object_key,
        )
        if key
    ]


def _format_size(size: int) -> str:
    value = float(size or 0)
    for unit in ("B", "KB", "MB", "GB"):
        if value < 1024 or unit == "GB":
            return f"{value:.1f}{unit}"
        value /= 1024
    return f"{value:.1f}GB"


def run(db: Session, *, apply: bool, days: int = DEFAULT_RETENTION_DAYS) -> dict:
    """列出/清理孤儿图，返回统计（`--dry-run` 什么都不写）。"""
    orphans = find_orphans(db, days=days)
    storage = None
    if apply and orphans:
        from web.backend.services.file_storage import get_object_storage

        storage = get_object_storage()

    deleted = 0
    failed = 0
    freed = 0
    for asset in orphans:
        created = (
            asset.created_at.strftime("%Y-%m-%d %H:%M") if asset.created_at else "-"
        )
        prefix = "[DRY-RUN] " if not apply else ""
        print(
            f"{prefix}id={asset.id} {asset.object_key} "
            f"{_format_size(asset.size_bytes)} {created} {asset.original_filename or '-'}"
        )
        if not apply:
            freed += int(asset.size_bytes or 0)
            continue
        try:
            for key in _keys_of(asset):
                storage.delete_object(key)
        except Exception as exc:  # noqa: BLE001 删对象失败就留台账行，下次再清
            failed += 1
            print(f"    !! 删除对象失败，保留台账行：{type(exc).__name__}: {exc}")
            continue
        db.delete(asset)
        deleted += 1
        freed += int(asset.size_bytes or 0)

    if apply:
        db.commit()

    return {
        "scene": COLLAB_SCENE,
        "days": days,
        "candidates": len(orphans),
        "deleted": deleted,
        "failed": failed,
        "freed_bytes": freed,
        "applied": apply,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="清理协作台账的孤儿图（scene=collab 且没人引用）"
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--dry-run", action="store_true", help="只列清单，不写库")
    group.add_argument("--apply", action="store_true", help="真删（先备份库！）")
    parser.add_argument(
        "--days",
        type=int,
        default=DEFAULT_RETENTION_DAYS,
        help=f"保留天数，超过才删（默认 {DEFAULT_RETENTION_DAYS}）",
    )
    args = parser.parse_args()

    db = SessionLocal()
    try:
        result = run(db, apply=args.apply, days=args.days)
        prefix = "[DRY-RUN] " if not args.apply else ""
        print(
            f"{prefix}scene={result['scene']} 保留 {result['days']} 天："
            f"候选 {result['candidates']} 张 / 已删 {result['deleted']} 张"
            f" / 失败 {result['failed']} 张 / 释放 {_format_size(result['freed_bytes'])}"
        )
        if not args.apply and result["candidates"]:
            print("[DRY-RUN] 确认无误后再跑 --apply（对象删了不可恢复）")
        return 1 if result["failed"] else 0
    finally:
        db.close()


if __name__ == "__main__":
    raise SystemExit(main())