From fd083e0438df2efe981bf515540bdd7ed0157d84 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E6=B8=85?= Date: Tue, 1 Sep 2026 23:00:55 +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=E7=8B=AC=E7=AB=8B=E5=90=AF=E5=8A=A8=E7=88=AC=E8=99=AB?= =?UTF-8?q?=E9=87=87=E9=9B=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- acceptance_service.py | 31 +++- apifox-business-tasks.openapi.json | 9 + app.py | 8 + tests/test_acceptance_creation.py | 262 +++++++++++++++++++++++++++++ 4 files changed, 309 insertions(+), 1 deletion(-) create mode 100644 tests/test_acceptance_creation.py diff --git a/acceptance_service.py b/acceptance_service.py index 2763783..3beedf0 100644 --- a/acceptance_service.py +++ b/acceptance_service.py @@ -950,7 +950,7 @@ def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]: raise TaskError("KEYWORD_LIMIT_EXCEEDED", "单次最多支持 20 个搜索词", 422) if method in {"screenshot", "combined"} and not keywords: raise TaskError("KEYWORD_REQUIRED", "截图取证请至少输入一个搜索词", 422) - if method in {"data-analysis", "combined"} and not links: + if method in {"data-analysis", "combined"} and not links and not keywords: raise TaskError("FEISHU_LINK_REQUIRED", "笔记收录请至少输入一个公开飞书链接", 422) resources = { "cookie_count": max(0, int(raw_resources.get("cookie_count") or 0)), @@ -1194,6 +1194,35 @@ def collect_acceptance_now(public_id: str) -> dict[str, Any]: return _copy_acceptance_task(public_id, trigger_source="acceptance") +def collect_acceptance_crawler_now(public_id: str) -> dict[str, Any]: + """按项目搜索词启动独立的 Cookie 爬虫采集批次，不下发截图设备。""" + 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() + keywords = [row["keyword"] for row in conn.execute("SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (root_id,)).fetchall()] + matches = [row["keyword"] for row in conn.execute("SELECT keyword FROM acceptance_match_keywords WHERE task_id=? ORDER BY keyword_order", (root_id,)).fetchall()] + child = create_acceptance_task({ + "name": f"{root['name']}｜爬虫数据采集", + "acceptance_method": "data-analysis", + "feishu_links": [], + "keywords": keywords, + "match_keywords": matches, + "created_by_id": root["created_by_id"], + "created_by_name": root["created_by_name"], + "trigger_source": "acceptance", + }) + with connection() as conn: + conn.execute( + """UPDATE acceptance_tasks SET scheduled_from_task_id=?,dashboard_batch_key=? + WHERE public_task_id=?""", + (root_id, root["dashboard_batch_key"], child["task_id"]), + ) + return get_acceptance_task(child["task_id"]) + + def run_due_acceptance_schedules() -> int: current = datetime.now().astimezone() now = current.isoformat(timespec="seconds") diff --git a/apifox-business-tasks.openapi.json b/apifox-business-tasks.openapi.json index 07f1de8..8fae6d9 100644 --- a/apifox-business-tasks.openapi.json +++ b/apifox-business-tasks.openapi.json @@ -55,6 +55,15 @@ "responses": {"201": {"description": "新一轮验收任务已创建"}} } }, + "/api/v1/acceptance-tasks/{task_id}/collect-crawler-now": { + "post": { + "summary": "立即启动验收爬虫数据采集", + "description": "使用项目的搜索词和当前全部可用 Cookie 创建独立数据采集批次，不下发截图设备。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": {"201": {"description": "爬虫采集批次已启动"}} + } + }, "/api/v1/acceptance-tasks/{task_id}": { "patch": { "summary": "修改验收项目配置", diff --git a/app.py b/app.py index 52367c2..c6e6987 100644 --- a/app.py +++ b/app.py @@ -33,6 +33,7 @@ from acceptance_service import ( create_acceptance_device, create_acceptance_task, collect_acceptance_now, + collect_acceptance_crawler_now, get_acceptance_resources, get_acceptance_link_ai_data, get_acceptance_task, @@ -353,6 +354,13 @@ def create_app() -> Flask: def collect_project_acceptance_now(task_id: str): return jsonify({"data": collect_acceptance_now(task_id), "request_id": g.request_id}), 201 + @app.route("/api/v1/acceptance-tasks//collect-crawler-now", methods=["POST"]) + def collect_project_acceptance_crawler_now(task_id: str): + return jsonify({ + "data": collect_acceptance_crawler_now(task_id), + "request_id": g.request_id, + }), 201 + @app.route("/api/v1/acceptance-tasks//retry-failed-devices", methods=["POST"]) def retry_project_acceptance_failed_devices(task_id: str): return jsonify({ diff --git a/tests/test_acceptance_creation.py b/tests/test_acceptance_creation.py new file mode 100644 index 0000000..6d28f11 --- /dev/null +++ b/tests/test_acceptance_creation.py @@ -0,0 +1,262 @@ +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from acceptance_service import ( + XHS_URL_PATTERN, + cancel_acceptance_task, + collect_acceptance_now, + collect_acceptance_crawler_now, + create_acceptance_task, + get_acceptance_task, + run_due_acceptance_schedules, + retry_failed_acceptance_devices, + stop_acceptance_schedule, + update_acceptance_schedule, +) +from config import Config +from database import connection, init_database, now_iso + + +class _ThreadWithoutStart: + def __init__(self, *args, **kwargs): + pass + + def start(self): + pass + + +class AcceptanceCreationTest(unittest.TestCase): + def test_xhs_short_link_cn_is_recognized(self): + text = "笔记 https://xhslink.cn/o/example123 进入小红书" + self.assertEqual(XHS_URL_PATTERN.findall(text), ["https://xhslink.cn/o/example123"]) + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.original_database_path = Config.DATABASE_PATH + self.original_file_storage_path = Config.FILE_STORAGE_PATH + root = Path(self.temp_dir.name) + Config.DATABASE_PATH = root / "holy.db" + Config.FILE_STORAGE_PATH = root / "files" + init_database() + + def tearDown(self): + Config.DATABASE_PATH = self.original_database_path + Config.FILE_STORAGE_PATH = self.original_file_storage_path + self.temp_dir.cleanup() + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.get_wen_resources") + def test_screenshot_task_requires_keywords_but_not_feishu_links(self, resources): + resources.return_value = {"cookie_count": 0, "device_count": 2} + + task = create_acceptance_task({ + "acceptance_method": "screenshot", + "feishu_links": [], + "keywords": ["山姆特价"], + "device_count": 1, + }) + + self.assertEqual(task["acceptance_method"], "screenshot") + self.assertEqual(task["keywords"], ["山姆特价"]) + self.assertEqual(task["feishu_links"], []) + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.get_wen_resources") + def test_active_acceptance_task_can_be_cancelled(self, resources): + resources.return_value = {"cookie_count": 0, "device_count": 2} + task = create_acceptance_task({ + "acceptance_method": "screenshot", + "keywords": ["山姆特价"], + "device_count": 1, + }) + update_acceptance_schedule(task["task_id"], [1], "00:00", "23:59") + + cancelled = cancel_acceptance_task(task["task_id"]) + + self.assertEqual(cancelled["status"], "cancelled") + self.assertTrue(cancelled["schedule_enabled"]) + self.assertIsNotNone(cancelled["next_run_at"]) + self.assertEqual(cancel_acceptance_task(task["task_id"])["status"], "cancelled") + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.get_wen_resources") + def test_custom_cookie_is_stored_as_secret_and_not_returned(self, resources): + resources.return_value = {"cookie_count": 0, "device_count": 0, "credential_ids": []} + custom_cookie = "web_session=secret-value; a1=another-secret" + + task = create_acceptance_task({ + "acceptance_method": "data-analysis", + "feishu_links": ["https://example.feishu.cn/docx/project"], + "keywords": ["山姆特价"], + "custom_cookie": custom_cookie, + }) + + self.assertEqual(task["device_count"], 1) + self.assertNotIn(custom_cookie, str(task)) + self.assertTrue(task["resource_progress"][0]["resource_id"].startswith("custom-")) + with connection() as conn: + row = conn.execute( + "SELECT secret_value FROM acceptance_task_secrets WHERE secret_type='custom_cookie'" + ).fetchone() + self.assertEqual(row["secret_value"], custom_cookie) + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.get_wen_resources") + def test_weekly_schedule_and_immediate_collection(self, resources): + resources.return_value = {"cookie_count": 0, "device_count": 2} + source = create_acceptance_task({ + "acceptance_method": "screenshot", + "keywords": ["山姆特价"], + "device_count": 1, + }) + + scheduled = update_acceptance_schedule( + source["task_id"], [1, 3, 5], "01:30", "04:30", + ) + self.assertTrue(scheduled["schedule_enabled"]) + self.assertEqual(scheduled["schedule_weekday"], 1) + self.assertEqual(scheduled["schedule_weekdays"], [1, 3, 5]) + self.assertEqual(scheduled["schedule_start_time"], "01:30") + self.assertEqual(scheduled["schedule_end_time"], "04:30") + self.assertIsNotNone(scheduled["next_run_at"]) + + child = collect_acceptance_now(source["task_id"]) + self.assertNotEqual(child["task_id"], source["task_id"]) + self.assertEqual(child["keywords"], ["山姆特价"]) + self.assertFalse(child["schedule_enabled"]) + + stopped = stop_acceptance_schedule(source["task_id"]) + self.assertFalse(stopped["schedule_enabled"]) + self.assertIsNone(stopped["next_run_at"]) + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.get_wen_resources") + def test_screenshot_project_can_start_cookie_crawler_batch(self, resources): + resources.return_value = {"cookie_count": 2, "device_count": 3, "credential_ids": ["cookie-1", "cookie-2"]} + source = create_acceptance_task({ + "acceptance_method": "screenshot", "keywords": ["山姆会员卡"], + "match_keywords": ["民生"], "device_count": 3, + }) + + child = collect_acceptance_crawler_now(source["task_id"]) + + self.assertEqual(child["acceptance_method"], "data-analysis") + self.assertEqual(child["keywords"], ["山姆会员卡"]) + self.assertEqual(child["match_keywords"], ["民生"]) + self.assertEqual(child["cookie_count"], 2) + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service._copy_acceptance_task") + @patch("acceptance_service.get_wen_resources") + def test_due_schedule_reads_persisted_multi_day_range(self, resources, copy_task): + resources.return_value = {"cookie_count": 0, "device_count": 2} + source = create_acceptance_task({ + "acceptance_method": "screenshot", + "keywords": ["山姆特价"], + "device_count": 1, + }) + update_acceptance_schedule(source["task_id"], [1, 4, 7], "00:00", "23:59") + with connection() as conn: + conn.execute( + "UPDATE acceptance_tasks SET next_run_at=? WHERE public_task_id=?", + (now_iso(), source["task_id"]), + ) + + self.assertEqual(run_due_acceptance_schedules(), 1) + copy_task.assert_called_once_with(source["task_id"], trigger_source="acceptance-weekly") + refreshed = get_acceptance_task(source["task_id"]) + self.assertEqual(refreshed["schedule_weekdays"], [1, 4, 7]) + self.assertIsNotNone(refreshed["last_run_at"]) + self.assertIsNotNone(refreshed["next_run_at"]) + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.schedule_xapi_screenshots") + @patch("acceptance_service.get_wen_resources") + def test_retry_failed_devices_dispatches_only_failed_accounts(self, resources, schedule): + resources.return_value = {"cookie_count": 0, "device_count": 2} + schedule.return_value = { + "execution_mode": "immediate", + "batches": [{ + "batch_id": "retry-batch-1", + "items": [{"xhs_account_id": "failed-account"}], + }], + } + source = create_acceptance_task({ + "acceptance_method": "screenshot", + "keywords": ["山姆特价"], + "device_count": 2, + }) + with connection() as conn: + task = conn.execute( + "SELECT id FROM acceptance_tasks WHERE public_task_id=?", (source["task_id"],), + ).fetchone() + now = now_iso() + run = conn.execute( + """INSERT INTO acceptance_keyword_runs + (task_id,keyword,run_type,run_order,status,created_at,updated_at) + VALUES (?,'山姆特价','screenshot',0,'failed',?,?)""", + (task["id"], now, now), + ) + conn.executemany( + """INSERT INTO acceptance_task_resources + (task_id,resource_type,resource_id,resource_order,created_at) + VALUES (?,'device',?,?,?)""", + [ + (task["id"], "failed-account", 0, now), + (task["id"], "successful-account", 1, now), + ], + ) + conn.executemany( + """INSERT INTO acceptance_xapi_runs + (task_id,keyword_run_id,batch_id,xapi_task_id,keyword,xhs_account_id, + status,artifacts_json,feishu_sync_status,raw_json,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", + [ + (task["id"], run.lastrowid, "old-1", "task-failed", "山姆特价", "failed-account", "failed", "[]", "failed", "{}", now, now), + (task["id"], run.lastrowid, "old-2", "task-success", "山姆特价", "successful-account", "completed", '[{"image_url":"https://example.test/ok.png"}]', "succeeded", "{}", now, now), + ], + ) + + retried = retry_failed_acceptance_devices(source["task_id"]) + + self.assertEqual(retried["retry_dispatched_device_count"], 1) + self.assertEqual(retried["retry_dispatched_pair_count"], 1) + schedule.assert_called_once_with( + ["山姆特价"], trigger_source="acceptance", + xhs_account_ids=["failed-account"], + ) + with connection() as conn: + keyword_run = conn.execute( + "SELECT status,crawler_task_id FROM acceptance_keyword_runs WHERE id=?", + (run.lastrowid,), + ).fetchone() + self.assertEqual(keyword_run["status"], "waiting") + self.assertEqual(keyword_run["crawler_task_id"], "retry-batch-1") + + @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) + @patch("acceptance_service.get_wen_resources") + def test_combined_task_uses_all_resources_and_keeps_match_keywords(self, resources): + resources.return_value = { + "cookie_count": 3, "device_count": 5, + "credential_ids": ["cookie-1", "cookie-2", "cookie-3"], + } + task = create_acceptance_task({ + "acceptance_method": "combined", + "feishu_links": ["https://example.feishu.cn/docx/project"], + "keywords": ["山姆特价"], + "match_keywords": ["民生", "联名卡"], + }) + self.assertEqual(task["device_count"], 5) + self.assertEqual(task["cookie_count"], 3) + self.assertEqual(task["match_keywords"], ["民生", "联名卡"]) + self.assertEqual( + len([item for item in task["resource_progress"] if item["resource_type"] == "cookie"]), + 3, + ) + + +if __name__ == "__main__": + unittest.main() + cancel_acceptance_task,