diff --git a/acceptance_service.py b/acceptance_service.py index 2cd27b0..a97a546 100644 --- a/acceptance_service.py +++ b/acceptance_service.py @@ -22,6 +22,8 @@ from playwright.sync_api import sync_playwright from config import Config from crawler_client import ( collect_keyword_sources, + create_xapi_device, + delete_xapi_device, fetch_content_detail, get_wen_resources, list_xhs_wen_xapi_devices, @@ -1162,13 +1164,16 @@ def _copy_acceptance_task(public_id: str, *, trigger_source: str = "acceptance") WHERE task_id=? AND secret_type='custom_cookie'""", (task["id"],), ).fetchone() + current_device_count = task["device_count"] + if task["acceptance_method"] in {"screenshot", "combined"} and task["use_all_devices"]: + current_device_count = max(1, int(get_wen_resources().get("device_count") or 0)) child = create_acceptance_task({ "name": task["name"], "acceptance_method": task["acceptance_method"], "feishu_links": links, "keywords": keywords, "match_keywords": match_keywords, - "device_count": task["device_count"], + "device_count": current_device_count, "created_by_id": task["created_by_id"], "created_by_name": task["created_by_name"], "custom_cookie": secret["secret_value"] if secret else None, @@ -1176,8 +1181,9 @@ def _copy_acceptance_task(public_id: str, *, trigger_source: str = "acceptance") }) with connection() as conn: conn.execute( - "UPDATE acceptance_tasks SET scheduled_from_task_id=? WHERE public_task_id=?", - (task["id"], child["task_id"]), + """UPDATE acceptance_tasks SET scheduled_from_task_id=?,dashboard_batch_key=?, + use_all_devices=? WHERE public_task_id=?""", + (task["id"], task["dashboard_batch_key"], task["use_all_devices"], child["task_id"]), ) return get_acceptance_task(child["task_id"]) @@ -1663,6 +1669,8 @@ def _task_dict(conn, task) -> dict[str, Any]: "acceptance_method": task["acceptance_method"], "status": task["status"], "progress_percent": task["progress_percent"], "device_count": task["device_count"], "cookie_count": task["cookie_count"], + "dashboard_batch_key": task["dashboard_batch_key"], + "use_all_devices": bool(task["use_all_devices"]), "schedule_enabled": bool(task["schedule_enabled"]), "schedule_weekday": task["schedule_weekday"], "schedule_weekdays": _decode_json(task["schedule_weekdays"], None) or ( @@ -1700,6 +1708,18 @@ def list_acceptance_tasks() -> dict[str, Any]: return {"items": items, "total": len(items)} +def create_acceptance_device(data: dict[str, Any]) -> dict[str, Any]: + result = create_xapi_device(data) + result["resources"] = get_acceptance_resources() + return result + + +def remove_acceptance_device(account_id: str) -> dict[str, Any]: + result = delete_xapi_device(account_id) + result["resources"] = get_acceptance_resources() + return result + + def list_acceptance_notes(public_id: str) -> dict[str, Any]: with connection() as conn: task = conn.execute("SELECT id FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone() diff --git a/apifox-business-tasks.openapi.json b/apifox-business-tasks.openapi.json index 54dc773..38ea10d 100644 --- a/apifox-business-tasks.openapi.json +++ b/apifox-business-tasks.openapi.json @@ -28,6 +28,24 @@ "responses": {"201": {"description": "验收任务创建成功"}} } }, + "/api/v1/acceptance-devices": { + "post": { + "summary": "添加验收截图设备", + "description": "向 framework-db 设备库新增或重新启用设备，后续周期下发读取最新启用设备。", + "tags": ["项目验收"], + "requestBody": {"required": true, "content": {"application/json": {"schema": {"type": "object", "required": ["xhs_account_id"], "properties": {"xhs_account_id": {"type": "string"}, "account_name": {"type": "string"}, "device_code": {"type": "string"}, "phone": {"type": "string"}}}}}}, + "responses": {"201": {"description": "设备添加成功"}} + } + }, + "/api/v1/acceptance-devices/{account_id}": { + "delete": { + "summary": "删除验收截图设备", + "description": "将设备标记为停用；历史采集数据保留，后续下发不再选择。", + "tags": ["项目验收"], + "parameters": [{"name": "account_id", "in": "path", "required": true, "schema": {"type": "string"}}], + "responses": {"200": {"description": "设备已停用"}, "404": {"description": "设备不存在"}} + } + }, "/api/v1/acceptance-tasks/{task_id}/collect-now": { "post": { "summary": "立即采集并下发指令", diff --git a/app.py b/app.py index 05fc67a..1399cfd 100644 --- a/app.py +++ b/app.py @@ -24,11 +24,13 @@ from ocr_dashboard_service import ( list_batches as list_ocr_batches, list_collections as list_ocr_collections, list_evidences as list_ocr_evidences, + list_configured_notes as list_ocr_configured_notes, export_evidences as export_ocr_evidences, list_results as list_ocr_results, ) from acceptance_service import ( cancel_acceptance_task, + create_acceptance_device, create_acceptance_task, collect_acceptance_now, get_acceptance_resources, @@ -38,6 +40,7 @@ from acceptance_service import ( list_acceptance_tasks, resume_acceptance_task, retry_failed_acceptance_devices, + remove_acceptance_device, resume_unfinished_acceptance_tasks, start_acceptance_schedule_worker, stop_acceptance_schedule, @@ -293,6 +296,20 @@ def create_app() -> Flask: def query_project_acceptance_resources(): return jsonify({"data": get_acceptance_resources(), "request_id": g.request_id}) + @app.route("/api/v1/acceptance-devices", methods=["POST"]) + def create_project_acceptance_device(): + return jsonify({ + "data": create_acceptance_device(request.get_json(silent=True) or {}), + "request_id": g.request_id, + }), 201 + + @app.route("/api/v1/acceptance-devices/", methods=["DELETE"]) + def delete_project_acceptance_device(account_id: str): + return jsonify({ + "data": remove_acceptance_device(account_id), + "request_id": g.request_id, + }) + @app.route("/api/v1/acceptance-tasks", methods=["GET"]) def query_project_acceptance_tasks(): return jsonify({"data": list_acceptance_tasks(), "request_id": g.request_id}) @@ -398,6 +415,10 @@ def create_app() -> Flask: def query_ocr_dashboard_evidences(): return jsonify({"data": list_ocr_evidences(request.args.to_dict()), "request_id": g.request_id}) + @app.route("/api/v1/data-dashboard/ocr/configured-notes", methods=["GET"]) + def query_ocr_configured_notes(): + return jsonify({"data": list_ocr_configured_notes(), "request_id": g.request_id}) + @app.route("/api/v1/data-dashboard/ocr/evidences/export", methods=["GET"]) def export_ocr_dashboard_evidences(): payload, filename, count = export_ocr_evidences(request.args.to_dict()) diff --git a/crawler_client.py b/crawler_client.py index 242a1f3..0da7086 100644 --- a/crawler_client.py +++ b/crawler_client.py @@ -5,6 +5,7 @@ from typing import Any import requests import json +import hashlib from config import Config @@ -352,6 +353,66 @@ def schedule_xapi_screenshots( ) +def create_xapi_device(data: dict[str, Any]) -> dict[str, Any]: + """在 framework-db 中新增或重新启用人工维护的 XAPI 设备。""" + import pymysql + account_id = str(data.get("xhs_account_id") or "").strip() + if not account_id: + raise CrawlerError("xhs_account_id 不能为空") + now = time.strftime("%Y-%m-%d %H:%M:%S") + record_id = "manual-" + hashlib.sha256(account_id.encode()).hexdigest()[:24] + db = pymysql.connect( + host=Config.CRAWLER_DATABASE_HOST, port=Config.CRAWLER_DATABASE_PORT, + user=Config.CRAWLER_DATABASE_USER, password=Config.CRAWLER_DATABASE_PASSWORD, + database=Config.CRAWLER_DATABASE_NAME, charset="utf8mb4", + cursorclass=pymysql.cursors.DictCursor, autocommit=True, + ) + try: + with db.cursor() as cursor: + cursor.execute( + """INSERT INTO xhs_device_accounts + (feishu_record_id,base_token,table_id,view_id,xhs_account_id, + account_name,phone,device_code,text_value,active,raw_json, + first_created_at,last_synced_at,updated_at) + VALUES (%s,'manual','manual',NULL,%s,%s,%s,%s,'人工添加',1,%s,%s,%s,%s) + ON DUPLICATE KEY UPDATE account_name=VALUES(account_name), + phone=VALUES(phone),device_code=VALUES(device_code),active=1, + raw_json=VALUES(raw_json),last_synced_at=VALUES(last_synced_at), + updated_at=VALUES(updated_at)""", + (record_id, account_id, str(data.get("account_name") or "").strip() or None, + str(data.get("phone") or "").strip() or None, + str(data.get("device_code") or "").strip() or None, + json.dumps(data, ensure_ascii=False), now, now, now), + ) + finally: + db.close() + return {"xhs_account_id": account_id, "active": True} + + +def delete_xapi_device(account_id: str) -> dict[str, Any]: + """软删除设备；历史任务保留，后续下发不再选择该设备。""" + import pymysql + normalized = str(account_id or "").strip() + if not normalized: + raise CrawlerError("xhs_account_id 不能为空") + db = pymysql.connect( + host=Config.CRAWLER_DATABASE_HOST, port=Config.CRAWLER_DATABASE_PORT, + user=Config.CRAWLER_DATABASE_USER, password=Config.CRAWLER_DATABASE_PASSWORD, + database=Config.CRAWLER_DATABASE_NAME, charset="utf8mb4", autocommit=True, + ) + try: + with db.cursor() as cursor: + cursor.execute( + "UPDATE xhs_device_accounts SET active=0,updated_at=NOW() WHERE xhs_account_id=%s AND active=1", + (normalized,), + ) + if not cursor.rowcount: + raise CrawlerError("设备不存在或已经删除") + finally: + db.close() + return {"xhs_account_id": normalized, "active": False} + + def query_xapi_batch_runs(batch_id: str, *, refresh: bool = True) -> dict[str, Any]: """读取某个 Framework 调度批次下的设备任务、截图与飞书结果。""" normalized = str(batch_id or "").strip() diff --git a/database.py b/database.py index f46a0c8..fa98ee3 100644 --- a/database.py +++ b/database.py @@ -386,6 +386,8 @@ def init_database() -> None: "schedule_error": "TEXT", "trigger_source": "TEXT NOT NULL DEFAULT 'acceptance'", "cookie_count": "INTEGER NOT NULL DEFAULT 0", + "dashboard_batch_key": "TEXT", + "use_all_devices": "INTEGER NOT NULL DEFAULT 0", } for column, definition in acceptance_migrations.items(): if column not in acceptance_columns: diff --git a/ocr_dashboard_service.py b/ocr_dashboard_service.py index 511ca37..0ea6a79 100644 --- a/ocr_dashboard_service.py +++ b/ocr_dashboard_service.py @@ -152,6 +152,24 @@ def list_evidences(params: dict[str, Any]) -> dict[str, Any]: return data +def list_configured_notes() -> list[dict[str, Any]]: + """读取凭证后台中由用户配置的多篇展示笔记。""" + if not Config.EVIDENCE_ADMIN_BASE_URL: + return [] + try: + response = SESSION.get( + f"{Config.EVIDENCE_ADMIN_BASE_URL}/api/notes", + timeout=min(Config.CRAWLER_TIMEOUT_SECONDS, 15), + ) + response.raise_for_status() + payload = response.json() + except (requests.RequestException, ValueError) as exc: + raise TaskError("OCR_CONFIGURED_NOTES_UNAVAILABLE", "配置笔记服务暂时不可用", 502) from exc + if payload.get("error"): + raise TaskError("OCR_CONFIGURED_NOTES_ERROR", str(payload["error"].get("message") or "配置笔记读取失败"), 502) + return payload.get("data") or [] + + def export_evidences(params: dict[str, Any]) -> tuple[io.BytesIO, str, int]: query = dict(params) query["limit"] = 100 diff --git a/tests/test_ocr_dashboard_service.py b/tests/test_ocr_dashboard_service.py index 69ab3a2..4f4ffd3 100644 --- a/tests/test_ocr_dashboard_service.py +++ b/tests/test_ocr_dashboard_service.py @@ -1,11 +1,20 @@ import unittest from unittest.mock import Mock, patch -from ocr_dashboard_service import export_evidences, get_comparison, get_summary, list_batches, list_results +from ocr_dashboard_service import export_evidences, get_comparison, get_summary, list_batches, list_configured_notes, list_results from task_service import TaskError class OcrDashboardServiceTest(unittest.TestCase): + @patch("ocr_dashboard_service.SESSION.get") + def test_configured_notes_reads_admin_list(self, get): + response = Mock() + response.json.return_value = {"data": [{"note_id": "note-1"}]} + get.return_value = response + + self.assertEqual(list_configured_notes(), [{"note_id": "note-1"}]) + self.assertTrue(get.call_args.args[0].endswith("/api/notes")) + @patch("ocr_dashboard_service.SESSION.get") def test_list_batches_unwraps_framework_response(self, get): response = Mock()