From 5d4bdac501afe2353bbff0f51dffedc2e521d4b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E6=B8=85?= Date: Thu, 3 Sep 2026 21:59:59 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=87=E6=8D=A2=E7=9B=91=E6=B5=8B=20V2=20?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=8F=8A=E5=B7=B2=E5=8F=91=E5=B8=83=E5=9B=BE?= =?UTF-8?q?=E7=89=87=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 1 + crawler_client.py | 68 ++++++++++++----------------- ocr_dashboard_service.py | 56 ++++++++++++------------ tests/test_ocr_dashboard_service.py | 22 ++++++++-- 4 files changed, 75 insertions(+), 72 deletions(-) diff --git a/app.py b/app.py index 3702bdc..ca839f6 100644 --- a/app.py +++ b/app.py @@ -409,6 +409,7 @@ def create_app() -> Flask: request.args.get("collection_key", "").strip(), request.args.get("start_date", "").strip(), request.args.get("end_date", "").strip(), + request.args.get("subject_name", "").strip(), ), "request_id": g.request_id}) @app.route("/api/v1/data-dashboard/ocr/series", methods=["GET"]) diff --git a/crawler_client.py b/crawler_client.py index 16a1dc7..b142181 100644 --- a/crawler_client.py +++ b/crawler_client.py @@ -416,9 +416,10 @@ def delete_xapi_device(account_id: str) -> dict[str, Any]: def sync_dashboard_acceptance_project( batch_key: str, name: str, keywords: list[str], match_keywords: list[str], ) -> None: - """同步验收项目名称和问题配置到数据看板同一 batch_key。""" + """把验收项目配置同步到 V2；问题和匹配词均按主体隔离。""" import pymysql normalized_key = str(batch_key or "").strip() + subject_name = str(name or "").strip() or "未命名监测项目" if not normalized_key: return db = pymysql.connect( @@ -429,44 +430,33 @@ def sync_dashboard_acceptance_project( ) try: with db.cursor() as cursor: - cursor.execute( - "UPDATE xhs_ocr_monitor_batches SET subject_name=%s,updated_at=NOW() WHERE batch_key=%s", - (name, normalized_key), - ) - cursor.execute( - "UPDATE xhs_ocr_batch_questions SET active=0,updated_at=NOW() WHERE batch_key=%s", - (normalized_key,), - ) - for index, keyword in enumerate(keywords): - cursor.execute( - "SELECT question_code FROM xhs_ocr_questions WHERE search_keyword=%s LIMIT 1", - (keyword,), - ) - row = cursor.fetchone() - code = row["question_code"] if row else "X" + hashlib.sha256(keyword.encode()).hexdigest()[:15] - if not row: - cursor.execute( - """INSERT INTO xhs_ocr_questions - (question_code,search_keyword,question_direction,display_order,active,created_at,updated_at) - VALUES (%s,%s,%s,%s,1,NOW(),NOW())""", - (code, keyword, name, index + 1), - ) - cursor.execute( - """INSERT INTO xhs_ocr_batch_questions(batch_key,question_code,active,created_at,updated_at) - VALUES (%s,%s,1,NOW(),NOW()) ON DUPLICATE KEY UPDATE active=1,updated_at=NOW()""", - (normalized_key, code), - ) - for keyword in match_keywords: - keyword_key = hashlib.sha256(keyword.encode()).hexdigest() - department_key = hashlib.sha256(b"").hexdigest() - cursor.execute( - """INSERT INTO xhs_ocr_match_keywords - (keyword_key,department_key,keyword,department_name,case_sensitive, - active,remark,first_created_at,updated_at) - VALUES (%s,%s,%s,'',0,1,'验收项目同步',NOW(),NOW()) - ON DUPLICATE KEY UPDATE active=1,updated_at=NOW()""", - (keyword_key, department_key, keyword), - ) + cursor.execute("INSERT INTO monitor_subjects(name) VALUES(%s) ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id),status='active'",(subject_name,)) + subject_id = cursor.lastrowid + cursor.execute("SELECT id FROM monitor_subjects WHERE id=%s FOR UPDATE",(subject_id,)) + cursor.execute("SELECT id FROM monitor_batches WHERE legacy_batch_key=%s AND subject_id=%s AND status<>'archived' ORDER BY collected_at DESC LIMIT 1 FOR UPDATE",(normalized_key,subject_id)) + batch = cursor.fetchone() + if batch: + batch_id=batch['id'] + else: + collection_key='acceptance-'+hashlib.sha256((normalized_key+subject_name).encode()).hexdigest()[:24] + cursor.execute("INSERT INTO monitor_batches(subject_id,batch_key,legacy_batch_key,name,collected_at,status) VALUES(%s,%s,%s,%s,NOW(),'completed')",(subject_id,collection_key,normalized_key,subject_name)) + batch_id=cursor.lastrowid + cursor.execute("UPDATE monitor_batch_questions SET status='archived' WHERE batch_id=%s",(batch_id,)) + scopes=[] + for index, keyword in enumerate(dict.fromkeys(str(v).strip() for v in keywords if str(v).strip()),1): + cursor.execute("INSERT INTO monitor_questions(subject_id,question_text,display_order) VALUES(%s,%s,%s) ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id),display_order=VALUES(display_order),status='active'",(subject_id,keyword,index)) + question_id=cursor.lastrowid + cursor.execute("""INSERT INTO monitor_batch_questions(subject_id,batch_id,question_id,question_text_snapshot,display_order,status) + VALUES(%s,%s,%s,%s,%s,'active') ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id),question_text_snapshot=VALUES(question_text_snapshot),display_order=VALUES(display_order),status='active'""",(subject_id,batch_id,question_id,keyword,index)) + scopes.append(cursor.lastrowid) + normalized_matches=[str(v).strip() for v in match_keywords if str(v).strip()] + for scope_id in scopes: + cursor.execute("UPDATE monitor_batch_question_keywords SET status='archived' WHERE batch_question_id=%s",(scope_id,)) + for order,keyword in enumerate(dict.fromkeys(normalized_matches),1): + cursor.execute("INSERT INTO monitor_keywords(subject_id,keyword_text) VALUES(%s,%s) ON DUPLICATE KEY UPDATE id=LAST_INSERT_ID(id),status='active'",(subject_id,keyword)) + keyword_id=cursor.lastrowid + cursor.execute("""INSERT INTO monitor_batch_question_keywords(subject_id,batch_question_id,keyword_id,keyword_text_snapshot,match_method,display_order,status) + VALUES(%s,%s,%s,%s,'substring',%s,'active') ON DUPLICATE KEY UPDATE keyword_text_snapshot=VALUES(keyword_text_snapshot),match_method='substring',display_order=VALUES(display_order),status='active'""",(subject_id,scope_id,keyword_id,keyword,order)) db.commit() except Exception: db.rollback() diff --git a/ocr_dashboard_service.py b/ocr_dashboard_service.py index cf18c83..e5fdd8d 100644 --- a/ocr_dashboard_service.py +++ b/ocr_dashboard_service.py @@ -60,10 +60,10 @@ def list_collections_by_subject(subject_name: str, start_date: str = "", end_dat def get_summary(batch_key: str = "", collection_key: str = "", - start_date: str = "", end_date: str = "") -> dict[str, Any]: + start_date: str = "", end_date: str = "", subject_name: str = "") -> dict[str, Any]: data = _get("/xhs/ocr/dashboard/summary", { "batch_key": batch_key, "collection_key": collection_key, - "start_date": start_date, "end_date": end_date, + "start_date": start_date, "end_date": end_date, "subject_name": subject_name, }) or {} return _overlay_acceptance_questions(data, batch_key) @@ -176,36 +176,34 @@ def get_result_matches(result_id: int) -> dict[str, Any]: def list_evidences(params: dict[str, Any]) -> dict[str, Any]: allowed = {"batch_key", "collection_key", "start_date", "end_date", "question_code", "evidence_type", "limit"} query = {key: value for key, value in params.items() if key in allowed} - if not (str(query.get("batch_key") or "").strip() or str(query.get("collection_key") or "").strip() - or str(query.get("start_date") or "").strip() or str(query.get("end_date") or "").strip()): - raise TaskError("OCR_EVIDENCE_SCOPE_REQUIRED", "batch_key/collection_key 或 start_date/end_date 至少传入一种", 422) - limit = int(query.get("limit") or 10) + if not any(str(query.get(key) or "").strip() for key in + ("batch_key", "collection_key", "start_date", "end_date")): + raise TaskError("OCR_EVIDENCE_SCOPE_REQUIRED", "请指定采集批次或日期范围", 422) + try: + limit = int(query.get("limit") or 10) + except (ValueError, TypeError) as exc: + raise TaskError("OCR_EVIDENCE_LIMIT_INVALID", "limit 必须在 1 到 100 之间", 422) from exc if not 1 <= limit <= 100: raise TaskError("OCR_EVIDENCE_LIMIT_INVALID", "limit 必须在 1 到 100 之间", 422) query["limit"] = limit - data = _get("/xhs/ocr/dashboard/evidences", query) or { - "items": [], "total": 0, "limit": limit, "is_complete": False, - } - if Config.EVIDENCE_ADMIN_BASE_URL and query.get("batch_key") and query.get("question_code"): - try: - response = SESSION.get( - f"{Config.EVIDENCE_ADMIN_BASE_URL}/api/evidences", - params={ - "batch_key": query["batch_key"], - "collection_key": query.get("collection_key", ""), - "question_code": query["question_code"], - }, - timeout=min(Config.CRAWLER_TIMEOUT_SECONDS, 10), - ) - response.raise_for_status() - manual_items = (response.json().get("data") or []) - data["items"] = manual_items - data["total"] = len(manual_items) - data["limit"] = limit - data["is_complete"] = True - except (requests.RequestException, ValueError): - pass - return data + if not Config.EVIDENCE_ADMIN_BASE_URL: + raise TaskError("OCR_PUBLISHED_EVIDENCE_UNAVAILABLE", "图片展示服务未配置", 503) + try: + response = SESSION.get( + f"{Config.EVIDENCE_ADMIN_BASE_URL}/api/published-evidences", + params=query, timeout=min(Config.CRAWLER_TIMEOUT_SECONDS, 10), + ) + response.raise_for_status() + payload = response.json() + data = payload.get("data") + if payload.get("error") or not isinstance(data, dict) or not isinstance(data.get("items"), list): + raise ValueError("Invalid published image response") + # Fail closed if an old or incompatible admin server returns unapproved images. + if any(item.get("is_published") is not True for item in data["items"]): + raise ValueError("Unpublished image returned by admin") + return data + except (requests.RequestException, ValueError, TypeError, AttributeError) as exc: + raise TaskError("OCR_PUBLISHED_EVIDENCE_UNAVAILABLE", "已发布图片暂时无法读取，请稍后重试", 502) from exc def list_configured_notes(params: dict[str, Any] | None = None) -> list[dict[str, Any]]: diff --git a/tests/test_ocr_dashboard_service.py b/tests/test_ocr_dashboard_service.py index af29856..2f894f7 100644 --- a/tests/test_ocr_dashboard_service.py +++ b/tests/test_ocr_dashboard_service.py @@ -12,10 +12,10 @@ class OcrDashboardServiceTest(unittest.TestCase): def test_evidences_use_exact_admin_scope_as_display_source(self, get, upstream_get): upstream_get.return_value = {"items": [{"cos_object_key": "old.jpg"}], "total": 1} response = Mock() - response.json.return_value = {"data": [{ + response.json.return_value = {"data": {"items": [{ "id": 2, "image_id": 3, "evidence_type": "first_screen", - "cos_object_key": "managed.jpg", "sort_order": 1, - }]} + "cos_object_key": "managed.jpg", "sort_order": 1, "is_published": True, + }], "total": 1, "limit": 100, "is_complete": True}} get.return_value = response result = list_evidences({ @@ -25,8 +25,10 @@ class OcrDashboardServiceTest(unittest.TestCase): self.assertEqual([item["cos_object_key"] for item in result["items"]], ["managed.jpg"]) self.assertEqual(result["total"], 1) self.assertEqual(get.call_args.kwargs["params"], { - "batch_key": "b1", "collection_key": "c1", "question_code": "Q1", + "batch_key": "b1", "collection_key": "c1", "question_code": "Q1", "limit": 100, }) + self.assertTrue(get.call_args.args[0].endswith("/api/published-evidences")) + upstream_get.assert_not_called() @patch("ocr_dashboard_service.SESSION.get") def test_configured_notes_reads_admin_list(self, get): @@ -52,6 +54,18 @@ class OcrDashboardServiceTest(unittest.TestCase): get_summary("b1", "c1") self.assertEqual(get.call_args.kwargs["params"], {"batch_key": "b1", "collection_key": "c1"}) + @patch("ocr_dashboard_service.SESSION.get") + def test_summary_forwards_subject_name(self, get): + response = Mock() + response.json.return_value = {"success": True, "code": 200, "data": {"questions": []}} + get.return_value = response + + get_summary(start_date="2026-08-01", end_date="2026-09-03", subject_name="民生银行") + + self.assertEqual(get.call_args.kwargs["params"], { + "start_date": "2026-08-01", "end_date": "2026-09-03", "subject_name": "民生银行", + }) + @patch("ocr_dashboard_service.SESSION.get") def test_comparison_forwards_scope(self, get): response = Mock()