diff --git a/ocr_dashboard_service.py b/ocr_dashboard_service.py index 0ea6a79..33f56a8 100644 --- a/ocr_dashboard_service.py +++ b/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]: diff --git a/tests/test_ocr_dashboard_overlay.py b/tests/test_ocr_dashboard_overlay.py new file mode 100644 index 0000000..95cf05b --- /dev/null +++ b/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()