fix: 禁止统计兜底图并限次使用主题

.codex/TASK_DISABLE_STAT_IMAGE_FALLBACK.md

@@ -0,0 +1,68 @@
+# 禁止统计兜底图进入日报生图流程
+
+## 背景
+
+2026-09-05 的两个群在真实图片事实校验失败后生成了 Level 3 Pillow 统计诊断图。后续重试仅按 PNG 可解码性复用了 `daily_image.png`，并将它记录为 `fallback_level=0`、`image_variant=normal`。另一个群因真实图及兜底图均包含无证据数字而没有图片。
+
+## 目标
+
+- 以后日报只接受正常 AI 生图或现有安全化 AI 生图流程的产物。
+- 图片生成或事实校验失败时明确失败，不生成排行榜/统计表兜底图。
+- 重试不得把旧 Level 3/Pillow 诊断图当作正常图片复用。
+- 群级自定义关键词主题默认只作用于下一次新运行；界面可明确选择有限使用次数；额度消费后自动回到 `random_preset`。
+- 修复和生产生效后，将 2026-09-05 的六个群全部切回 `random_preset`，重新生成并在逐图、逐目标验收后发送。
+
+## 允许修改范围
+
+- `app/image/`
+- `app/pipeline/`（仅图片结果状态衔接需要时）
+- `app/db/`、`app/api/groups.py`、调度任务快照（仅主题有限次数所需字段与原子消费）
+- `frontend/src/` 中主题次数选择及对应 API 类型
+- `tests/` 中对应图片生成、流水线测试及 CI 随机顺序隔离修复
+- 本任务说明文件
+
+## 禁止修改范围
+
+- 与本次六群随机重生成无关的历史 `output/`、日志和数据库内容
+- 微信发送实现、群聊目标
+- Provider、Codex 登录/认证方式
+- 部署、服务重启、PR 合并
+
+## 已确定实现要求
+
+- 移除生产失败路径对 `render_local_infographic` 的调用，不再写出统计兜底 `daily_image.png`。
+- 失败前清理本次失败路径遗留的不可交付目标文件，避免后续仅按图片格式复用。
+- 现有文件复用必须结合运行元数据；Level 3、Pillow、`diagnostic_fallback` 或已知诊断图不得返回成功。
+- 保留真实图片两次事实校验及现有安全化 AI 重试流程。
+- 失败状态保持 fail-closed，不进入 `READY_TO_SEND`。
+- 生图前去除摘要阶段擅自附加、但原始聊天没有依据的数字单位；首次 OCR 事实校验失败后，使用列明拒绝项的纠错 Prompt 正常重画一次。
+- 自定义主题保存时默认次数为 1；允许选择有限正整数。图片成功生成并记录结果时原子扣减一次，同一天运行的重试和 Prompt 重建不重复扣减，失败或结果未知不扣减；扣至 0 后群配置恢复每日随机。
+- `ai_free` 与 `random_preset` 是持续模式，不消耗次数。
+- 六群实际生成必须先清除当天旧图的可复用条件并保留发送锁；所有图片通过 PNG、尺寸、事实契约、归属路径与哈希验收后，才逐群验证发送目标并串行发送。
+
+## 验收标准
+
+- 图片失败时无统计兜底图，结果为失败且不可发送。
+- 旧诊断图即使是可解码 PNG，也不会被 `existing_output_reused` 接纳。
+- 正常已有图片仍可复用，不额外调用 Provider。
+- 自定义关键词主题默认 1 次，可选多次；新运行只扣一次，重试不重复扣，耗尽后显示每日随机。
+- 相关测试、完整图片/流水线测试、静态检查通过。
+- 实际 diff 无范围外修改、敏感信息、TODO/debug 或临时文件。
+
+## 测试命令
+
+- `.venv\Scripts\python.exe -m pytest tests/test_v2_image_task.py tests/test_v2_pipeline.py -q`
+- `.venv\Scripts\python.exe -m pytest tests/test_image_fact_verification.py tests/test_runtime_status.py tests/test_v2_ui_router_contract.py -q`
+- `.venv\Scripts\python.exe -m pytest tests/test_group_image_theme_batch_api.py tests/test_generation_concurrency.py tests/test_daily_random_theme.py -q`
+- `npm --prefix frontend test -- --run`
+- `npm --prefix frontend run build`
+- `.venv\Scripts\python.exe -m compileall app tests`
+- `git diff --check`
+
+## 返回格式
+
+- 根因证据
+- 修改文件与行为变化
+- 测试命令和结果
+- Git 分支、提交、Push、PR、CI 状态
+- 六群随机主题配置、生成验收、逐群发送证据；生产运行是否已生效

