From e2b48fbdbbf131c18323aab4f3886c87fc6dcdb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E6=B8=85?= Date: Tue, 1 Sep 2026 22:44:13 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E9=AA=8C=E6=94=B6=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE=E4=B8=8E=E6=95=B0=E6=8D=AE=E7=9C=8B=E6=9D=BF=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- acceptance_service.py | 23 +++++++++++ apifox-business-tasks.openapi.json | 10 +++++ app.py | 8 ++++ crawler_client.py | 62 ++++++++++++++++++++++++++++++ 4 files changed, 103 insertions(+) diff --git a/acceptance_service.py b/acceptance_service.py index a97a546..2763783 100644 --- a/acceptance_service.py +++ b/acceptance_service.py @@ -31,6 +31,7 @@ from crawler_client import ( query_collection, query_xapi_batch_runs, schedule_xapi_screenshots, + sync_dashboard_acceptance_project, ) from database import connection, now_iso, upsert_feishu_note from task_service import TaskError @@ -1720,6 +1721,28 @@ def remove_acceptance_device(account_id: str) -> dict[str, Any]: return result +def update_acceptance_task(public_id: str, data: dict[str, Any]) -> dict[str, Any]: + name = str(data.get("name") or "").strip() + keywords = list(dict.fromkeys(str(item).strip() for item in data.get("keywords", []) if str(item).strip())) + matches = list(dict.fromkeys(str(item).strip() for item in data.get("match_keywords", []) if str(item).strip())) + if not name or not keywords or not matches: + raise TaskError("INVALID_ACCEPTANCE_PROJECT", "项目名称、搜索关键词和匹配关键词不能为空", 422) + with connection() as conn: + task = conn.execute("SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone() + if not task: + raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404) + root_id = task["scheduled_from_task_id"] or task["id"] + root = conn.execute("SELECT * FROM acceptance_tasks WHERE id=?", (root_id,)).fetchone() + conn.execute("UPDATE acceptance_tasks SET name=?,updated_at=? WHERE id=?", (name, now_iso(), root_id)) + conn.execute("DELETE FROM acceptance_keywords WHERE task_id=?", (root_id,)) + conn.execute("DELETE FROM acceptance_match_keywords WHERE task_id=?", (root_id,)) + conn.executemany("INSERT INTO acceptance_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", [(root_id, i, value, now_iso()) for i, value in enumerate(keywords)]) + conn.executemany("INSERT INTO acceptance_match_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", [(root_id, i, value, now_iso()) for i, value in enumerate(matches)]) + dashboard_key = root["dashboard_batch_key"] or root["public_task_id"] + sync_dashboard_acceptance_project(dashboard_key, name, keywords, matches) + return get_acceptance_task(root["public_task_id"]) + + 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 38ea10d..07f1de8 100644 --- a/apifox-business-tasks.openapi.json +++ b/apifox-business-tasks.openapi.json @@ -55,6 +55,16 @@ "responses": {"201": {"description": "新一轮验收任务已创建"}} } }, + "/api/v1/acceptance-tasks/{task_id}": { + "patch": { + "summary": "修改验收项目配置", + "description": "修改根项目名称、搜索关键词和匹配关键词，并同步更新相同 dashboard_batch_key 的数据看板主体和问题选项。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "requestBody": {"required": true, "content": {"application/json": {"schema": {"type": "object", "required": ["name", "keywords", "match_keywords"], "properties": {"name": {"type": "string"}, "keywords": {"type": "array", "items": {"type": "string"}}, "match_keywords": {"type": "array", "items": {"type": "string"}}}}}}}, + "responses": {"200": {"description": "项目与数据看板配置同步成功"}} + } + }, "/api/v1/acceptance-tasks/{task_id}/cancel": { "post": { "summary": "停止项目验收采集", diff --git a/app.py b/app.py index 1399cfd..52367c2 100644 --- a/app.py +++ b/app.py @@ -45,6 +45,7 @@ from acceptance_service import ( start_acceptance_schedule_worker, stop_acceptance_schedule, update_acceptance_schedule, + update_acceptance_task, ) from rule_service import ( delete_rule_document, @@ -318,6 +319,13 @@ def create_app() -> Flask: def get_project_acceptance_task(task_id: str): return jsonify({"data": get_acceptance_task(task_id), "request_id": g.request_id}) + @app.route("/api/v1/acceptance-tasks/", methods=["PATCH"]) + def update_project_acceptance_task(task_id: str): + return jsonify({ + "data": update_acceptance_task(task_id, request.get_json(silent=True) or {}), + "request_id": g.request_id, + }) + @app.route("/api/v1/acceptance-tasks//notes", methods=["GET"]) def get_project_acceptance_notes(task_id: str): return jsonify({"data": list_acceptance_notes(task_id), "request_id": g.request_id}) diff --git a/crawler_client.py b/crawler_client.py index 0da7086..255a469 100644 --- a/crawler_client.py +++ b/crawler_client.py @@ -413,6 +413,68 @@ def delete_xapi_device(account_id: str) -> dict[str, Any]: return {"xhs_account_id": normalized, "active": False} +def sync_dashboard_acceptance_project( + batch_key: str, name: str, keywords: list[str], match_keywords: list[str], +) -> None: + """同步验收项目名称和问题配置到数据看板同一 batch_key。""" + import pymysql + normalized_key = str(batch_key or "").strip() + if not normalized_key: + return + 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=False, + ) + 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), + ) + db.commit() + except Exception: + db.rollback() + raise + finally: + db.close() + + def query_xapi_batch_runs(batch_id: str, *, refresh: bool = True) -> dict[str, Any]: """读取某个 Framework 调度批次下的设备任务、截图与飞书结果。""" normalized = str(batch_id or "").strip()