看板问题列表同步验收项目配置

ocr_dashboard_service.py

@@ -1,5 +1,7 @@
 import io
+import hashlib
 import re
+import sqlite3
 import zipfile
 from pathlib import Path
 from typing import Any
@@ -8,6 +10,7 @@ from urllib.parse import urlparse
 import requests
 
 from config import Config
+from database import connection
 from task_service import TaskError
 
 
@@ -48,10 +51,62 @@ def list_collections(batch_key: str) -> list[dict[str, Any]]:
 
 def get_summary(batch_key: str = "", collection_key: str = "",
                  start_date: str = "", end_date: str = "") -> dict[str, Any]:
-    return _get("/xhs/ocr/dashboard/summary", {
+    data = _get("/xhs/ocr/dashboard/summary", {
         "batch_key": batch_key, "collection_key": collection_key,
         "start_date": start_date, "end_date": end_date,
     }) or {}
+    return _overlay_acceptance_questions(data, batch_key)
+
+
+def _overlay_acceptance_questions(data: dict[str, Any], batch_key: str) -> dict[str, Any]:
+    """让 Holy蟹看板立即采用同 batch_key 的验收项目当前问题配置。"""
+    normalized_key = str(batch_key or "").strip()
+    if not normalized_key:
+        return data
+    try:
+        with connection() as conn:
+            task = conn.execute(
+                """SELECT id,name FROM acceptance_tasks
+                   WHERE dashboard_batch_key=? AND scheduled_from_task_id IS NULL
+                   ORDER BY id LIMIT 1""",
+                (normalized_key,),
+            ).fetchone()
+            if not task:
+                return data
+            keywords = [
+                row["keyword"] for row in conn.execute(
+                    "SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order",
+                    (task["id"],),
+                ).fetchall()
+            ]
+    except sqlite3.OperationalError:
+        return data
+    if not keywords:
+        return data
+    current = {
+        str(item.get("search_keyword") or ""): item
+        for item in data.get("questions", [])
+    }
+    questions = []
+    for index, keyword in enumerate(keywords):
+        item = dict(current.get(keyword, {}))
+        item.update({
+            "question_code": item.get("question_code") or "X" + hashlib.sha256(keyword.encode()).hexdigest()[:15],
+            "search_keyword": keyword,
+            "question_direction": item.get("question_direction") or task["name"],
+            "display_order": index + 1,
+        })
+        for field in ("result_count", "device_count", "valid_result_count", "minsheng_matched_count", "cobranded_matched_count"):
+            item[field] = int(item.get(field) or 0)
+        item.setdefault("minsheng_match_rate", 0)
+        item.setdefault("cobranded_match_rate", 0)
+        item.setdefault("match_keywords", [])
+        questions.append(item)
+    data["questions"] = questions
+    data["question_pass_count"] = sum(
+        1 for item in questions if float(item.get("minsheng_match_rate") or 0) >= 0.5
+    )
+    return data
 
 
 def get_series(batch_key: str = "", start_date: str = "", end_date: str = "") -> dict[str, Any]:

tests/test_ocr_dashboard_overlay.py

@@ -0,0 +1,48 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from config import Config
+from database import connection, init_database, now_iso
+from ocr_dashboard_service import _overlay_acceptance_questions
+
+
+class OcrDashboardOverlayTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.original_database_path = Config.DATABASE_PATH
+        Config.DATABASE_PATH = Path(self.temp_dir.name) / "holy.db"
+        init_database()
+        now = now_iso()
+        with connection() as conn:
+            task_id = conn.execute(
+                """INSERT INTO acceptance_tasks
+                   (public_task_id,name,acceptance_method,status,progress_percent,device_count,
+                    created_by_id,created_by_name,created_at,updated_at,dashboard_batch_key)
+                   VALUES ('root','民生银行','combined','completed',100,1,'u','u',?,?,?)""",
+                (now, now, "batch-1"),
+            ).lastrowid
+            conn.executemany(
+                "INSERT INTO acceptance_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)",
+                [(task_id, 0, "修改后的第一题", now), (task_id, 1, "第二题", now)],
+            )
+
+    def tearDown(self):
+        Config.DATABASE_PATH = self.original_database_path
+        self.temp_dir.cleanup()
+
+    def test_current_acceptance_questions_replace_historical_list(self):
+        data = {"questions": [
+            {"question_code": "Q1", "search_keyword": "旧的第一题", "result_count": 10},
+            {"question_code": "Q2", "search_keyword": "第二题", "result_count": 8},
+        ]}
+
+        result = _overlay_acceptance_questions(data, "batch-1")
+
+        self.assertEqual([item["search_keyword"] for item in result["questions"]], ["修改后的第一题", "第二题"])
+        self.assertEqual(result["questions"][0]["result_count"], 0)
+        self.assertEqual(result["questions"][1]["result_count"], 8)
+
+
+if __name__ == "__main__":
+    unittest.main()