fix: policy 加载 bug + DB 索引 + 归档脚本

【Bug 修复】policy.py _load_db_or_yaml
  根因: select(PolicyVersion).order_by(desc).scalar_one_or_none()
        表有多条历史版本 → Multiple rows 异常 → 每次启动都回退 YAML
  修复: 改成 db.query(PolicyVersion).order_by(desc).first()
        用 ORM Query API 取最新一条, 返回真正的 PolicyVersion 对象
  验证: 容器启动后日志中不再出现 'DB 加载策略失败（回退 YAML）'

【索引】models.py ScanTask 加复合索引
  ix_scan_tasks_channel_created ON (channel, created_at DESC)
  解决 dashboard/tasks_page 高频查询的全表扫描问题
  (已在运行中 DB 用 CREATE INDEX CONCURRENTLY 补上)

【归档】scripts/archive_old_tasks.py
  - scan_tasks 90天前 清 detail JSONB (省空间)
  - scan_tasks 365天前 移到 scan_tasks_archive 归档表
  - idempotency_records 7天 TTL 清过期幂等缓存
  - policy_versions 只留最新 5 个
  - 支持 --dry-run (默认) + --apply (真正执行)
  - 分批 1000 条处理, 避免长事务锁表
  - 归档表 scan_tasks_archive 已建 (含 4 个索引)

数据增长: 日均 523 条 → 年 ~19万条 ≈ 910 MB
当前 3660 条 ≈ 4.5 MB, 增长可控, 脚本可每周/每月 cron 跑一次

app/console.py

@@ -18,6 +18,48 @@ import threading
 import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+
+# ============================================================================
+# 内部测试 channel 黑名单
+# ============================================================================
+# 控制台页面（Dashboard / 任务历史）默认隐藏内部测试 / 对账脚本产生的记录。
+# 这些 channel 由开发 / 运维工具在内部调用 scanner.scan_file() 时自动产生，
+# 不应暴露给控制台用户。若后续新增内部脚本 channel，在此补充即可。
+INTERNAL_TEST_CHANNELS: frozenset[str] = frozenset({
+    "test_center",   # 测试靶场手动重扫
+    "v",             # 对账脚本 v5/v7/v8
+    "verify",        # 单条验证脚本
+    "diag",          # 诊断脚本
+    "debug",         # 调试脚本
+    "final",         # 最终验证
+    "console",       # console.py 内部调用
+    "console-test",  # console 测试模式
+    "regen_v2",      # 批量重建
+    "regenerate_index",  # index.json 重建
+    "audit_f3",      # 审计脚本
+    "audit",         # 审计脚本
+    "testcases",     # 生成 TEST_CASES.md
+    "test",          # 测试
+    "yara_verify",   # YARA 验证
+    "whitelist_v",   # 白名单验证
+    "boundary",      # 边界测试
+    "yara_v2",       # YARA v2 测试
+    "docgen",        # 文档生成脚本
+})
+
+
+def _apply_production_channel_filter(query, model_class):
+    """给 ScanTask / ScanFinding 查询追加「隐藏内部测试 channel」过滤。
+
+    用法: _apply_production_channel_filter(
+              db.query(func.count(ScanTask.id)), ScanTask
+          ).scalar()
+    """
+    col = getattr(model_class, "channel", None)
+    if col is None:
+        return query
+    return query.filter(col.notin_(INTERNAL_TEST_CHANNELS))
+
 from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, Response, UploadFile
@@ -535,30 +577,31 @@ async def dashboard(request: Request, current=Depends(require_login)):
     db: Session = SessionLocal()
     try:
         from sqlalchemy import func
