265 lines
9.1 KiB
Python
265 lines
9.1 KiB
Python
|
"""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()
|