fix: 排行榜支持总消息计数并保留图片事实校验

app/api/groups.py

@@ -49,6 +49,7 @@ class GroupCreate(BaseModel):
     prompt_provider: str = ""
     summary_model: str = ""
     prompt_model: str = ""
+    strict_image_fact_check: bool = False
     image_enabled: bool = True
     send_target: str = ""
     ranking_template: str = "default"
@@ -75,6 +76,7 @@ class GroupUpdate(BaseModel):
     prompt_provider: str | None = None
     summary_model: str | None = None
     prompt_model: str | None = None
+    strict_image_fact_check: bool | None = None
     image_enabled: bool | None = None
     send_target: str | None = None
     ranking_template: str | None = None
@@ -240,6 +242,7 @@ def list_groups(
             "prompt_provider": g.prompt_provider,
             "summary_model": g.summary_model,
             "prompt_model": g.prompt_model,
+            "strict_image_fact_check": g.strict_image_fact_check,
             "image_enabled": g.image_enabled,
             "send_target": g.send_target,
             "effective_send_target": effective_send_target(g),

app/api/v2_ui_read.py

@@ -73,12 +73,15 @@ def dashboard(
         image_url = ""
         if image_path.exists() and Path(image_path).stat().st_size > 0:
             image_url = f"/api/v2/files/{quote(name)}/{selected_run_date}/{FILE_IMAGE}"
+        ranking_count_policy = "all_messages"
         ranking_preview: list[dict[str, object]] = []
         ranking_error = ""
         ranking_path = store.ranking_json_path(name, selected_run_date)
         if ranking_path.exists() and ranking_path.stat().st_size > 0:
             try:
                 ranking = json.loads(ranking_path.read_text(encoding="utf-8"))
+                if isinstance(ranking, dict) and ranking.get("count_policy") == "text_primary_with_interactions":
+                    ranking_count_policy = "text_primary_with_interactions"
                 speakers = ranking.get("top_speakers", []) if isinstance(ranking, dict) else []
                 if not isinstance(speakers, list):
                     raise ValueError("top_speakers 不是数组")
@@ -114,9 +117,7 @@ def dashboard(
                     getattr(group, "wechat_send_enabled", False)
                 ),
                 "ranking_template": group.ranking_template,
-                "ranking_count_policy": getattr(
-                    group, "ranking_count_policy", "all_messages"
-                ),
+                "ranking_count_policy": ranking_count_policy,
                 "image_prompt_template": group.image_prompt_template,
                 "status": status,
                 "period_start": run.get("period_start", ""),

app/db/models.py

@@ -36,6 +36,7 @@ class Group(SQLModel, table=True):
     prompt_provider: str = ""  # 日报 Prompt AI；空值继承全局
     summary_model: str = "gpt-6-astra"  # 总结主模型
     prompt_model: str = "gpt-6-astra"  # Prompt 主模型
+    strict_image_fact_check: bool = False  # 独立开启图片事实校验
     image_enabled: bool = True  # 是否生图
     send_target: str = ""  # 可选人工发送目标；为空时自动跟随 wechat_group_name
     ranking_template: str = "default"  # 排行榜模板名

app/db/repository.py

@@ -183,6 +183,7 @@ def _ensure_relationship_schema_current() -> None:
     "prompt_model": "VARCHAR(64) NOT NULL DEFAULT 'gpt-6-astra'",
     "image_enabled": "BOOLEAN NOT NULL DEFAULT 1",
     "send_target": "VARCHAR(256) NOT NULL DEFAULT ''",
+    "strict_image_fact_check": "BOOLEAN NOT NULL DEFAULT 0",
     "ranking_template": "VARCHAR(64) NOT NULL DEFAULT 'default'",
     "ranking_count_policy": "VARCHAR(64) NOT NULL DEFAULT 'all_messages'",
     "sender_name_policy": "VARCHAR(64) NOT NULL DEFAULT 'resolved'",

app/image/fact_verification.py

@@ -57,7 +57,8 @@ def strict_fact_verification_enabled(prompt_file: Path) -> bool:
     if not isinstance(run, dict):
         return False
     return bool(
-        run.get("image_fact_contract") == "strict_evidence_v1"
+        run.get("strict_image_fact_check") is True
+        or run.get("image_fact_contract") == "strict_evidence_v1"
         or run.get("ranking_count_policy") == RANKING_POLICY_TEXT_PRIMARY
     )
 

app/pipeline/daily_pipeline.py

@@ -154,7 +154,7 @@ def generate_all(
                 "wechat_group_id", "wechat_group_name", "provider_preference",
                 "schedule_rule", "summary_provider", "summary_model",
                 "prompt_provider", "prompt_model", "image_enabled", "ranking_template",
-                "ranking_count_policy", "sender_name_policy",
+                "ranking_count_policy", "sender_name_policy", "strict_image_fact_check",
                 "image_prompt_template", "image_theme", "image_theme_custom",
                 "image_theme_remaining_runs",
                 "image_prompt_override", "send_target",

app/pipeline/generation_stages.py

@@ -324,6 +324,9 @@ def _prepare_run(
             "period_start": context.period_start,
             "period_end": context.period_end,
             "send_time": self.settings.schedule_send_time,
+            "strict_image_fact_check": bool(
+                run.get("strict_image_fact_check", group.strict_image_fact_check)
+            ),
             "image_enabled": bool(group.image_enabled),
             "ranking_template": group.ranking_template,
             "ranking_count_policy": getattr(
@@ -857,8 +860,12 @@ def _execute_prompt_operation(
             prompt_meta = committed.get("prompt_meta")
 
         self._record_prompt_timing(context, started_at)
+        run = self.store.load_run(context.group_name, context.run_date)
         if uses_strict_image_fact_contract(
-            getattr(context.group, "ranking_count_policy", "all_messages")
+            getattr(context.group, "ranking_count_policy", "all_messages"),
+            strict_image_fact_check=bool(
+                run.get("strict_image_fact_check", context.group.strict_image_fact_check)
+            ),
         ):
             prompt_path = self.store.prompt_path(context.group_name, context.run_date)
             strict_prompt = append_strict_image_fact_contract(

app/ranking/policies.py

@@ -30,5 +30,9 @@ def normalize_sender_name_policy(value: object) -> str:
     return policy
 
 
-def uses_strict_image_fact_contract(ranking_policy: object) -> bool:
-    return normalize_ranking_policy(ranking_policy) == RANKING_POLICY_TEXT_PRIMARY
+def uses_strict_image_fact_contract(
+    ranking_policy: object, strict_image_fact_check: bool = False,
+) -> bool:
+    # 兼容旧文字排行，允许全消息排行独立启用校验。
+    policy = normalize_ranking_policy(ranking_policy)
+    return strict_image_fact_check or policy == RANKING_POLICY_TEXT_PRIMARY

app/scheduler/task_manifest.py

@@ -47,6 +47,7 @@ def build_expected_groups(
                 "prompt_provider": str(getattr(group, "prompt_provider", "") or ""),
                 "prompt_model": str(group.prompt_model or ""),
                 "send_time": str(schedule_send_time or "08:30"),
+                "strict_image_fact_check": bool(getattr(group, "strict_image_fact_check", False)),
                 "image_enabled": bool(group.image_enabled),
                 "ranking_template": str(group.ranking_template or "default"),
                 "ranking_count_policy": str(

docs/TOTAL_MESSAGE_RANKING_ROLLOUT.md

@@ -0,0 +1,36 @@
+# 六群总消息排行榜切换
+
+## 行为
+下一次新任务使用 all_messages + default，所有可计数非系统消息各计一条。
+只展示发言人数、总消息与 Top 名单。旧模板、历史排名和图片保持原样。
+strict_image_fact_check 是独立布尔群配置，默认 false；本次六群设为 true。
+旧 text_primary_with_interactions 继续隐式开启严格图片校验。
+群 API 支持读写该字段；任务清单、run.json 和恢复路径保存该设置。
+Dashboard 根据选中 ranking.json 的 count_policy 展示，缺少字段按 all_messages 兼容。
+
+## 正式生效步骤（尚未执行）
+须先完成 PR 验收并获得用户当次合并、生产更新和服务重启授权。
+1. 核实正式监听进程、管理器、实际数据库路径和当天任务状态；如有在途生成或发送，等待完成。
+2. 记录当天及历史产物校验值和发送状态，在停止正式服务后确认 8766 已释放，调度器和独立任务进程均已退出。
+3. 在已更新代码的目录使用 Python 执行以下脚本。数据库参数必须是上一步核实的正式文件；不得使用工作目录中新建的空数据库。
+4. 应用后构建前端、按原管理方式启动服务，核实监听 PID、进程 ancestry、HTTP 健康、调度状态和群配置。
+5. 六个活动群 ID 23–28 应均为 all_messages / default / strict_image_fact_check=true；其他字段保持原值。核对历史产物校验值与发送状态未改变。不得通过真实取数、生图或发送来测试这次切换。
+
+只读预览：
+```powershell
+python scripts/configure_total_message_ranking.py --database "已核实的数据库绝对路径"
+```
+
+取得授权且服务停止后应用：
+```powershell
+python scripts/configure_total_message_ranking.py --database "已核实的数据库绝对路径" --apply --service-stopped
+```
+
+脚本先验证数据库和活动群集合，再通过 SQLite backup API 生成同目录带时间戳的完整备份。
+事务中补齐新列、只更新六群的三个配置字段，并检查完整性、外键和所有其他群字段。
+活动群集合或旧配置发生变化时拒绝执行。脚本不读写日报产物、不调用网络业务接口。
+
+## 回滚
+配置切换后若尚无新业务写入，可在服务停止状态下使用已验证备份恢复。
+若已有新业务写入，禁止整库覆盖；从备份读取六群旧配置，仅事务恢复三个字段，保留其他业务数据。
+独立图片校验列可保留，新代码仍兼容旧排行榜。

frontend/src/api.ts

@@ -75,6 +75,7 @@ export interface LatestReport {
 // ================= V2 =================
 
 export interface GroupV2 extends Group {
+  strict_image_fact_check?: boolean;
   schedule_rule: string;
   send_time: string;
   summary_provider: string;
@@ -97,6 +98,7 @@ export interface GroupV2 extends Group {
 }
 
 export interface GroupPayload {
+  strict_image_fact_check?: boolean;
   display_name: string;
   wechat_group_id: string;
   wechat_group_name?: string;

frontend/src/pages/v2/GroupDetail.tsx

@@ -61,6 +61,7 @@ const EMPTY_FORM: GroupPayload = {
   prompt_provider: "",
   summary_model: "",
   prompt_model: "",
+  strict_image_fact_check: false,
   image_enabled: true,
   send_target: "",
   ranking_template: "default",
@@ -91,6 +92,7 @@ function toForm(group: GroupV2): GroupPayload {
     prompt_provider: group.prompt_provider || "",
     summary_model: group.summary_model || "",
     prompt_model: group.prompt_model || "",
+    strict_image_fact_check: group.strict_image_fact_check ?? false,
     image_enabled: group.image_enabled,
     send_target: group.send_target || "",
     ranking_template: group.ranking_template || "default",
@@ -436,6 +438,10 @@ export default function GroupDetail({ groupId, invalidGroupId }: GroupDetailProp
             <input id="image-enabled" type="checkbox" checked={form.image_enabled} onChange={(event) => setField("image_enabled", event.target.checked)} />
             <span><strong>启用 AI 图片</strong><small>启用后会进入最多 2 路的受控生图阶段</small></span>
           </label>
+          <label className="group-detail-switch" htmlFor="strict-image-fact-check">
+            <input id="strict-image-fact-check" type="checkbox" checked={Boolean(form.strict_image_fact_check) || form.ranking_count_policy === "text_primary_with_interactions"} disabled={form.ranking_count_policy === "text_primary_with_interactions"} onChange={(event) => setField("strict_image_fact_check", event.target.checked)} />
+            <span><strong>严格核对图片事实</strong><small>核对图片内容与聊天证据；旧文字排行始终启用</small></span>
+          </label>
           <Field id="send-target" label="发送目标（可选人工覆盖）" error={errors.send_target}>
             <input id="send-target" value={form.send_target} onChange={(event) => setField("send_target", event.target.value)} placeholder="留空则自动跟随微信当前群名" />
             <span className="group-detail-field-help">

frontend/src/pages/v2/rankingPolicy.test.ts

@@ -21,7 +21,7 @@ describe("ranking policy display", () => {
 
   it("keeps legacy rankings compatible", () => {
     expect(isTextPrimaryRanking("all_messages")).toBe(false);
-    expect(formatRankingCount("all_messages", { count: 89 })).toBe("89 条");
+    expect(formatRankingCount("all_messages", { count: 89, text_count: 55, interaction_count: 34 })).toBe("89 条");
   });
 
   it("falls back safely when an early strict record lacks extended counts", () => {

scripts/configure_total_message_ranking.py

@@ -0,0 +1,113 @@
+"""六群切换总消息排行。默认只读；--apply 仅可在正式服务停止后使用。"""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime
+import json
+from pathlib import Path
+import socket
+import sqlite3
+import sys
+
+GROUP_IDS = (23, 24, 25, 26, 27, 28)
+
+
+def inspect_groups(connection: sqlite3.Connection) -> list[dict]:
+    connection.row_factory = sqlite3.Row
+    rows = [dict(row) for row in connection.execute(
+        "SELECT * FROM groups WHERE enabled = 1 AND deleted_at IS NULL ORDER BY id"
+    )]
+    if tuple(row["id"] for row in rows) != GROUP_IDS:
+        raise ValueError("活动群已变化：必须重新核对目标，当前脚本只接受群 23–28")
+    for row in rows:
+        old = (row["ranking_count_policy"], row["ranking_template"])
+        if old not in {
+            ("text_primary_with_interactions", "text_interactions"),
+            ("all_messages", "default"),
+        } or row["sender_name_policy"] != "wechat_data_analysis":
+            raise ValueError(f"群 {row['id']} 配置与预期不符，停止更新")
+    return rows
+
+
+def check_database(connection: sqlite3.Connection) -> None:
+    if connection.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
+        raise ValueError("数据库完整性校验失败")
+    if connection.execute("PRAGMA foreign_key_check").fetchall():
+        raise ValueError("数据库外键校验失败")
+
+
+def configure(database: Path, *, apply: bool = False) -> dict:
+    database = database.resolve(strict=True)
+    with sqlite3.connect(database.as_uri() + "?mode=ro", uri=True) as source:
+        check_database(source)
+        before = inspect_groups(source)
+        preview = [
+            {
+                "id": row["id"],
+                "display_name": row["display_name"],
+                "before": {key: row.get(key, False) for key in (
+                    "ranking_count_policy", "ranking_template", "strict_image_fact_check"
+                )},
+                "after": {
+                    "ranking_count_policy": "all_messages",
+                    "ranking_template": "default",
+                    "strict_image_fact_check": True,
+                },
+            }
+            for row in before
+        ]
+        if not apply:
+            return {"applied": False, "database": str(database), "groups": preview}
+        backup = database.with_name(
+            database.name + ".ranking-backup-" + datetime.now().strftime("%Y%m%d-%H%M%S-%f")
+        )
+        with sqlite3.connect(backup) as target:
+            source.backup(target)
+            check_database(target)
+    with sqlite3.connect(database) as connection:
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("BEGIN IMMEDIATE")
+        if inspect_groups(connection) != before:
+            raise ValueError("备份后配置发生变化，停止更新")
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(groups)")}
+        if "strict_image_fact_check" not in columns:
+            connection.execute(
+                "ALTER TABLE groups ADD COLUMN strict_image_fact_check BOOLEAN NOT NULL DEFAULT 0"
+            )
+        connection.execute(
+            "UPDATE groups SET ranking_count_policy='all_messages', "
+            "ranking_template='default', strict_image_fact_check=1 "
+            "WHERE id IN (23,24,25,26,27,28)"
+        )
+        after = inspect_groups(connection)
+        for old, new in zip(before, after):
+            expected = {
+                **old, "ranking_count_policy": "all_messages",
+                "ranking_template": "default", "strict_image_fact_check": 1,
+            }
+            if new != expected:
+                raise ValueError("更新后的群配置不符合预期")
+        check_database(connection)
+    return {"applied": True, "database": str(database), "backup": str(backup), "groups": preview}
+
+
+def main() -> None:
+    if hasattr(sys.stdout, "reconfigure"):
+        sys.stdout.reconfigure(encoding="utf-8")
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--database", type=Path, required=True)
+    parser.add_argument("--apply", action="store_true")
+    parser.add_argument("--service-stopped", action="store_true")
+    args = parser.parse_args()
+    if args.apply:
+        if not args.service_stopped:
+            parser.error("应用前须授权并停止正式服务，再加 --service-stopped")
+        with socket.socket() as probe:
+            probe.settimeout(1)
+            if probe.connect_ex(("127.0.0.1", 8766)) == 0:
+                parser.error("8766 仍在监听，拒绝在正式服务运行时更新")
+    print(json.dumps(configure(args.database, apply=args.apply), ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/test_generation_concurrency.py

@@ -256,7 +256,10 @@ def test_prompt_ready_group_starts_image_before_other_prompts_finish(tmp_path, m
     assert all(item["status"] == "ready_to_send" for item in results)
 
 
-def test_group_overrides_rebuild_loaded_group_with_live_sqlalchemy_state(tmp_path, monkeypatch):
+@pytest.mark.parametrize("saved_fact_check", [None, True, False])
+def test_group_overrides_rebuild_loaded_group_with_live_sqlalchemy_state(
+    tmp_path, monkeypatch, saved_fact_check,
+):
     engine = create_engine(
         f"sqlite:///{tmp_path / 'group-overrides.db'}",
         connect_args={"check_same_thread": False},
@@ -289,12 +292,17 @@ def test_group_overrides_rebuild_loaded_group_with_live_sqlalchemy_state(tmp_pat
         dry_run=True,
     )
 
+    if saved_fact_check is not None:
+        pipeline.store.update(
+            "覆盖配置群", "2026-08-21", strict_image_fact_check=saved_fact_check,
+        )
     results = pipeline.generate_all(
         run_date="2026-08-21",
         group_overrides={
             group_id: {
                 "image_enabled": False,
                 "image_theme": "ai_free",
+                "strict_image_fact_check": True,
             }
         },
     )
@@ -309,6 +317,9 @@ def test_group_overrides_rebuild_loaded_group_with_live_sqlalchemy_state(tmp_pat
     run = pipeline.store.load_run("覆盖配置群", "2026-08-21")
     assert run["image_enabled"] is False
     assert run["image_theme"] == "ai_free"
+    expected_fact_check = True if saved_fact_check is None else saved_fact_check
+    assert run["strict_image_fact_check"] is expected_fact_check
+    assert (run.get("image_fact_contract") == "strict_evidence_v1") is expected_fact_check
 
 
 def test_unexpected_worker_and_image_errors_are_isolated(tmp_path, monkeypatch):

tests/test_image_fact_verification.py

@@ -253,3 +253,18 @@ def test_strict_contract_removes_bmi_display_instructions():
     assert "话题延伸到身高和婚后发福" in strict_prompt
     assert "猜体重" in strict_prompt
     assert "我的" not in strict_prompt
+
+
+def test_independent_fact_check_survives_all_message_ranking(tmp_path):
+    import json
+    from app.image.fact_verification import strict_fact_verification_enabled
+
+    prompt = tmp_path / "prompt.txt"
+    (tmp_path / "run.json").write_text(json.dumps({
+        "ranking_count_policy": "all_messages", "strict_image_fact_check": True,
+    }), encoding="utf-8")
+    assert strict_fact_verification_enabled(prompt) is True
+    (tmp_path / "run.json").write_text(json.dumps({
+        "ranking_count_policy": "all_messages", "strict_image_fact_check": False,
+    }), encoding="utf-8")
+    assert strict_fact_verification_enabled(prompt) is False