7z-encrypt-server/clean_orphans.py

#!/usr/bin/env python3
"""孤儿文件清理脚本: 删除数据库无记录的传输临时目录与存储残留。

清理目标:
  1. data/tmp/<transfer_id>/  中断/失败/僵尸任务的临时卷目录
  2. data/storage/ 下 files 表无记录的孤儿目录与单文件 (含旧架构残留)

安全边界 (只删确认孤儿, 绝不碰有效数据):
  - tmp: 状态 uploading/storing 且 mtime 新 (< 7 天) = 进行中, 保留;
    终态 (done/failed) 超 1 小时 = 任务已结束, tmp 卷无价值, 删;
    无任务记录超 1 小时 = 孤儿, 删; uploading/storing 超 7 天 = 僵尸任务
    (客户端早已不在), 删
  - storage: 只删 files.path 集合中不存在的目录/文件; 目录删除前检查其下
    没有任何 files.path 前缀 (防父目录误删); mtime 超 24 小时才动
    (给客户端 complete 重试窗口)
  - 删除一律 shutil.rmtree, 失败跳过并记录

用法:
  clean_orphans.py            # 实际清理
  clean_orphans.py --dry-run  # 预览不删
  clean_orphans.py --quiet    # 静默 (日志仍写 /tmp/sz-clean.log, watchdog 调用用)
"""

import argparse
import os
import shutil
import sqlite3
import sys
from datetime import datetime
from pathlib import Path

DB_PATH = Path("data/app.db")
TMP_ROOT = Path("data/tmp")
STORAGE_ROOT = Path("data/storage")
CLEAN_LOG = Path("/tmp/sz-clean.log")

# 终态任务 tmp 保留窗口 (秒): 任务结束 1 小时后卷可删
TMP_FINAL_AGE = 3600
# 进行中任务僵尸判定 (秒): 7 天无动静 = 客户端早死, 卷可删
TMP_STALE_AGE = 7 * 86400
# storage 孤儿 mtime 保留窗口 (秒): 24 小时, 给 complete 重试窗口
STORAGE_MIN_AGE = 86400
# 进行中的任务状态 (tmp 目录保留)
ACTIVE_STATUSES = {"uploading", "storing"}


def log(msg: str, quiet: bool = False) -> None:
    """写日志到 stdout 与 /tmp/sz-clean.log, 中文业务日志。"""
    line = f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {msg}"
    if not quiet:
        print(line)
    try:
        with CLEAN_LOG.open("a", encoding="utf-8") as f:
            f.write(line + "\n")
    except OSError:
        pass


def load_db(db_path: Path) -> tuple[dict[str, str], set[str]]:
    """读数据库, 返回 (transfer_id->status 映射, files.path 集合)。

    files.path 可能存相对路径 (data/storage/...) 或绝对路径, 统一 resolve
    成绝对路径, 与磁盘遍历的 resolve 结果可比。
    """
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    status_map: dict[str, str] = {}
    for r in cur.execute("SELECT transfer_id, status FROM transfers"):
        status_map[str(r["transfer_id"])] = str(r["status"])
    paths: set[str] = set()
    for r in cur.execute("SELECT path FROM files"):
        paths.add(str(Path(str(r["path"])).resolve()))
    conn.close()
    return status_map, paths


def dir_age_ok(p: Path, min_age: int) -> bool:
    """路径 mtime 距今超过 min_age 秒才返回 True (过新的一律跳过)。"""
    try:
        return (datetime.now().timestamp() - p.stat().st_mtime) >= min_age
    except OSError:
        return False


def dir_size(p: Path) -> int:
    """目录内文件总字节数 (删除失败兜底统计用)。"""
    return sum(f.stat().st_size for f in p.rglob("*") if f.is_file())


def clean_tmp(
    tmp_root: Path, status_map: dict[str, str], dry: bool, quiet: bool
) -> tuple[int, int]:
    """清理 tmp 下的孤儿任务目录, 返回 (删除数, 释放字节)。"""
    removed = 0
    freed = 0
    if not tmp_root.is_dir():
        return removed, freed
    for tid_dir in sorted(tmp_root.iterdir()):
        name = tid_dir.name
        if not tid_dir.is_dir():
            continue  # 散文件不动
        status = status_map.get(name)  # None = 无任务记录
        if status in ACTIVE_STATUSES:
            # 进行中: 超 7 天没动静 = 僵尸任务, 删; 否则保留
            if not dir_age_ok(tid_dir, TMP_STALE_AGE):
                log(f"[tmp] {name} 任务进行中, 保留", quiet)
                continue
            log(f"[tmp] {name} 僵尸任务 (进行中超 7 天), 删除", quiet)
        else:
            # 无记录或终态 (done/failed): 超 1 小时可删
            if not dir_age_ok(tid_dir, TMP_FINAL_AGE):
                log(f"[tmp] {name} 目录过新, 跳过", quiet)
                continue
            reason = "无任务记录" if status is None else f"任务已{status}"
            log(f"[tmp] {reason}, 删除孤儿任务目录 {name} ({dir_size(tid_dir) / 1024 / 1024:.1f} MB)", quiet)
        if not dry:
            size = dir_size(tid_dir)
            try:
                shutil.rmtree(tid_dir)
                removed += 1
                freed += size
            except OSError as e:
                log(f"[tmp] 删除失败 {name}: {e}", quiet)
    return removed, freed


