支持验收项目与数据看板配置同步

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()