diff --git a/acceptance_service.py b/acceptance_service.py index 552326f..bbbf43e 100644 --- a/acceptance_service.py +++ b/acceptance_service.py @@ -1,5 +1,6 @@ import json import csv +import hashlib import io import platform import re @@ -8,9 +9,10 @@ import string import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import datetime +from datetime import datetime, timedelta from html import unescape from html.parser import HTMLParser +from http.cookies import SimpleCookie from typing import Any from urllib.parse import urlparse @@ -22,8 +24,10 @@ from crawler_client import ( collect_keyword_sources, fetch_content_detail, get_wen_resources, + list_xhs_wen_xapi_devices, match_feishu_notes, query_collection, + query_xapi_batch_runs, schedule_xapi_screenshots, ) from database import connection, now_iso, upsert_feishu_note @@ -31,9 +35,11 @@ from task_service import TaskError ALLOWED_HOST_SUFFIXES = (".feishu.cn", ".larksuite.com") -METHODS = {"screenshot", "data-analysis"} +METHODS = {"screenshot", "data-analysis", "combined"} +WEEKDAYS = set(range(1, 8)) +_schedule_started = False XHS_URL_PATTERN = re.compile( - r"https?://(?:www\.)?(?:xiaohongshu\.com|xhslink\.com)/[^\s\t\r\n<>\"']+", + r"https?://(?:www\.)?(?:xiaohongshu\.com|xhslink\.(?:com|cn))/[^\s\t\r\n<>\"']+", re.IGNORECASE, ) @@ -78,6 +84,34 @@ def _public_id() -> str: return f"HV-{datetime.now():%Y%m%d}-{''.join(secrets.choice(chars) for _ in range(5))}" +def _normalize_custom_cookie(value: Any) -> str: + cookie = str(value or "").strip() + if cookie.lower().startswith("cookie:"): + cookie = cookie.split(":", 1)[1].strip() + if not cookie: + return "" + if len(cookie) > 16384: + raise TaskError("CUSTOM_COOKIE_TOO_LONG", "自定义 Cookie 不能超过 16384 个字符", 422) + parsed = SimpleCookie() + try: + parsed.load(cookie) + except Exception as exc: + raise TaskError("INVALID_CUSTOM_COOKIE", "自定义 Cookie 格式无效", 422) from exc + if not parsed: + raise TaskError("INVALID_CUSTOM_COOKIE", "自定义 Cookie 格式无效", 422) + return cookie + + +def _custom_cookie_for_task(task_id: int) -> str: + with connection() as conn: + row = conn.execute( + """SELECT secret_value FROM acceptance_task_secrets + WHERE task_id=? AND secret_type='custom_cookie'""", + (task_id,), + ).fetchone() + return str(row["secret_value"] if row else "") + + def _validate_url(value: Any) -> str: url = str(value or "").strip() parsed = urlparse(url) @@ -343,11 +377,17 @@ def _update_run(run_id: int, *, status: str, result: dict[str, Any] | None = Non ) -def get_acceptance_resources() -> dict[str, int]: +def get_acceptance_resources() -> dict[str, Any]: resources = get_wen_resources() + try: + devices = list_xhs_wen_xapi_devices() + except Exception: + devices = [] return { "cookie_count": max(0, int(resources.get("cookie_count") or 0)), "device_count": max(0, int(resources.get("device_count") or 0)), + "ocr_questions": resources.get("ocr_questions") or [], + "devices": devices, } @@ -418,11 +458,20 @@ def _run_keyword_with_cookie_failover( task_id: int, run_id: int, keyword: str, target_count: int, ) -> bool: """补足目标成功数；成功 Cookie 不重跑，失败时自动尝试池中其他 Cookie。""" - resources = get_wen_resources() - available_ids = list(dict.fromkeys( - str(item).strip() for item in resources.get("credential_ids", []) - if str(item).strip() - )) + custom_cookie = _custom_cookie_for_task(task_id) + custom_credential_id = ( + f"custom-{hashlib.sha256(custom_cookie.encode('utf-8')).hexdigest()[:10]}" + if custom_cookie else "" + ) + resources = get_wen_resources() if not custom_cookie else {} + available_ids = ( + [custom_credential_id] + if custom_cookie + else list(dict.fromkeys( + str(item).strip() for item in resources.get("credential_ids", []) + if str(item).strip() + )) + ) with connection() as conn: now = now_iso() for index, credential_id in enumerate(available_ids[:target_count]): @@ -459,7 +508,11 @@ def _run_keyword_with_cookie_failover( ) try: result, crawler_id, remote_status = _wait_collection_result( - collect_keyword_sources(keyword, credential_id) + collect_keyword_sources( + keyword, + None if custom_cookie else credential_id, + custom_cookie or None, + ) ) if crawler_id: crawler_ids.append(crawler_id) @@ -510,12 +563,53 @@ def _execute_acceptance_method(task_id: int, method: str, keywords: list[str], r failures = 0 if method == "screenshot": try: + with connection() as conn: + trigger_row = conn.execute( + "SELECT trigger_source FROM acceptance_tasks WHERE id=?", (task_id,), + ).fetchone() + trigger_source = str(trigger_row["trigger_source"] if trigger_row else "acceptance") for run_id in run_ids: _update_run(run_id, status="running") - result = schedule_xapi_screenshots(keywords, resource_count) - crawler_id = str(result.get("task_id") or result.get("schedule_id") or "") or None - for run_id in run_ids: - _update_run(run_id, status="completed", result=result, crawler_task_id=crawler_id, completed=True) + result = schedule_xapi_screenshots( + keywords, resource_count, trigger_source=trigger_source, + ) + if result.get("execution_mode") == "immediate": + plans = {keyword: {"batches": result.get("batches") or []} for keyword in keywords} + else: + plans = { + str(plan.get("search_query") or ""): plan + for plan in result.get("plans") or [] if isinstance(plan, dict) + } + plan_rows = [] + for run_id, keyword in zip(run_ids, keywords): + plan = plans.get(keyword) or {} + batch_ids = [ + str(batch.get("batch_id") or "").strip() + for batch in plan.get("batches") or [] + if str(batch.get("batch_id") or "").strip() + ] + _update_run( + run_id, status="waiting", result=plan or result, + crawler_task_id=",".join(batch_ids) or None, + ) + plan_rows.append(plan) + with connection() as conn: + resource_order = 0 + seen_resources = set() + for plan in plan_rows: + for batch in plan.get("batches") or []: + for item in batch.get("items") or []: + account_id = str(item.get("xhs_account_id") or "").strip() + if not account_id or account_id in seen_resources: + continue + seen_resources.add(account_id) + conn.execute( + """INSERT OR IGNORE INTO acceptance_task_resources + (task_id,resource_type,resource_id,resource_order,created_at) + VALUES (?,'device',?,?,?)""", + (task_id, account_id, resource_order, now_iso()), + ) + resource_order += 1 except Exception as exc: failures = len(run_ids) for run_id in run_ids: @@ -551,25 +645,218 @@ def _execute_acceptance_method(task_id: int, method: str, keywords: list[str], r return failures +def _decode_json(value: Any, fallback: Any) -> Any: + if not value: + return fallback + if not isinstance(value, str): + return value + try: + return json.loads(value) + except (TypeError, json.JSONDecodeError): + return fallback + + +def _upsert_acceptance_xapi_runs( + task_id: int, keyword_run_id: int | None, batch_id: str, + runs: list[dict[str, Any]], +) -> None: + now = now_iso() + with connection() as conn: + for run in runs: + xapi_task_id = str(run.get("task_id") or run.get("id") or "").strip() + if not xapi_task_id: + continue + artifacts = run.get("artifacts") + if artifacts is None: + artifacts = _decode_json(run.get("artifacts_json"), []) + feishu_record = run.get("feishu_record") + if feishu_record is None: + feishu_record = _decode_json(run.get("feishu_record_json"), None) + conn.execute( + """INSERT INTO acceptance_xapi_runs + (task_id,keyword_run_id,batch_id,xapi_task_id,keyword,xhs_account_id, + device_serial,device_label,status,writeback_status,source_base_url, + base_record_id,artifacts_json,feishu_result_text,feishu_record_json, + feishu_sync_status,feishu_sync_error,raw_json,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(xapi_task_id) DO UPDATE SET + keyword_run_id=excluded.keyword_run_id,batch_id=excluded.batch_id, + keyword=excluded.keyword,xhs_account_id=excluded.xhs_account_id, + device_serial=excluded.device_serial,device_label=excluded.device_label, + status=excluded.status,writeback_status=excluded.writeback_status, + source_base_url=excluded.source_base_url,base_record_id=excluded.base_record_id, + artifacts_json=excluded.artifacts_json, + feishu_result_text=excluded.feishu_result_text, + feishu_record_json=excluded.feishu_record_json, + feishu_sync_status=excluded.feishu_sync_status, + feishu_sync_error=excluded.feishu_sync_error, + raw_json=excluded.raw_json,updated_at=excluded.updated_at""", + ( + task_id, keyword_run_id, batch_id, xapi_task_id, + run.get("search_query"), run.get("xhs_account_id"), + run.get("device_serial"), run.get("device_label"), run.get("status"), + run.get("writeback_status"), run.get("source_base_url"), + run.get("base_record_id"), json.dumps(artifacts or [], ensure_ascii=False), + run.get("feishu_result_text"), + json.dumps(feishu_record, ensure_ascii=False) if feishu_record is not None else None, + run.get("feishu_sync_status"), run.get("feishu_sync_error"), + json.dumps(run, ensure_ascii=False, default=str), now, now, + ), + ) + + +def _poll_screenshot_results(public_id: str) -> None: + """轮询已计划批次，持久化截图和飞书结果，并更新前端任务进度。""" + terminal_statuses = {"completed", "failed", "cancelled", "canceled"} + with connection() as conn: + task = conn.execute( + "SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,), + ).fetchone() + if not task or task["acceptance_method"] not in {"screenshot", "combined"}: + return + keyword_runs = conn.execute( + """SELECT id,keyword,crawler_task_id,result_json FROM acceptance_keyword_runs + WHERE task_id=? AND run_type='screenshot' ORDER BY run_order""", + (task["id"],), + ).fetchall() + batch_map: dict[str, tuple[int, str, int]] = {} + for keyword_run in keyword_runs: + plan = _decode_json(keyword_run["result_json"], {}) or {} + batches = plan.get("batches") or [] + expected_by_id = { + str(batch.get("batch_id") or ""): len(batch.get("items") or []) + for batch in batches if str(batch.get("batch_id") or "").strip() + } + for batch_id in str(keyword_run["crawler_task_id"] or "").split(","): + normalized = batch_id.strip() + if normalized: + batch_map[normalized] = ( + keyword_run["id"], keyword_run["keyword"], + max(1, expected_by_id.get(normalized, 1)), + ) + if not batch_map: + with connection() as conn: + conn.execute( + """UPDATE acceptance_tasks SET status='failed',progress_percent=100, + error_message=?,updated_at=? WHERE id=?""", + ("截图调度没有返回可回查的 batch_id", now_iso(), task["id"]), + ) + return + + deadline = time.monotonic() + max(1.0, Config.XAPI_RESULT_MAX_WAIT_HOURS) * 3600 + while True: + poll_errors = [] + for batch_id, (keyword_run_id, _, _) in batch_map.items(): + try: + result = query_xapi_batch_runs(batch_id, refresh=True) + _upsert_acceptance_xapi_runs( + task["id"], keyword_run_id, batch_id, result.get("runs") or [], + ) + except Exception as exc: + poll_errors.append(f"{batch_id}: {exc}") + + with connection() as conn: + rows = conn.execute( + "SELECT * FROM acceptance_xapi_runs WHERE task_id=?", (task["id"],), + ).fetchall() + expected_total = sum(item[2] for item in batch_map.values()) + terminal_count = sum( + str(row["status"] or "").lower() in terminal_statuses for row in rows + ) + feishu_ready = all( + str(row["status"] or "").lower() != "completed" + or str(row["feishu_sync_status"] or "").lower() == "succeeded" + for row in rows + ) + received = min(len(rows), expected_total) + progress = min(99, 70 + round(29 * received / max(1, expected_total))) + for keyword_run in keyword_runs: + related = [row for row in rows if row["keyword_run_id"] == keyword_run["id"]] + related_batch_ids = [ + value.strip() for value in str(keyword_run["crawler_task_id"] or "").split(",") + if value.strip() + ] + keyword_expected = sum(batch_map[item][2] for item in related_batch_ids) + keyword_terminal = sum( + str(row["status"] or "").lower() in terminal_statuses for row in related + ) + if keyword_expected and keyword_terminal >= keyword_expected: + failed = any(str(row["status"] or "").lower() != "completed" for row in related) + conn.execute( + """UPDATE acceptance_keyword_runs SET status=?,error_message=?, + completed_at=?,updated_at=? WHERE id=?""", + ("failed" if failed else "completed", + "部分设备截图失败" if failed else None, + now_iso(), now_iso(), keyword_run["id"]), + ) + all_terminal = expected_total > 0 and terminal_count >= expected_total + timed_out = time.monotonic() >= deadline + if all_terminal and feishu_ready: + failed_count = sum( + str(row["status"] or "").lower() != "completed" for row in rows + ) + final_status = "partial" if failed_count else "completed" + conn.execute( + """UPDATE acceptance_tasks SET status=?,progress_percent=100, + error_message=?,updated_at=? WHERE id=?""", + (final_status, "部分设备截图失败" if failed_count else None, + now_iso(), task["id"]), + ) + return + if timed_out: + reason = "截图或飞书结果等待超时" + if poll_errors: + reason += "；" + "；".join(poll_errors[:3]) + conn.execute( + """UPDATE acceptance_tasks SET status='partial',progress_percent=100, + error_message=?,updated_at=? WHERE id=?""", + (reason[:2000], now_iso(), task["id"]), + ) + return + conn.execute( + "UPDATE acceptance_tasks SET status='running',progress_percent=?,updated_at=? WHERE id=?", + (progress, now_iso(), task["id"]), + ) + time.sleep(max(5.0, Config.XAPI_RESULT_POLL_INTERVAL_SECONDS)) + + def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]: links = list(dict.fromkeys(_validate_url(item) for item in data.get("feishu_links", []))) - if not links: - raise TaskError("FEISHU_LINK_REQUIRED", "至少需要一个公开飞书链接", 422) method = str(data.get("acceptance_method") or "") if method not in METHODS: raise TaskError("INVALID_ACCEPTANCE_METHOD", "验收方式无效", 422) - keywords = list(dict.fromkeys(str(item).strip() for item in data.get("keywords", []) if str(item).strip())) + custom_cookie = _normalize_custom_cookie(data.get("custom_cookie")) + if custom_cookie and method not in {"data-analysis", "combined"}: + raise TaskError( + "CUSTOM_COOKIE_NOT_SUPPORTED", + "自定义 Cookie 仅支持数据分析验收", + 422, + ) raw_resources = get_wen_resources() + keywords = list(dict.fromkeys( + str(item).strip() for item in data.get("keywords", []) if str(item).strip() + )) + match_keywords = list(dict.fromkeys( + str(item).strip() for item in data.get("match_keywords", []) if str(item).strip() + )) + if len(keywords) > 20: + 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: + raise TaskError("FEISHU_LINK_REQUIRED", "笔记收录请至少输入一个公开飞书链接", 422) resources = { "cookie_count": max(0, int(raw_resources.get("cookie_count") or 0)), "device_count": max(0, int(raw_resources.get("device_count") or 0)), } - device_count = ( - resources["cookie_count"] if method == "data-analysis" - else int(data.get("device_count") or 1) + cookie_count = 1 if custom_cookie else resources["cookie_count"] + device_count = resources["device_count"] if method == "combined" else ( + cookie_count if method == "data-analysis" else int(data.get("device_count") or 1) ) - available_count = resources["device_count" if method == "screenshot" else "cookie_count"] - resource_name = "设备" if method == "screenshot" else "Cookie" + available_count = resources["device_count"] if method in {"screenshot", "combined"} else cookie_count + resource_name = "设备" if method in {"screenshot", "combined"} else "Cookie" + if method == "combined" and cookie_count < 1: + raise TaskError("INVALID_RESOURCE_COUNT", "当前没有可用的数据分析 Cookie", 422) if not 1 <= device_count <= available_count: raise TaskError( "INVALID_RESOURCE_COUNT", @@ -583,22 +870,38 @@ def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]: cursor = conn.execute( """INSERT INTO acceptance_tasks (public_task_id,name,acceptance_method,status,progress_percent,device_count, - created_by_id,created_by_name,created_at,updated_at) - VALUES (?,?,?,'waiting',0,?,?,?,?,?)""", + created_by_id,created_by_name,created_at,updated_at,trigger_source,cookie_count) + VALUES (?,?,?,'waiting',0,?,?,?,?,?,?,?)""", (public_id, name, method, device_count, str(data.get("created_by_id") or "local-dev"), - str(data.get("created_by_name") or "本地开发用户"), now, now), + str(data.get("created_by_name") or "本地开发用户"), now, now, + str(data.get("trigger_source") or "acceptance"), cookie_count), ) task_id = cursor.lastrowid conn.executemany( "INSERT INTO acceptance_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", [(task_id, index, keyword, now) for index, keyword in enumerate(keywords)], ) - if method == "data-analysis": - credential_ids = [ - str(item).strip() - for item in raw_resources.get("credential_ids", []) - if str(item).strip() - ][:device_count] + conn.executemany( + "INSERT INTO acceptance_match_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", + [(task_id, index, keyword, now) for index, keyword in enumerate(match_keywords)], + ) + if method in {"data-analysis", "combined"}: + if custom_cookie: + credential_ids = [ + f"custom-{hashlib.sha256(custom_cookie.encode('utf-8')).hexdigest()[:10]}" + ] + conn.execute( + """INSERT INTO acceptance_task_secrets + (task_id,secret_type,secret_value,created_at) + VALUES (?,'custom_cookie',?,?)""", + (task_id, custom_cookie, now), + ) + else: + credential_ids = [ + str(item).strip() + for item in raw_resources.get("credential_ids", []) + if str(item).strip() + ][:cookie_count] conn.executemany( """INSERT OR IGNORE INTO acceptance_task_resources (task_id,resource_type,resource_id,resource_order,created_at) @@ -615,6 +918,143 @@ def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]: return get_acceptance_task(public_id) +def _next_weekly_run(weekday: int, *, after: datetime | None = None) -> str: + current = after or datetime.now().astimezone() + days = (weekday - current.isoweekday()) % 7 + candidate = (current + timedelta(days=days)).replace( + hour=0, minute=0, second=0, microsecond=0, + ) + if candidate <= current: + candidate += timedelta(days=7) + return candidate.isoformat(timespec="seconds") + + +def update_acceptance_schedule(public_id: str, weekday: int) -> dict[str, Any]: + try: + weekday = int(weekday) + except (TypeError, ValueError) as exc: + raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422) from exc + if weekday not in WEEKDAYS: + raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422) + with connection() as conn: + task = conn.execute( + "SELECT id FROM acceptance_tasks WHERE public_task_id=?", (public_id,), + ).fetchone() + if not task: + raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404) + conn.execute( + """UPDATE acceptance_tasks SET schedule_enabled=1,schedule_weekday=?, + next_run_at=?,schedule_error=NULL,updated_at=? WHERE id=?""", + (weekday, _next_weekly_run(weekday), now_iso(), task["id"]), + ) + return get_acceptance_task(public_id) + + +def stop_acceptance_schedule(public_id: str) -> dict[str, Any]: + with connection() as conn: + result = conn.execute( + """UPDATE acceptance_tasks SET schedule_enabled=0,next_run_at=NULL, + schedule_error=NULL,updated_at=? WHERE public_task_id=?""", + (now_iso(), public_id), + ) + if not result.rowcount: + raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404) + return get_acceptance_task(public_id) + + +def _copy_acceptance_task(public_id: str, *, trigger_source: str = "acceptance") -> dict[str, Any]: + 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) + keywords = [row["keyword"] for row in conn.execute( + "SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", + (task["id"],), + ).fetchall()] + match_keywords = [row["keyword"] for row in conn.execute( + "SELECT keyword FROM acceptance_match_keywords WHERE task_id=? ORDER BY keyword_order", + (task["id"],), + ).fetchall()] + links = [row["source_url"] for row in conn.execute( + "SELECT source_url FROM acceptance_feishu_sources WHERE task_id=? ORDER BY source_order", + (task["id"],), + ).fetchall()] + secret = conn.execute( + """SELECT secret_value FROM acceptance_task_secrets + WHERE task_id=? AND secret_type='custom_cookie'""", + (task["id"],), + ).fetchone() + 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"], + "created_by_id": task["created_by_id"], + "created_by_name": task["created_by_name"], + "custom_cookie": secret["secret_value"] if secret else None, + "trigger_source": trigger_source, + }) + with connection() as conn: + conn.execute( + "UPDATE acceptance_tasks SET scheduled_from_task_id=? WHERE public_task_id=?", + (task["id"], child["task_id"]), + ) + return get_acceptance_task(child["task_id"]) + + +def collect_acceptance_now(public_id: str) -> dict[str, Any]: + """复用任务配置创建一次独立采集，截图任务会立即向 Framework 下发。""" + return _copy_acceptance_task(public_id, trigger_source="acceptance") + + +def run_due_acceptance_schedules() -> int: + now = now_iso() + with connection() as conn: + due = conn.execute( + """SELECT public_task_id,schedule_weekday FROM acceptance_tasks + WHERE schedule_enabled=1 AND next_run_at IS NOT NULL AND next_run_at<=? + ORDER BY next_run_at LIMIT 20""", + (now,), + ).fetchall() + for row in due: + conn.execute( + """UPDATE acceptance_tasks SET next_run_at=?,last_run_at=?, + schedule_error=NULL,updated_at=? WHERE public_task_id=?""", + (_next_weekly_run(row["schedule_weekday"]), now, now, row["public_task_id"]), + ) + for row in due: + try: + _copy_acceptance_task(row["public_task_id"], trigger_source="acceptance-weekly") + except Exception as exc: + with connection() as conn: + conn.execute( + "UPDATE acceptance_tasks SET schedule_error=?,updated_at=? WHERE public_task_id=?", + (str(exc)[:2000], now_iso(), row["public_task_id"]), + ) + return len(due) + + +def start_acceptance_schedule_worker() -> None: + global _schedule_started + if _schedule_started: + return + _schedule_started = True + + def worker() -> None: + while True: + try: + run_due_acceptance_schedules() + except Exception: + pass + time.sleep(30) + + threading.Thread(target=worker, daemon=True, name="acceptance-weekly-scheduler").start() + + def _fetch_sources(public_id: str) -> None: with connection() as conn: task = conn.execute("SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone() @@ -629,11 +1069,12 @@ def _fetch_sources(public_id: str) -> None: # 数据分析时，表单中已知的关键词无需等待飞书解析完成，立即并行采集。 analysis_pool = None initial_analysis_future = None - if task["acceptance_method"] == "data-analysis" and initial_keywords: + if task["acceptance_method"] in {"data-analysis", "combined"} and initial_keywords: analysis_pool = ThreadPoolExecutor(max_workers=1, thread_name_prefix=f"analysis-{public_id}") initial_analysis_future = analysis_pool.submit( _execute_acceptance_method, - task["id"], task["acceptance_method"], initial_keywords, task["device_count"], + task["id"], "data-analysis", initial_keywords, + task["cookie_count"] if task["acceptance_method"] == "combined" else task["device_count"], ) failures = 0 note_failures = 0 @@ -711,7 +1152,7 @@ def _fetch_sources(public_id: str) -> None: conn.execute("UPDATE acceptance_tasks SET progress_percent=?,updated_at=? WHERE public_task_id=?", (progress, now_iso(), public_id)) run_failures = 0 run_error = None - if failures < len(sources): + if not sources or failures < len(sources): try: keywords = _sync_acceptance_keywords(task["id"], discovered_keywords) with connection() as conn: @@ -721,7 +1162,12 @@ def _fetch_sources(public_id: str) -> None: new_keywords = [keyword for keyword in keywords if keyword not in initial_keywords] if new_keywords: run_failures += _execute_acceptance_method( - task["id"], task["acceptance_method"], new_keywords, task["device_count"] + task["id"], "data-analysis", new_keywords, + task["cookie_count"] if task["acceptance_method"] == "combined" else task["device_count"], + ) + if task["acceptance_method"] == "combined": + run_failures += _execute_acceptance_method( + task["id"], "screenshot", keywords, task["device_count"], ) else: run_failures = _execute_acceptance_method( @@ -742,6 +1188,19 @@ def _fetch_sources(public_id: str) -> None: finally: if analysis_pool is not None: analysis_pool.shutdown(wait=False) + if task["acceptance_method"] in {"screenshot", "combined"} and not run_error and not run_failures: + with connection() as conn: + conn.execute( + """UPDATE acceptance_tasks SET status='running',progress_percent=70, + error_message=?,updated_at=? WHERE public_task_id=?""", + ("部分飞书资料或小红书笔记解析失败" if failures or note_failures else None, + now_iso(), public_id), + ) + threading.Thread( + target=_poll_screenshot_results, + args=(public_id,), daemon=True, name=f"acceptance-xapi-{public_id}", + ).start() + return status = "failed" if failures == len(sources) or run_error else ("partial" if failures or note_failures or run_failures else "completed") error = "；".join(filter(None, [ "部分飞书资料或小红书笔记解析失败" if failures or note_failures else None, @@ -782,13 +1241,23 @@ def resume_acceptance_task(public_id: str) -> dict[str, Any]: def resume_unfinished_acceptance_tasks() -> None: """服务重启后恢复等待中或运行中的验收任务。""" with connection() as conn: - task_ids = [row["public_task_id"] for row in conn.execute( - """SELECT public_task_id FROM acceptance_tasks + tasks = conn.execute( + """SELECT t.public_task_id,t.acceptance_method, + EXISTS(SELECT 1 FROM acceptance_keyword_runs r + WHERE r.task_id=t.id AND r.run_type='screenshot' + AND r.crawler_task_id IS NOT NULL) AS has_screenshot_batches + FROM acceptance_tasks t WHERE status IN ('waiting','running') ORDER BY created_at""" - ).fetchall()] - for public_id in task_ids: + ).fetchall() + for task in tasks: + public_id = task["public_task_id"] + target = ( + _poll_screenshot_results + if task["acceptance_method"] in {"screenshot", "combined"} and task["has_screenshot_batches"] + else _fetch_sources + ) threading.Thread( - target=_fetch_sources, args=(public_id,), daemon=True, + target=target, args=(public_id,), daemon=True, name=f"acceptance-recover-{public_id}", ).start() @@ -843,6 +1312,17 @@ def _task_dict(conn, task) -> dict[str, Any]: keywords = [row["keyword"] for row in conn.execute( "SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (task["id"],) ).fetchall()] + match_keywords = [row["keyword"] for row in conn.execute( + "SELECT keyword FROM acceptance_match_keywords WHERE task_id=? ORDER BY keyword_order", + (task["id"],), + ).fetchall()] + root_id = task["scheduled_from_task_id"] or task["id"] + batches = [dict(row) for row in conn.execute( + """SELECT public_task_id,status,progress_percent,created_at,updated_at + FROM acceptance_tasks WHERE id=? OR scheduled_from_task_id=? + ORDER BY created_at DESC""", + (root_id, root_id), + ).fetchall()] sources = [dict(row) for row in conn.execute( """SELECT source_url,final_url,source_type,title,content_text,fetch_status,http_status, error_message,fetched_at, @@ -879,8 +1359,57 @@ def _task_dict(conn, task) -> dict[str, Any]: FROM acceptance_cookie_attempts WHERE task_id=? ORDER BY id""", (task["id"],), ).fetchall() + xapi_rows = conn.execute( + """SELECT batch_id,xapi_task_id,keyword,xhs_account_id,device_serial, + device_label,status,writeback_status,source_base_url,base_record_id, + artifacts_json,feishu_result_text,feishu_record_json, + feishu_sync_status,feishu_sync_error,updated_at + FROM acceptance_xapi_runs WHERE task_id=? ORDER BY id""", + (task["id"],), + ).fetchall() resource_progress = [] for resource in task_resources: + if resource["resource_type"] == "device": + related_runs = [ + row for row in xapi_rows if row["xhs_account_id"] == resource["resource_id"] + ] + completed_keywords = len({ + row["keyword"] for row in related_runs + if str(row["status"] or "").lower() == "completed" + and str(row["feishu_sync_status"] or "").lower() == "succeeded" + }) + failed_keywords = len({ + row["keyword"] for row in related_runs + if str(row["status"] or "").lower() in {"failed", "cancelled", "canceled"} + }) + running_keywords = len({ + row["keyword"] for row in related_runs + if str(row["status"] or "").lower() not in {"completed", "failed", "cancelled", "canceled"} + }) + total_keywords = len(keywords) + terminal_keywords = completed_keywords + failed_keywords + resource_progress.append({ + "resource_type": "device", + "resource_id": resource["resource_id"], + "resource_order": resource["resource_order"], + "status": ( + "failed" if failed_keywords + else "completed" if total_keywords and completed_keywords >= total_keywords + else "running" if related_runs + else "waiting" + ), + "keyword_total": total_keywords, + "keyword_completed": completed_keywords, + "keyword_failed": failed_keywords, + "keyword_running": running_keywords, + "progress_percent": round(terminal_keywords / max(1, total_keywords) * 100), + "attempt_count": len(related_runs), + "error_message": next( + (row["feishu_sync_error"] for row in reversed(related_runs) if row["feishu_sync_error"]), + None, + ), + }) + continue related = [row for row in cookie_attempts if row["credential_id"] == resource["resource_id"]] completed_keywords = failed_keywords = running_keywords = 0 for keyword in keywords: @@ -914,14 +1443,29 @@ def _task_dict(conn, task) -> dict[str, Any]: "attempt_count": len(related), "error_message": next((row["error_message"] for row in reversed(related) if row["error_message"]), None), }) + xapi_runs = [] + for row in xapi_rows: + item = dict(row) + item["artifacts"] = _decode_json(item.pop("artifacts_json"), []) + item["feishu_record"] = _decode_json(item.pop("feishu_record_json"), None) + xapi_runs.append(item) return { "task_id": task["public_task_id"], "name": task["name"], "acceptance_method": task["acceptance_method"], "status": task["status"], "progress_percent": task["progress_percent"], "device_count": task["device_count"], - "keywords": keywords, "feishu_links": [item["source_url"] for item in sources], + "cookie_count": task["cookie_count"], + "schedule_enabled": bool(task["schedule_enabled"]), + "schedule_weekday": task["schedule_weekday"], + "next_run_at": task["next_run_at"], "last_run_at": task["last_run_at"], + "scheduled_from_task_id": task["scheduled_from_task_id"], + "schedule_error": task["schedule_error"], + "trigger_source": task["trigger_source"], + "keywords": keywords, "match_keywords": match_keywords, + "batches": batches, "feishu_links": [item["source_url"] for item in sources], "feishu_sources": sources, "created_by": {"id": task["created_by_id"], "name": task["created_by_name"]}, "note_stats": {"total": note_stats["total"] or 0, "completed": note_stats["completed"] or 0, "failed": note_stats["failed"] or 0}, "keyword_runs": keyword_runs, + "xapi_runs": xapi_runs, "resource_progress": resource_progress, "error_message": task["error_message"], "created_at": task["created_at"], "updated_at": task["updated_at"], } diff --git a/apifox-business-tasks.openapi.json b/apifox-business-tasks.openapi.json index 97c1f64..1339af6 100644 --- a/apifox-business-tasks.openapi.json +++ b/apifox-business-tasks.openapi.json @@ -12,6 +12,53 @@ } ], "paths": { + "/api/v1/acceptance-tasks": { + "post": { + "summary": "创建项目验收任务", + "description": "关键词截图取证必须填写 keywords，无需飞书链接；笔记收录必须填写 feishu_links，可选传入仅供本任务使用的 custom_cookie。", + "tags": ["项目验收"], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/CreateAcceptanceTask"} + } + } + }, + "responses": {"201": {"description": "验收任务创建成功"}} + } + }, + "/api/v1/acceptance-tasks/{task_id}/collect-now": { + "post": { + "summary": "立即采集并下发指令", + "description": "复用原任务配置创建独立的新一轮采集。截图取证会立即请求 Framework 下发设备指令。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": {"201": {"description": "新一轮验收任务已创建"}} + } + }, + "/api/v1/acceptance-tasks/{task_id}/schedule": { + "put": { + "summary": "设置每周采集周期", + "description": "按 Asia/Shanghai 时区，在指定星期的 00:00 创建新一轮采集。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "requestBody": { + "required": true, + "content": {"application/json": {"schema": {"type": "object", "required": ["weekday"], "properties": {"weekday": {"type": "integer", "minimum": 1, "maximum": 7, "description": "1=周一，7=周日"}}}}} + }, + "responses": {"200": {"description": "周期设置成功"}} + } + }, + "/api/v1/acceptance-tasks/{task_id}/schedule/stop": { + "post": { + "summary": "停止周期采集", + "description": "关闭该任务后续的每周执行，不影响已经开始的采集任务。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": {"200": {"description": "周期采集已停止"}} + } + }, "/api/v1/tasks/{task_id}/cancel": { "post": { "summary": "停止任务", @@ -58,6 +105,36 @@ } }, "components": { + "schemas": { + "CreateAcceptanceTask": { + "type": "object", + "required": ["acceptance_method"], + "properties": { + "name": {"type": "string"}, + "acceptance_method": {"type": "string", "enum": ["screenshot", "data-analysis"]}, + "feishu_links": { + "type": "array", + "items": {"type": "string", "format": "uri"}, + "description": "笔记收录必填；关键词截图取证可为空。" + }, + "keywords": { + "type": "array", + "maxItems": 20, + "items": {"type": "string"}, + "description": "关键词截图取证至少填写 1 个；笔记收录选填。" + }, + "device_count": {"type": "integer", "minimum": 1}, + "custom_cookie": { + "type": "string", + "maxLength": 16384, + "writeOnly": true, + "description": "仅用于笔记收录。填写后仅使用该 Cookie；留空时使用 Cookie 池全部可用凭据。不会在任务详情中返回。" + }, + "created_by_id": {"type": "string"}, + "created_by_name": {"type": "string"} + } + } + }, "parameters": { "TaskId": { "name": "task_id", diff --git a/app.py b/app.py index 89e1b5a..60c9ebb 100644 --- a/app.py +++ b/app.py @@ -17,6 +17,7 @@ from auth_service import ( from config import Config from database import init_database from ocr_dashboard_service import ( + get_comparison as get_ocr_dashboard_comparison, get_result_matches as get_ocr_result_matches, get_summary as get_ocr_dashboard_summary, list_batches as list_ocr_batches, @@ -27,6 +28,7 @@ from ocr_dashboard_service import ( ) from acceptance_service import ( create_acceptance_task, + collect_acceptance_now, get_acceptance_resources, get_acceptance_link_ai_data, get_acceptance_task, @@ -34,6 +36,9 @@ from acceptance_service import ( list_acceptance_tasks, resume_acceptance_task, resume_unfinished_acceptance_tasks, + start_acceptance_schedule_worker, + stop_acceptance_schedule, + update_acceptance_schedule, ) from rule_service import ( delete_rule_document, @@ -308,6 +313,22 @@ def create_app() -> Flask: def resume_project_acceptance_task(task_id: str): return jsonify({"data": resume_acceptance_task(task_id), "request_id": g.request_id}) + @app.route("/api/v1/acceptance-tasks//collect-now", methods=["POST"]) + 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//schedule", methods=["PUT"]) + def configure_project_acceptance_schedule(task_id: str): + payload = request.get_json(silent=True) or {} + return jsonify({ + "data": update_acceptance_schedule(task_id, payload.get("weekday")), + "request_id": g.request_id, + }) + + @app.route("/api/v1/acceptance-tasks//schedule/stop", methods=["POST"]) + def stop_project_acceptance_schedule(task_id: str): + return jsonify({"data": stop_acceptance_schedule(task_id), "request_id": g.request_id}) + @app.route("/api/v1/data-dashboard/ocr/batches", methods=["GET"]) def query_ocr_dashboard_batches(): return jsonify({"data": list_ocr_batches(), "request_id": g.request_id}) @@ -323,6 +344,13 @@ def create_app() -> Flask: request.args.get("collection_key", "").strip(), ), "request_id": g.request_id}) + @app.route("/api/v1/data-dashboard/ocr/comparison", methods=["GET"]) + def query_ocr_dashboard_comparison(): + return jsonify({"data": get_ocr_dashboard_comparison( + request.args.get("batch_key", "").strip(), + request.args.get("collection_key", "").strip(), + ), "request_id": g.request_id}) + @app.route("/api/v1/data-dashboard/ocr/results", methods=["GET"]) def query_ocr_dashboard_results(): return jsonify({"data": list_ocr_results(request.args.to_dict()), "request_id": g.request_id}) @@ -437,6 +465,7 @@ def create_app() -> Flask: resume_unfinished_tasks() resume_unfinished_acceptance_tasks() + start_acceptance_schedule_worker() return app diff --git a/config.py b/config.py index c897457..25c657c 100644 --- a/config.py +++ b/config.py @@ -23,6 +23,8 @@ class Config: CRAWLER_RETRY_TIMES = int(os.getenv("CRAWLER_RETRY_TIMES", "3")) CRAWLER_RETRY_DELAY_SECONDS = float(os.getenv("CRAWLER_RETRY_DELAY_SECONDS", "2")) CRAWLER_POLL_INTERVAL_SECONDS = float(os.getenv("CRAWLER_POLL_INTERVAL_SECONDS", "5")) + XAPI_RESULT_POLL_INTERVAL_SECONDS = float(os.getenv("XAPI_RESULT_POLL_INTERVAL_SECONDS", "300")) + XAPI_RESULT_MAX_WAIT_HOURS = float(os.getenv("XAPI_RESULT_MAX_WAIT_HOURS", "48")) ACCEPTANCE_BROWSER_TIMEOUT_SECONDS = int(os.getenv("ACCEPTANCE_BROWSER_TIMEOUT_SECONDS", "60")) ACCEPTANCE_NOTE_WORKERS = int(os.getenv("ACCEPTANCE_NOTE_WORKERS", "4")) SEARCH_QA_API_BASE_URL = os.getenv( diff --git a/crawler_client.py b/crawler_client.py index 8366a2b..3f4de72 100644 --- a/crawler_client.py +++ b/crawler_client.py @@ -306,6 +306,17 @@ def get_wen_resources() -> dict[str, Any]: return _get("/api/v1/xhs/wen/resources") +def list_xhs_wen_xapi_devices() -> list[dict[str, Any]]: + """查询 Framework `xhs_device_accounts` 中启用的设备清单。 + + 通过 Framework 的调度配置接口读取，返回每台设备的账号、设备编码与 + 可用时间窗口等调度字段，供前端「截图设备列表」直接展示真实设备。 + """ + data = _get("/api/v1/xhs/wen/xapi/devices/settings") + items = data.get("items") + return items if isinstance(items, list) else [] + + def fetch_content_detail(note_url: str) -> dict[str, Any]: """按小红书笔记链接读取完整笔记详情。""" return _post( @@ -315,21 +326,43 @@ def fetch_content_detail(note_url: str) -> dict[str, Any]: ) -def schedule_xapi_screenshots(search_queries: list[str], device_count: int) -> dict[str, Any]: - """向手机集群下发关键词搜索与截图任务。""" +def schedule_xapi_screenshots( + search_queries: list[str], device_count: int, *, trigger_source: str = "acceptance", +) -> dict[str, Any]: + """向手机集群立即下发关键词搜索与截图任务。""" queries = normalize_collection_keywords(search_queries) return _post( "/api/v1/xhs/wen/xapi/schedule", - {"keywords": queries, "device_count": device_count}, + { + "keywords": queries, + "device_count": device_count, + "execute_now": True, + "trigger_source": trigger_source, + }, retry_transient=False, ) -def collect_keyword_sources(keyword: str, credential_id: str | None = None) -> dict[str, Any]: +def query_xapi_batch_runs(batch_id: str, *, refresh: bool = True) -> dict[str, Any]: + """读取某个 Framework 调度批次下的设备任务、截图与飞书结果。""" + normalized = str(batch_id or "").strip() + if not normalized: + raise CrawlerError("batch_id 不能为空") + return _get( + "/api/v1/xhs/wen/xapi/runs" + f"?batch_id={normalized}&refresh={'true' if refresh else 'false'}" + ) + + +def collect_keyword_sources( + keyword: str, credential_id: str | None = None, cookie: str | None = None, +) -> dict[str, Any]: """使用指定 Cookie 采集单个关键词。""" normalized = normalize_collection_keywords([keyword])[0] body: dict[str, Any] = {"keyword": normalized} - if credential_id: + if cookie: + body["cookie"] = cookie + elif credential_id: body["credential_id"] = credential_id return _post( "/api/v1/xhs/wen/sources", diff --git a/database.py b/database.py index 34dc98f..3380046 100644 --- a/database.py +++ b/database.py @@ -193,6 +193,17 @@ def init_database() -> None: FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE ); + CREATE TABLE IF NOT EXISTS acceptance_match_keywords ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + keyword_order INTEGER NOT NULL, + keyword TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(task_id, keyword_order), + UNIQUE(task_id, keyword), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE + ); + CREATE TABLE IF NOT EXISTS acceptance_feishu_sources ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, @@ -288,6 +299,34 @@ def init_database() -> None: CREATE INDEX IF NOT EXISTS idx_acceptance_keyword_runs_task ON acceptance_keyword_runs(task_id, status); + CREATE TABLE IF NOT EXISTS acceptance_xapi_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + keyword_run_id INTEGER, + batch_id TEXT NOT NULL, + xapi_task_id TEXT NOT NULL UNIQUE, + keyword TEXT, + xhs_account_id TEXT, + device_serial TEXT, + device_label TEXT, + status TEXT, + writeback_status TEXT, + source_base_url TEXT, + base_record_id TEXT, + artifacts_json TEXT, + feishu_result_text TEXT, + feishu_record_json TEXT, + feishu_sync_status TEXT, + feishu_sync_error TEXT, + raw_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE, + FOREIGN KEY(keyword_run_id) REFERENCES acceptance_keyword_runs(id) ON DELETE SET NULL + ); + CREATE INDEX IF NOT EXISTS idx_acceptance_xapi_runs_task + ON acceptance_xapi_runs(task_id, status, batch_id); + CREATE TABLE IF NOT EXISTS acceptance_cookie_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, @@ -321,8 +360,33 @@ def init_database() -> None: ); CREATE INDEX IF NOT EXISTS idx_acceptance_task_resources ON acceptance_task_resources(task_id, resource_order); + + CREATE TABLE IF NOT EXISTS acceptance_task_secrets ( + task_id INTEGER NOT NULL, + secret_type TEXT NOT NULL, + secret_value TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY(task_id, secret_type), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE + ); """ ) + acceptance_columns = { + row["name"] for row in conn.execute("PRAGMA table_info(acceptance_tasks)").fetchall() + } + acceptance_migrations = { + "schedule_enabled": "INTEGER NOT NULL DEFAULT 0", + "schedule_weekday": "INTEGER", + "next_run_at": "TEXT", + "last_run_at": "TEXT", + "scheduled_from_task_id": "INTEGER", + "schedule_error": "TEXT", + "trigger_source": "TEXT NOT NULL DEFAULT 'acceptance'", + "cookie_count": "INTEGER NOT NULL DEFAULT 0", + } + for column, definition in acceptance_migrations.items(): + if column not in acceptance_columns: + conn.execute(f"ALTER TABLE acceptance_tasks ADD COLUMN {column} {definition}") # 将旧版本保存在 acceptance_xhs_notes.parsed_json 中的成功结果迁移到 # 全局去重的笔记业务表；任务关联仍由 acceptance_xhs_notes 保留。 legacy_rows = conn.execute( diff --git a/ocr_dashboard_service.py b/ocr_dashboard_service.py index 3814a22..c1a64ac 100644 --- a/ocr_dashboard_service.py +++ b/ocr_dashboard_service.py @@ -52,6 +52,12 @@ def get_summary(batch_key: str = "", collection_key: str = "") -> dict[str, Any] }) or {} +def get_comparison(batch_key: str = "", collection_key: str = "") -> dict[str, Any]: + return _get("/xhs/ocr/dashboard/comparison", { + "batch_key": batch_key, "collection_key": collection_key, + }) or {} + + def list_results(params: dict[str, Any]) -> dict[str, Any]: allowed = { "batch_key", "collection_key", "question_code", "search_keyword", "city", diff --git a/tests/test_ocr_dashboard_service.py b/tests/test_ocr_dashboard_service.py index f8e2653..69ab3a2 100644 --- a/tests/test_ocr_dashboard_service.py +++ b/tests/test_ocr_dashboard_service.py @@ -1,7 +1,7 @@ import unittest from unittest.mock import Mock, patch -from ocr_dashboard_service import export_evidences, get_summary, list_batches, list_results +from ocr_dashboard_service import export_evidences, get_comparison, get_summary, list_batches, list_results from task_service import TaskError @@ -21,6 +21,23 @@ 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_comparison_forwards_scope(self, get): + response = Mock() + response.json.return_value = { + "success": True, "code": 200, + "data": {"current": {}, "previous": {}}, + } + get.return_value = response + + result = get_comparison("b1", "c1") + + self.assertIn("current", result) + self.assertEqual( + get.call_args.kwargs["params"], + {"batch_key": "b1", "collection_key": "c1"}, + ) + def test_results_rejects_oversized_page(self): with self.assertRaises(TaskError): list_results({"page_size": "201"})