-        total_scans = db.query(func.count(ScanTask.id)).scalar() or 0
+        _skip_test = ScanTask.channel.notin_(INTERNAL_TEST_CHANNELS)
+        total_scans = db.query(func.count(ScanTask.id)).filter(_skip_test).scalar() or 0
         total_blocked = db.query(func.count(ScanTask.id)).filter(
-            ScanTask.action == "reject"
+            ScanTask.action == "reject", _skip_test
         ).scalar() or 0
         total_flagged = db.query(func.count(ScanTask.id)).filter(
-            ScanTask.action.in_(("reject", "manual_review", "pass_with_flag"))
+            ScanTask.action.in_(("reject", "manual_review", "pass_with_flag")), _skip_test
         ).scalar() or 0
         total_findings = db.query(func.count(ScanFinding.id)).scalar() or 0
 
         # 结论分布
-        verdict_rows = db.query(ScanTask.verdict, func.count(ScanTask.id)).group_by(
-            ScanTask.verdict
-        ).all()
+        verdict_rows = db.query(ScanTask.verdict, func.count(ScanTask.id)).filter(
+            _skip_test
+        ).group_by(ScanTask.verdict).all()
         verdict_dist = {str(v or "unknown"): int(c) for v, c in verdict_rows}
 
         # 最近任务（8 条）
-        recent_tasks = db.query(ScanTask).order_by(
+        recent_tasks = db.query(ScanTask).filter(_skip_test).order_by(
             ScanTask.created_at.desc()
         ).limit(8).all()
 
         # 各客户端/渠道的任务数（摆渡关联性）
         client_rows = db.query(
             ScanTask.client_id, ScanTask.channel, func.count(ScanTask.id)
-        ).group_by(ScanTask.client_id, ScanTask.channel).all()
+        ).filter(_skip_test).group_by(ScanTask.client_id, ScanTask.channel).all()
     finally:
         db.close()
 
@@ -825,7 +868,8 @@ async def tasks_page(request: Request, current=Depends(require_login)):
 
     db: Session = SessionLocal()
     try:
-        total = db.query(func.count(ScanTask.id)).scalar() or 0
+        _skip_test = ScanTask.channel.notin_(INTERNAL_TEST_CHANNELS)
+        total = db.query(func.count(ScanTask.id)).filter(_skip_test).scalar() or 0
         total_pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
         # 越界时回退到最后一页
         if page > total_pages:
@@ -834,6 +878,7 @@ async def tasks_page(request: Request, current=Depends(require_login)):
 
         tasks = (
             db.query(ScanTask)
+            .filter(_skip_test)
             .order_by(ScanTask.created_at.desc())
             .limit(PAGE_SIZE)
             .offset(offset)

app/models.py

@@ -71,6 +71,11 @@ class ScanTask(Base):
         back_populates="task", cascade="all, delete-orphan", lazy="selectin"
     )
 
+    __table_args__ = (
+        # 控制台高频查询: WHERE channel = X ORDER BY created_at DESC LIMIT N
+        Index("ix_scan_tasks_channel_created", "channel", "created_at", postgresql_using="btree"),
+    )
+
 
 class ScanFinding(Base):
     __tablename__ = "scan_findings"

app/services/policy.py

@@ -300,15 +300,18 @@ class PolicyStore:
     def _load_db_or_yaml(self) -> Policy:
         """优先 DB，回退 YAML。"""
         try:
-            from sqlalchemy import select
             from app.db import SessionLocal
             from app.models import PolicyVersion
 
             db = SessionLocal()
             try:
-                pv = db.execute(
-                    select(PolicyVersion).order_by(PolicyVersion.created_at.desc())
-                ).scalar_one_or_none()
+                # 用 ORM Query API — 直接返回 PolicyVersion 对象
+                # 而不是 Row 对象（db.execute(select) 会返回 Row）
+                pv = (
+                    db.query(PolicyVersion)
+                    .order_by(PolicyVersion.created_at.desc())
+                    .first()   # 取最新一条，不管历史版本多少条
+                )
                 if pv:
                     policy = Policy(
                         version=pv.version,

scripts/archive_old_tasks.py

@@ -0,0 +1,264 @@
+"""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()