"""DB 归档 + 清理脚本。 职责: 1. scan_tasks 90 天前: 清 detail JSONB (节省空间, 扫描详情留 90 天) 2. scan_tasks 365 天前: 移到 scan_tasks_archive (含 ScanFinding) 3. idempotency_records: 清 7 天前的 (幂等缓存过期重扫) 4. policy_versions: 只保留最新 5 个版本 (历史版本清掉) 用法: docker compose exec api python3 scripts/archive_old_tasks.py # dry-run, 只报告不动数据 docker compose exec api python3 scripts/archive_old_tasks.py --apply # 真正执行 建议: 每天 cron 跑一次 设计原则: - 小批量提交 (1000 条一批), 避免长事务锁表 - 先 dry-run 再 apply, 输出清晰 - pg_repack / VACUUM ANALYZE 在脚本最后触发 (让 autovacuum 兜底也行) """ from __future__ import annotations import argparse import logging import sys import os from datetime import timedelta, datetime, timezone from pathlib import Path sys.path.insert(0, "/app") os.environ.setdefault("MESHSAFE_MASTER_KEY", "Dt0ktRs8SjEH8VxWZJDX125NV1qiII88HoiXZPhWers=") from app.db import SessionLocal # noqa: E402 from app.models import ( # noqa: E402 ScanTask, ScanFinding, IdempotencyRecord, PolicyVersion, ) from sqlalchemy import select, func # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("archive") # ---- 配置 ---- DETAIL_RETENTION_DAYS = 90 # 90 天前清 detail JSONB ARCHIVE_RETENTION_DAYS = 365 # 365 天前移归档 IDEM_RETENTION_DAYS = 7 # 幂等缓存 7 天 TTL POLICY_KEEP_VERSIONS = 5 # policy_versions 只留最新 5 个 BATCH_SIZE = 1000 # 分批处理 def _cutoff(days: int) -> datetime: return datetime.now(timezone.utc) - timedelta(days=days) def archive_tasks(dry_run: bool) -> None: """归档 365 天前的 scan_tasks 到 scan_tasks_archive。""" cutoff = _cutoff(ARCHIVE_RETENTION_DAYS) db = SessionLocal() try: total = db.execute( select(func.count(ScanTask.id)).where(ScanTask.created_at < cutoff) ).scalar() or 0 log.info(f"[归档] {ARCHIVE_RETENTION_DAYS} 天前的 scan_tasks: {total} 条") if total == 0: log.info(" → 无需归档") return if dry_run: log.info(f" [dry-run] 将归档 {total} 条") return # 分批 moved = 0 while True: batch = db.execute( select(ScanTask).where(ScanTask.created_at < cutoff).limit(BATCH_SIZE) ).scalars().all() if not batch: break for task in batch: try: db.execute( __import__("sqlalchemy").text(""" INSERT INTO scan_tasks_archive (id, channel, client_id, filename, declared_ext, size, sha256, md5, real_type, status, mode, verdict, action, score, reason, policy_version, has_detail, has_findings, created_at, finished_at, duration_ms, archived_at) VALUES (:id, :channel, :client_id, :filename, :declared_ext, :size, :sha256, :md5, :real_type, :status, :mode, :verdict, :action, :score, :reason, :policy_version, :has_detail, :has_findings, :created_at, :finished_at, :duration_ms, NOW()) ON CONFLICT (id) DO NOTHING """), { "id": task.id, "channel": task.channel, "client_id": task.client_id, "filename": task.filename, "declared_ext": task.declared_ext, "size": task.size, "sha256": task.sha256, "md5": task.md5, "real_type": task.real_type, "status": task.status, "mode": task.mode, "verdict": task.verdict, "action": task.action, "score": task.score, "reason": task.reason, "policy_version": task.policy_version, "has_detail": bool(task.detail and len(task.detail) > 0), "has_findings": bool(task.findings), "created_at": task.created_at, "finished_at": task.finished_at, "duration_ms": task.duration_ms, }, ) # CASCADE 删 ScanFinding db.delete(task) moved += 1 except Exception as e: log.warning(f" task {task.id} 归档失败: {e}") db.rollback() continue db.commit() log.info(f" 已归档 {moved}/{total}") log.info(f" ✅ 归档完成: {moved} 条") finally: db.close() def scrub_detail(dry_run: bool) -> None: """90~365 天之间的任务清 detail JSONB (留摘要, 省空间)。""" db_cutoff = _cutoff(DETAIL_RETENTION_DAYS) arc_cutoff = _cutoff(ARCHIVE_RETENTION_DAYS) db = SessionLocal() try: total = db.execute( select(func.count(ScanTask.id)).where( ScanTask.created_at < db_cutoff, ScanTask.created_at >= arc_cutoff, ScanTask.detail.isnot(None), ) ).scalar() or 0 log.info(f"[清 detail] {DETAIL_RETENTION_DAYS}~{ARCHIVE_RETENTION_DAYS} 天之间有 detail 的: {total} 条") if total == 0: log.info(" → 无需清理") return if dry_run: log.info(f" [dry-run] 将清 {total} 条 detail") return # 分批 UPDATE updated = 0 while True: batch_ids = db.execute( select(ScanTask.id).where( ScanTask.created_at < db_cutoff, ScanTask.created_at >= arc_cutoff, ScanTask.detail.isnot(None), ).limit(BATCH_SIZE) ).scalars().all() if not batch_ids: break db.execute( __import__("sqlalchemy").update(ScanTask).where(ScanTask.id.in_(batch_ids)).values(detail=None) ) db.commit() updated += len(batch_ids) log.info(f" 已清 {updated}/{total}") log.info(f" ✅ detail 清理完成: {updated} 条") finally: db.close() def clean_idempotency(dry_run: bool) -> None: """清 7 天前的幂等缓存 (过期的可以重扫)。""" cutoff = _cutoff(IDEM_RETENTION_DAYS) db = SessionLocal() try: total = db.execute( select(func.count(IdempotencyRecord.id)).where(IdempotencyRecord.created_at < cutoff) ).scalar() or 0 log.info(f"[幂等清理] {IDEM_RETENTION_DAYS} 天前的 idempotency_records: {total} 条") if total == 0: log.info(" → 无需清理") return if dry_run: log.info(f" [dry-run] 将清 {total} 条") return db.execute( __import__("sqlalchemy").delete(IdempotencyRecord).where(IdempotencyRecord.created_at < cutoff) ) db.commit() log.info(f" ✅ 幂等清理完成: {total} 条") finally: db.close() def trim_policy_versions(dry_run: bool) -> None: """policy_versions 只留最新 N 个。""" db = SessionLocal() try: all_pv = db.execute( select(PolicyVersion).order_by(PolicyVersion.created_at.desc()) ).scalars().all() keep = all_pv[:POLICY_KEEP_VERSIONS] drop = all_pv[POLICY_KEEP_VERSIONS:] log.info(f"[policy] 共 {len(all_pv)} 条历史版本, 留 {len(keep)} 删 {len(drop)}") if not drop: log.info(" → 无需清理") return if dry_run: for pv in drop: log.info(f" [dry-run] 将删 {pv.version}") return for pv in drop: db.delete(pv) db.commit() log.info(f" ✅ policy_versions 清理完成: 删 {len(drop)} 条") finally: db.close() def main() -> None: parser = argparse.ArgumentParser(description="MeshSafe DB 归档 + 清理") parser.add_argument("--apply", action="store_true", help="真正执行 (默认 dry-run)") args = parser.parse_args() dry = not args.apply log.info(f"模式: {'DRY-RUN (仅报告)' if dry else 'APPLY (真正执行)'}") log.info("") archive_tasks(dry) log.info("") scrub_detail(dry) log.info("") clean_idempotency(dry) log.info("") trim_policy_versions(dry) if dry: log.info("\n💡 确认无误后加 --apply 参数正式执行") if __name__ == "__main__": main()