app/api/groups.py

@@ -57,6 +57,7 @@ class GroupCreate(BaseModel):
     image_prompt_template: str = "default"
     image_theme: str = DEFAULT_IMAGE_THEME
     image_theme_custom: str = ""
+    image_theme_apply_count: int | None = Field(default=None, ge=1, le=30)
     image_prompt_override: str = ""
     wechat_send_enabled: bool = False
 
@@ -82,6 +83,7 @@ class GroupUpdate(BaseModel):
     image_prompt_template: str | None = None
     image_theme: str | None = None
     image_theme_custom: str | None = None
+    image_theme_apply_count: int | None = Field(default=None, ge=1, le=30)
     image_prompt_override: str | None = None
     wechat_send_enabled: bool | None = None
 
@@ -96,13 +98,15 @@ class GroupImagePromptUpdate(BaseModel):
     inherit_global: bool = False
     image_theme: str
     image_theme_custom: str = ""
+    image_theme_apply_count: int = Field(default=1, ge=1, le=30)
     expected_revision: str = ""
 
 
 class BatchImageThemeUpdate(BaseModel):
     group_ids: list[int] = Field(min_length=1)
     image_theme: str
     image_theme_custom: str = ""
+    image_theme_apply_count: int = Field(default=1, ge=1, le=30)
 
     @field_validator("group_ids")
     @classmethod
@@ -122,6 +126,14 @@ def _validate_group_theme(theme: object, custom: object = "") -> tuple[str, str]
         raise HTTPException(status_code=422, detail=str(exc)) from exc
 
 
+def _theme_remaining_runs(theme: str, apply_count: int | None) -> int:
+    """自由/每日随机持续生效；任何手动主题默认只用于下一次成功生图。"""
+
+    if theme in {"ai_free", "random_preset"}:
+        return 0
+    return int(apply_count or 1)
+
+
 def _validate_output_group_name(value: object, *, field_name: str) -> str:
     """拒绝会被当作文件路径的群名称，同时保留普通显示名标点。"""
     text = str(value or "")
@@ -175,6 +187,7 @@ def _group_prompt_payload(group: Group) -> dict:
         "revision": prompt_revision(content),
         "image_theme": group.image_theme or DEFAULT_IMAGE_THEME,
         "image_theme_custom": group.image_theme_custom or "",
+        "image_theme_remaining_runs": int(group.image_theme_remaining_runs or 0),
         "resolved_theme": theme.to_meta(),
         "preview": preview,
     }