def clean_storage(
    storage_root: Path, file_paths: set[str], dry: bool, quiet: bool
) -> tuple[int, int]:
    """清理 storage 下 files 表无记录的孤儿目录, 返回 (删除数, 释放字节)。

    安全边界 (防误删有效数据):
      - files.path 记录的是**目录路径**, 目录有效 => 内部所有文件有效,
        **文件永不单独判断删除** (曾因逐文件判断把有效目录内的 chunk 误删)
      - 目录 rel 在 file_paths 中, 或其下有任意 files.path 前缀 => 有效,
        整个子树跳过 (剪枝不深入)
      - 目录无记录且无子记录 => 孤儿候选, 整个目录删除 (含旧架构单文件
        所在的日期目录, 一并覆盖)
    """
    removed = 0
    freed = 0
    if not storage_root.is_dir():
        return removed, freed
    # 收集候选 (仅目录), 统一删除, 避免遍历中 rmtree
    candidates: list[Path] = []
    for dirpath, dirnames, filenames in os.walk(storage_root):
        d = Path(dirpath)
        if d == storage_root:
            continue
        rel = str(d.resolve())
        if rel in file_paths:
            # 有效叶子目录: files 记录精确指向它, 整个子树跳过
            dirnames[:] = []
            continue
        if not any(fp.startswith(rel + "/") for fp in file_paths):
            # 无记录且无子记录 => 孤儿候选, 整个目录删除
            if dir_age_ok(d, STORAGE_MIN_AGE):
                candidates.append(d)
                dirnames[:] = []  # 目录将删, 剪枝不再深入
            continue
        # 有子记录 (父目录层如 user/日期): 不是孤儿, 深入遍历子目录
    for p in candidates:
        size = dir_size(p)
        log(f"[storage] 删除孤儿目录 {p} ({size / 1024 / 1024:.1f} MB)", quiet)
        if not dry:
            try:
                shutil.rmtree(p)
                removed += 1
                freed += size
            except OSError as e:
                log(f"[storage] 删除失败 {p}: {e}", quiet)
    return removed, freed


def main() -> int:
    parser = argparse.ArgumentParser(description="7z-encrypt 服务端孤儿文件清理")
    parser.add_argument("--dry-run", action="store_true", help="预览模式, 不实际删除")
    parser.add_argument("--quiet", action="store_true", help="静默模式, 只写日志文件")
    parser.add_argument("--db", default=str(DB_PATH), help="SQLite 数据库路径")
    parser.add_argument("--tmp", default=str(TMP_ROOT), help="临时目录根")
    parser.add_argument("--storage", default=str(STORAGE_ROOT), help="存储目录根")
    args = parser.parse_args()

    # 默认路径 (data/app.db 等) 相对脚本所在目录解析, 调用方 cwd 无关
    script_dir = Path(__file__).resolve().parent
    db_path = Path(args.db)
    if not db_path.is_absolute():
        db_path = script_dir / db_path
    tmp_root = Path(args.tmp)
    if not tmp_root.is_absolute():
        tmp_root = script_dir / tmp_root
    storage_root = Path(args.storage)
    if not storage_root.is_absolute():
        storage_root = script_dir / storage_root
    dry = args.dry_run
    quiet = args.quiet

    mode = "预览" if dry else "清理"
    log(f"=== {mode}开始 (db={db_path}) ===", quiet)
    if not db_path.is_file():
        log(f"[错误] 数据库不存在: {db_path}", quiet)
        return 1

    status_map, file_paths = load_db(db_path)
    log(f"[db] 任务 {len(status_map)} 个, 文件记录 {len(file_paths)} 条", quiet)

    t_removed, t_freed = clean_tmp(tmp_root, status_map, dry, quiet)
    s_removed, s_freed = clean_storage(storage_root, file_paths, dry, quiet)

    total_removed = t_removed + s_removed
    total_freed = t_freed + s_freed
    log(
        f"=== {mode}完成: 删除 {total_removed} 项, 释放 {total_freed / 1024 / 1024:.1f} MB ===",
        quiet,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())