@@ -237,6 +250,7 @@ def list_groups(
             "image_prompt_template": g.image_prompt_template,
             "image_theme": g.image_theme,
             "image_theme_custom": g.image_theme_custom,
+            "image_theme_remaining_runs": int(g.image_theme_remaining_runs or 0),
             "has_image_prompt_override": bool((g.image_prompt_override or "").strip()),
             "wechat_send_enabled": bool(g.wechat_send_enabled),
             "created_at": g.created_at.isoformat(),
@@ -270,9 +284,13 @@ def create_group(
             values["wechat_group_name"], field_name="wechat_group_name"
         )
     values["send_target"] = str(values.get("send_target") or "").strip()
+    apply_count = values.pop("image_theme_apply_count", None)
     values["image_theme"], values["image_theme_custom"] = _validate_group_theme(
         values.get("image_theme", DEFAULT_IMAGE_THEME), values.get("image_theme_custom", "")
     )
+    values["image_theme_remaining_runs"] = _theme_remaining_runs(
+        values["image_theme"], apply_count
+    )
     values["image_prompt_override"] = _validate_prompt_override(values.get("image_prompt_override", ""))
     wechat_group_id = str(values.get("wechat_group_id") or "").strip()
     existing = repo.find_group_by_wechat_id(session, wechat_group_id) if wechat_group_id else None
@@ -325,6 +343,9 @@ def batch_update_group_image_theme(
                 group_id,
                 image_theme=theme,
                 image_theme_custom=custom,
+                image_theme_remaining_runs=_theme_remaining_runs(
+                    theme, payload.image_theme_apply_count
+                ),
             )
         except SQLAlchemyError:
             session.rollback()
@@ -344,6 +365,9 @@ def batch_update_group_image_theme(
         successes.append({
             "group_id": group_id,
             "group_name": group_name,
+            "remaining_runs": _theme_remaining_runs(
+                theme, payload.image_theme_apply_count
+            ),
         })
 
     status = "success" if not failures else "partial" if successes else "failed"
@@ -369,6 +393,7 @@ def update_group(
         settings,
         explicitly_provided="send_time" in payload.model_fields_set,
     )
+    apply_count = updates.pop("image_theme_apply_count", None)
     try:
         updates = validate_group_provider_values(
             updates,
@@ -389,6 +414,13 @@ def update_group(
         )
         updates["image_theme"] = theme
         updates["image_theme_custom"] = custom
+        updates["image_theme_remaining_runs"] = _theme_remaining_runs(
+            theme, apply_count
+        )
+    elif apply_count is not None:
+        updates["image_theme_remaining_runs"] = _theme_remaining_runs(
+            group.image_theme, apply_count
+        )
     if "image_prompt_override" in updates:
         updates["image_prompt_override"] = _validate_prompt_override(updates["image_prompt_override"])
     if "send_target" in updates:
@@ -423,6 +455,9 @@ def update_group_image_prompt(
         raise HTTPException(status_code=422, detail="群级 Prompt 不能为空；如需继承请使用恢复全局模板")
     group.image_theme = theme
     group.image_theme_custom = custom
+    group.image_theme_remaining_runs = _theme_remaining_runs(
+        theme, payload.image_theme_apply_count
+    )
     group.image_prompt_override = "" if payload.inherit_global else _validate_prompt_override(payload.content)
     repo.save_group(session, group)
     return _group_prompt_payload(group)

app/db/models.py

@@ -44,6 +44,7 @@ class Group(SQLModel, table=True):
     image_prompt_template: str = "default"  # 生图 Prompt 模板名
     image_theme: str = "ai_free"  # 生图主题键（默认由 AI 按聊天内容自由发挥）
     image_theme_custom: str = ""  # 自定义生图大主题（image_theme=custom 时使用）
+    image_theme_remaining_runs: int = 0  # 手动主题剩余成功生图次数；0 表示持续模式不计次
     image_prompt_override: str = ""  # 本群专属 Prompt 模板；为空时继承全局模板
     wechat_send_enabled: bool = False  # 独立于生成开关，默认禁止自动对外发送
 

app/db/repository.py

@@ -8,6 +8,7 @@
 from typing import Any
 
 from sqlalchemy import event, update
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlmodel import Session, SQLModel, create_engine, select
 
 from app.config.settings import Settings
@@ -188,6 +189,7 @@ def _ensure_relationship_schema_current() -> None:
     "image_prompt_template": "VARCHAR(64) NOT NULL DEFAULT 'default'",
     "image_theme": "VARCHAR(64) NOT NULL DEFAULT 'ai_free'",
     "image_theme_custom": "VARCHAR(80) NOT NULL DEFAULT ''",
+    "image_theme_remaining_runs": "INTEGER NOT NULL DEFAULT 0",
     "image_prompt_override": "TEXT NOT NULL DEFAULT ''",
     "wechat_send_enabled": "BOOLEAN NOT NULL DEFAULT 0",
     "deleted_at": "DATETIME NULL",
@@ -245,6 +247,23 @@ def _migrate_parallel_summary_defaults() -> None:
         session.commit()
 
 
+def _migrate_finite_image_theme_defaults() -> None:
+    """让升级前已保存的手动主题只再使用一次，随后回到每日随机。"""
+
+    marker = "migration_finite_image_theme_uses_v1"
+    with Session(engine) as session:
+        if session.get(Setting, marker) is not None:
+            return
+        session.exec(
+            Group.__table__.update()
+            .where(Group.image_theme.notin_(("ai_free", "random_preset")))
+            .where(Group.image_theme_remaining_runs <= 0)
+            .values(image_theme_remaining_runs=1)
+        )
+        session.add(Setting(key=marker, value="done"))
+        session.commit()
+
+
 def _migrate_daily_schedule_defaults() -> None:
     """一次性把旧 00:01 生成/发送默认值升级为 00:30/08:30。
 
@@ -380,6 +399,7 @@ def init_db(settings: Settings) -> Any:
     _seed_defaults(settings)
     # 先种默认值再迁移，确保旧 .env 中的 deepseek-chat / 12000 也会升级。
     _migrate_parallel_summary_defaults()
+    _migrate_finite_image_theme_defaults()
     _migrate_codex_summary_defaults()
     _migrate_codex_image_timeout_default()
     _migrate_daily_schedule_defaults()
@@ -529,6 +549,7 @@ def update_group_image_theme(
     *,
     image_theme: str,
     image_theme_custom: str,
+    image_theme_remaining_runs: int,
 ) -> bool:
     """独立事务只更新群生图主题字段，避免通用保存路径带入其他配置。"""
 
@@ -538,13 +559,94 @@ def update_group_image_theme(
         .values(
             image_theme=image_theme,
             image_theme_custom=image_theme_custom,
+            image_theme_remaining_runs=max(int(image_theme_remaining_runs), 0),
             updated_at=_now(),
         )
     )
     session.commit()
     return result.rowcount == 1
 
 
+def consume_group_image_theme_for_run(
+    session: Session,
+    group_id: int,
+    *,
+    run_date: str,
+    expected_theme: str,
+    expected_custom: str,
+) -> dict[str, Any]:
+    """为一个新运行至多消费一次手动主题；同一事务内写幂等标记并更新群配置。"""
+
+    if expected_theme in {"ai_free", "random_preset"}:
+        return {
+            "consumed": False,
+            "already_consumed": False,
+            "remaining_runs": 0,
+            "next_theme": expected_theme,
+        }
+
+    marker = f"image_theme_use:{group_id}:{run_date}"
+    claim = session.execute(
+        sqlite_insert(Setting)
+        .values(key=marker, value=expected_theme[:64], updated_at=_now())
+        .on_conflict_do_nothing(index_elements=["key"])
+    )
+    if claim.rowcount != 1:
+        session.rollback()
+        group = session.get(Group, group_id)
+        return {
+            "consumed": False,
+            "already_consumed": True,
+            "remaining_runs": max(
+                int(getattr(group, "image_theme_remaining_runs", 0) or 0), 0
+            )
+            if group is not None
+            else 0,
+            "next_theme": str(getattr(group, "image_theme", "random_preset") or "random_preset")
+            if group is not None
+            else "random_preset",
+        }
+
+    group = session.get(Group, group_id)
+    matches_snapshot = bool(
+        group is not None
+        and group.deleted_at is None
+        and str(group.image_theme or "") == expected_theme
+        and str(group.image_theme_custom or "") == expected_custom
+    )
+    if not matches_snapshot:
+        session.commit()
+        return {
+            "consumed": False,
+            "already_consumed": False,
+            "config_changed": True,
+            "remaining_runs": max(
+                int(getattr(group, "image_theme_remaining_runs", 0) or 0), 0
+            )
+            if group is not None
+            else 0,
+            "next_theme": str(getattr(group, "image_theme", "random_preset") or "random_preset")
+            if group is not None
+            else "random_preset",
+        }
+
+    remaining = max(int(group.image_theme_remaining_runs or 0), 1)
+    next_remaining = remaining - 1
+    if next_remaining == 0:
+        group.image_theme = "random_preset"
+        group.image_theme_custom = ""
+    group.image_theme_remaining_runs = next_remaining
+    group.updated_at = _now()
+    session.add(group)
+    session.commit()
+    return {
+        "consumed": True,
+        "already_consumed": False,
+        "remaining_runs": next_remaining,
+        "next_theme": str(group.image_theme or "random_preset"),
+    }
+
+
 def delete_group(session: Session, group_id: int) -> Group | None:
     group = session.get(Group, group_id)
     if group is None:

app/image/delivery_guard.py

@@ -20,4 +20,35 @@ def image_delivery_eligible(metadata: Mapping[str, Any] | None) -> bool:
     metadata = metadata if isinstance(metadata, Mapping) else {}
     fallback_level = image_fallback_level(metadata)
     image_variant = str(metadata.get("image_variant") or "").strip().lower()
-    return fallback_level < 3 and image_variant != "pillow"
+    image_status = str(metadata.get("image_status") or "").strip().lower()
+    image_job = metadata.get("image_job")
+    job_status = (
+        str(image_job.get("status") or "").strip().lower()
+        if isinstance(image_job, Mapping)
+        else ""
+    )
+    if (
+        fallback_level >= 3
+        or image_variant == "pillow"
+        or image_status in {"failed", "diagnostic_fallback"}
+        or job_status in {"failed", "ambiguous_result", "diagnostic_fallback"}
+    ):
+        return False
+
+    recovery_status = str(
+        metadata.get("image_recovery_status")
+        or metadata.get("recovery_status")
+        or ""
+    ).strip().lower()
+    if recovery_status == "existing_output_reused":
+        diagnostic_history = " ".join(
+            str(metadata.get(key) or "")
+            for key in (
+                "last_error_summary",
+                "image_fallback_reason",
+                "prompt_fallback_reason",
+            )
+        ).lower()
+        if "本地诊断图" in diagnostic_history or "fallback=l3" in diagnostic_history:
+            return False
+    return True