import json import csv import hashlib import io import platform import re import secrets import string import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed 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 import requests from playwright.sync_api import sync_playwright from config import Config 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 from task_service import TaskError ALLOWED_HOST_SUFFIXES = (".feishu.cn", ".larksuite.com") 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|cn))/[^\s\t\r\n<>\"']+", re.IGNORECASE, ) class _ReadableHTML(HTMLParser): def __init__(self): super().__init__() self.title = "" self._in_title = False self._skip = 0 self.parts: list[str] = [] self.meta: dict[str, str] = {} def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: values = dict(attrs) if tag == "title": self._in_title = True if tag in {"script", "style", "noscript", "svg"}: self._skip += 1 if tag == "meta": key = values.get("property") or values.get("name") content = values.get("content") if key and content: self.meta[key] = content def handle_endtag(self, tag: str) -> None: if tag == "title": self._in_title = False if tag in {"script", "style", "noscript", "svg"} and self._skip: self._skip -= 1 def handle_data(self, data: str) -> None: value = data.strip() if self._in_title and value: self.title += value if not self._skip and value: self.parts.append(value) def _public_id() -> str: chars = string.ascii_uppercase + string.digits 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) host = (parsed.hostname or "").lower() if parsed.scheme != "https" or not any(host.endswith(suffix) for suffix in ALLOWED_HOST_SUFFIXES): raise TaskError("INVALID_FEISHU_URL", "仅支持公开的 https://*.feishu.cn 或 https://*.larksuite.com 链接", 422) return url def _source_type(url: str) -> str: path = urlparse(url).path for value in ("docx", "wiki", "sheets", "base", "file"): if f"/{value}/" in path: return value return "unknown" def _clean_xhs_url(value: str) -> str: return value.rstrip(".,;:!?，。；：！？、)]}）】》〉\"").replace("http://www.xiaohongshu.com", "https://www.xiaohongshu.com") def _extract_sheet_with_browser(url: str) -> tuple[str, list[dict[str, str]]]: """渲染公开飞书 Sheet，优先读取完整工作表模型，复制仅作为回退。""" timeout_ms = max(15, Config.ACCEPTANCE_BROWSER_TIMEOUT_SECONDS) * 1000 with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) context = browser.new_context( viewport={"width": 1600, "height": 1100}, locale="zh-CN", permissions=["clipboard-read", "clipboard-write"], ) page = context.new_page() page.goto(url, wait_until="domcontentloaded", timeout=timeout_ms) page.wait_for_timeout(min(30_000, timeout_ms // 2)) title = re.sub(r"\s*-\s*飞书云文档\s*$", "", page.title()).strip() # 飞书的 Ctrl/Command+A 复制会跳过筛选或隐藏行，导致实际 150 条只 # 得到 99 条。页面的 spread 数据模型已经包含完整工作表，直接按 # sheet id 读取可覆盖筛选行，同时不依赖 Canvas 当前渲染范围。 model_rows = page.evaluate( """() => { const workbook = window.spread && window.spread.toJSON ? window.spread.toJSON() : null; if (!workbook || !workbook.sheets) return null; const requestedId = new URL(location.href).searchParams.get('sheet'); const sheets = Object.values(workbook.sheets); const sheet = sheets.find(item => String(item.id) === String(requestedId)) || sheets[window.spread._activeSheetIndex || 0] || sheets[0]; if (!sheet || !sheet.data || !sheet.data.dataTable) return null; const table = sheet.data.dataTable; const rowCount = Number(sheet.rowCount || 0); const columnCount = Number(sheet.columnCount || 0); const valueOf = cell => { if (!cell) return ''; let value = cell.value; if (value === undefined || value === null) value = cell.formulaResult; if (value === undefined || value === null) value = cell.displayValue; if (value === undefined || value === null) return ''; if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') { return String(value); } if (Array.isArray(value)) { return value.map(item => item && (item.text ?? item.content ?? item.value ?? '')).join(''); } return String(value.text ?? value.content ?? value.value ?? ''); }; return Array.from({length: rowCount}, (_, row) => Array.from({length: columnCount}, (_, column) => valueOf(table[String(row)] && table[String(row)][String(column)]) ) ); }""" ) if isinstance(model_rows, list) and len(model_rows) >= 2: rows = model_rows browser.close() else: rows = None canvases = page.locator("canvas") if rows is None: candidates: list[tuple[float, int, dict[str, float]]] = [] for index in range(canvases.count()): box = canvases.nth(index).bounding_box() if box and box["width"] > 300 and box["height"] > 200 and box["x"] >= 0 and box["y"] >= 0: candidates.append((box["width"] * box["height"], index, box)) if not candidates: browser.close() raise RuntimeError("未找到可读取的飞书表格区域，请确认链接公开且指向 Sheet") _, _, box = max(candidates) page.mouse.click(box["x"] + min(150, box["width"] / 2), box["y"] + min(100, box["height"] / 2)) modifier = "Meta" if platform.system() == "Darwin" else "Control" page.keyboard.press(f"{modifier}+A") page.wait_for_timeout(300) page.keyboard.press(f"{modifier}+C") page.wait_for_timeout(1000) clipboard = page.evaluate("navigator.clipboard.readText()") browser.close() if not isinstance(clipboard, str) or "\t" not in clipboard: raise RuntimeError("飞书表格已打开，但没有复制到有效单元格数据") rows = list(csv.reader(io.StringIO(clipboard), delimiter="\t")) if len(rows) < 2: raise RuntimeError("飞书表格没有可解析的数据行") width = max(len(row) for row in rows) headers: list[str] = [] for index in range(width): value = rows[0][index].strip() if index < len(rows[0]) else "" headers.append(value or f"列{index + 1}") records = [ {headers[index]: (row[index] if index < len(row) else "") for index in range(width)} for row in rows[1:] if any(str(cell).strip() for cell in row) and not ( not str(row[0] if row else "").strip() and any(str(cell).strip() == "合计" for cell in row) ) ] return title or "未命名飞书表格", records def _store_records_and_notes(task_id: int, source_id: int, records: list[dict[str, str]]) -> list[int]: now = now_iso() note_ids: list[int] = [] with connection() as conn: conn.execute("DELETE FROM acceptance_feishu_records WHERE source_id=?", (source_id,)) for row_number, record in enumerate(records, start=2): conn.execute( "INSERT INTO acceptance_feishu_records(source_id,row_number,row_json,created_at) VALUES (?,?,?,?)", (source_id, row_number, json.dumps(record, ensure_ascii=False), now), ) for column_name, raw_value in record.items(): for match in XHS_URL_PATTERN.findall(str(raw_value)): note_url = _clean_xhs_url(match) cursor = conn.execute( """INSERT OR IGNORE INTO acceptance_xhs_notes (task_id,source_id,source_row_number,source_column_name,note_url,normalized_url, parse_status,created_at,updated_at) VALUES (?,?,?,?,?,?,'waiting',?,?)""", (task_id, source_id, row_number, column_name, note_url, note_url, now, now), ) if cursor.rowcount: note_ids.append(cursor.lastrowid) existing = conn.execute( "SELECT id FROM acceptance_xhs_notes WHERE source_id=? AND parse_status IN ('waiting','failed')", (source_id,), ).fetchall() note_ids.extend(row["id"] for row in existing if row["id"] not in note_ids) return note_ids def _parse_note(note_pk: int) -> bool: with connection() as conn: row = conn.execute("SELECT * FROM acceptance_xhs_notes WHERE id=?", (note_pk,)).fetchone() if not row: return False conn.execute("UPDATE acceptance_xhs_notes SET parse_status='running',updated_at=? WHERE id=?", (now_iso(), note_pk)) note_url = row["note_url"] try: result = fetch_content_detail(note_url) note_id = None match = re.search(r"/(?:explore|discovery/item)/([0-9a-f]{16,})", note_url, re.IGNORECASE) if match: note_id = match.group(1) with connection() as conn: note_row = conn.execute("SELECT * FROM acceptance_xhs_notes WHERE id=?", (note_pk,)).fetchone() if not note_row: return False upsert_feishu_note(conn, note_row, result) conn.execute( """UPDATE acceptance_xhs_notes SET parse_status='completed',note_id=?,parsed_json=NULL, error_message=NULL,parsed_at=?,updated_at=? WHERE id=?""", (note_id or result.get("noteId") or result.get("note_id"), now_iso(), now_iso(), note_pk), ) return True except Exception as exc: with connection() as conn: conn.execute( """UPDATE acceptance_xhs_notes SET parse_status='failed',error_message=?,parsed_at=?,updated_at=? WHERE id=?""", (str(exc)[:2000], now_iso(), now_iso(), note_pk), ) return False def _parse_notes(note_ids: list[int], progress_callback=None) -> tuple[int, int]: if not note_ids: return 0, 0 success = 0 with ThreadPoolExecutor(max_workers=max(1, Config.ACCEPTANCE_NOTE_WORKERS)) as pool: futures = [pool.submit(_parse_note, note_id) for note_id in note_ids] for completed_count, future in enumerate(as_completed(futures), start=1): success += int(future.result()) if progress_callback: progress_callback(completed_count, len(note_ids)) return success, len(note_ids) - success def _keywords_from_records(records: list[dict[str, str]]) -> list[str]: values: list[str] = [] for record in records: for column, raw_value in record.items(): normalized_column = re.sub(r"\s+", "", str(column)) if not any(alias in normalized_column for alias in ("绑定核心词", "核心词", "搜索词", "关键词")): continue for value in re.split(r"[\n,，、;；]+", str(raw_value)): value = value.strip() if value and value not in values: values.append(value) return values def _sync_acceptance_keywords(task_id: int, discovered: list[str]) -> list[str]: with connection() as conn: existing = [row["keyword"] for row in conn.execute( "SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (task_id,) ).fetchall()] # 用户显式填写的关键词是唯一执行范围。只有表单完全未填写时， # 才使用飞书“绑定核心词”等列作为兜底，避免一词任务被扩成多词。 keywords = existing or list(dict.fromkeys(discovered)) now = now_iso() for index, keyword in enumerate(keywords): conn.execute( "INSERT OR IGNORE INTO acceptance_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", (task_id, index, keyword, now), ) return keywords def _create_keyword_runs(task_id: int, method: str, keywords: list[str]) -> list[int]: now = now_iso() run_ids: list[int] = [] with connection() as conn: for index, keyword in enumerate(keywords): cursor = conn.execute( """INSERT OR IGNORE INTO acceptance_keyword_runs (task_id,keyword,run_type,run_order,status,created_at,updated_at) VALUES (?,?,?,?,'waiting',?,?)""", (task_id, keyword, method, index, now, now), ) if cursor.rowcount: run_ids.append(cursor.lastrowid) else: existing = conn.execute( """SELECT id FROM acceptance_keyword_runs WHERE task_id=? AND run_type=? AND keyword=?""", (task_id, method, keyword), ).fetchone() if existing: run_ids.append(existing["id"]) return run_ids def _update_run(run_id: int, *, status: str, result: dict[str, Any] | None = None, crawler_task_id: str | None = None, error: str | None = None, completed: bool = False) -> None: now = now_iso() with connection() as conn: conn.execute( """UPDATE acceptance_keyword_runs SET status=?,crawler_task_id=COALESCE(?,crawler_task_id), result_json=COALESCE(?,result_json),error_message=?,started_at=COALESCE(started_at,?), completed_at=CASE WHEN ? THEN ? ELSE completed_at END,updated_at=? WHERE id=?""", (status, crawler_task_id, json.dumps(result, ensure_ascii=False) if result is not None else None, error, now, completed, now, now, run_id), ) 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, } def _wait_collection_result(result: dict[str, Any]) -> tuple[dict[str, Any], str | None, str]: crawler_id = str(result.get("task_id") or "") or None status = str(result.get("status") or "") if crawler_id and status in {"waiting", "running", "queued", ""}: while True: time.sleep(Config.CRAWLER_POLL_INTERVAL_SECONDS) result = query_collection(crawler_id) status = str(result.get("status") or "running") if status in {"completed", "partial", "failed", "cancelled"}: break return result, crawler_id, status def _start_cookie_attempt( task_id: int, run_id: int, keyword: str, credential_id: str, ) -> tuple[int, str]: started_at = now_iso() with connection() as conn: next_order = conn.execute( """SELECT COALESCE(MAX(resource_order),-1)+1 AS value FROM acceptance_task_resources WHERE task_id=?""", (task_id,), ).fetchone()["value"] conn.execute( """INSERT OR IGNORE INTO acceptance_task_resources (task_id,resource_type,resource_id,resource_order,created_at) VALUES (?,'cookie',?,?,?)""", (task_id, credential_id, next_order, started_at), ) attempt_number = conn.execute( """SELECT COALESCE(MAX(attempt_number),0)+1 AS value FROM acceptance_cookie_attempts WHERE keyword_run_id=? AND credential_id=?""", (run_id, credential_id), ).fetchone()["value"] cursor = conn.execute( """INSERT INTO acceptance_cookie_attempts (task_id,keyword_run_id,keyword,credential_id,attempt_number,status, crawler_task_id,result_json,error_message,started_at,completed_at,created_at) VALUES (?,?,?,?,?,'running',NULL,NULL,NULL,?,?,?)""", ( task_id, run_id, keyword, credential_id, attempt_number, started_at, started_at, started_at, ), ) return cursor.lastrowid, started_at def _finish_cookie_attempt( attempt_id: int, *, status: str, crawler_task_id: str | None = None, result: dict[str, Any] | None = None, error: str | None = None, ) -> None: with connection() as conn: conn.execute( """UPDATE acceptance_cookie_attempts SET status=?,crawler_task_id=?, result_json=?,error_message=?,completed_at=? WHERE id=?""", ( status, crawler_task_id, json.dumps(result, ensure_ascii=False) if result is not None else None, error, now_iso(), attempt_id, ), ) def _run_keyword_with_cookie_failover( task_id: int, run_id: int, keyword: str, target_count: int, ) -> bool: """补足目标成功数；成功 Cookie 不重跑，失败时自动尝试池中其他 Cookie。""" 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]): conn.execute( """INSERT OR IGNORE INTO acceptance_task_resources (task_id,resource_type,resource_id,resource_order,created_at) VALUES (?,'cookie',?,?,?)""", (task_id, credential_id, index, now), ) planned_ids = [row["resource_id"] for row in conn.execute( """SELECT resource_id FROM acceptance_task_resources WHERE task_id=? AND resource_type='cookie' ORDER BY resource_order""", (task_id,), ).fetchall()] successful_ids = { row["credential_id"] for row in conn.execute( """SELECT DISTINCT credential_id FROM acceptance_cookie_attempts WHERE keyword_run_id=? AND status='completed'""", (run_id,), ).fetchall() } attempted_now: set[str] = set() crawler_ids: list[str] = [] _update_run(run_id, status="running", error=None) candidate_ids = list(dict.fromkeys([*planned_ids, *available_ids])) for credential_id in candidate_ids: if len(successful_ids) >= target_count: break if credential_id in successful_ids or credential_id in attempted_now: continue attempted_now.add(credential_id) attempt_id, _ = _start_cookie_attempt( task_id, run_id, keyword, credential_id, ) try: result, crawler_id, remote_status = _wait_collection_result( collect_keyword_sources( keyword, None if custom_cookie else credential_id, custom_cookie or None, ) ) if crawler_id: crawler_ids.append(crawler_id) succeeded = remote_status in {"completed", "partial", ""} _finish_cookie_attempt( attempt_id, status="completed" if succeeded else "failed", crawler_task_id=crawler_id, result=result, error=None if succeeded else str(result.get("error_message") or "采集失败")[:2000], ) if succeeded: successful_ids.add(credential_id) except Exception as exc: _finish_cookie_attempt( attempt_id, status="failed", error=str(exc)[:2000], ) with connection() as conn: attempts = conn.execute( """SELECT credential_id,attempt_number,status,crawler_task_id,result_json, error_message,started_at,completed_at FROM acceptance_cookie_attempts WHERE keyword_run_id=? ORDER BY id""", (run_id,), ).fetchall() attempt_items = [] for attempt in attempts: item = dict(attempt) item["result"] = json.loads(item.pop("result_json")) if item.get("result_json") else None attempt_items.append(item) completed = len(successful_ids) >= target_count _update_run( run_id, status="completed" if completed else "failed", result={ "keyword": keyword, "target_cookie_count": target_count, "successful_cookie_count": len(successful_ids), "attempts": attempt_items, }, crawler_task_id=",".join(crawler_ids) or None, error=None if completed else f"可用 Cookie 已尝试完，成功 {len(successful_ids)}/{target_count}", completed=True, ) return completed def _execute_acceptance_method(task_id: int, method: str, keywords: list[str], resource_count: int) -> int: if not keywords: raise RuntimeError("未在表单或飞书表格中找到可执行的关键词") run_ids = _create_keyword_runs(task_id, method, keywords) 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, 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: _update_run(run_id, status="failed", error=str(exc)[:2000], completed=True) return failures resources = get_wen_resources() available_ids = list(dict.fromkeys( str(item).strip() for item in resources.get("credential_ids", []) if str(item).strip() )) if len(available_ids) < resource_count: raise RuntimeError(f"请求使用 {resource_count} 个 Cookie，但当前仅有 {len(available_ids)} 个可用") with connection() as conn: now = now_iso() for index, credential_id in enumerate(available_ids[:resource_count]): conn.execute( """INSERT OR IGNORE INTO acceptance_task_resources (task_id,resource_type,resource_id,resource_order,created_at) VALUES (?,'cookie',?,?,?)""", (task_id, credential_id, index, now), ) for run_id, keyword in zip(run_ids, keywords): try: if not _run_keyword_with_cookie_failover( task_id, run_id, keyword, resource_count, ): failures += 1 except Exception as exc: failures += 1 _update_run(run_id, status="failed", error=str(exc)[:2000], completed=True) 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: if _acceptance_task_is_cancelled(public_id): return 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: current_task = conn.execute( "SELECT status FROM acceptance_tasks WHERE id=?", (task["id"],), ).fetchone() if not current_task or current_task["status"] == "cancelled": return rows = conn.execute( "SELECT * FROM acceptance_xapi_runs WHERE task_id=?", (task["id"],), ).fetchall() tracked_rows = [row for row in rows if row["batch_id"] in batch_map] 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 tracked_rows ) feishu_ready = all( str(row["status"] or "").lower() != "completed" or str(row["feishu_sync_status"] or "").lower() == "succeeded" for row in tracked_rows ) received = min(len(tracked_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 tracked_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: latest_rows: dict[tuple[str, str], Any] = {} for row in rows: key = (str(row["keyword"] or ""), str(row["xhs_account_id"] or "")) latest_rows[key] = row failed_count = sum( str(row["status"] or "").lower() != "completed" for row in latest_rows.values() ) 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 retry_failed_acceptance_devices(public_id: str) -> 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) if task["acceptance_method"] not in {"screenshot", "combined"}: raise TaskError("SCREENSHOT_RETRY_NOT_SUPPORTED", "该任务不包含截图采集", 409) rows = conn.execute( """SELECT * FROM acceptance_xapi_runs WHERE task_id=? ORDER BY id""", (task["id"],), ).fetchall() keyword_runs = { row["keyword"]: row["id"] for row in conn.execute( """SELECT id,keyword FROM acceptance_keyword_runs WHERE task_id=? AND run_type='screenshot'""", (task["id"],), ).fetchall() } latest: dict[tuple[str, str], Any] = {} for row in rows: keyword = str(row["keyword"] or "").strip() account_id = str(row["xhs_account_id"] or "").strip() if keyword and account_id: latest[(keyword, account_id)] = row failed_by_keyword: dict[str, list[str]] = {} terminal_failures = {"failed", "cancelled", "canceled", "error"} for (keyword, account_id), row in latest.items(): status = str(row["status"] or "").lower() artifacts = _decode_json(row["artifacts_json"], []) or [] has_screenshot = any( str(item.get("image_url") or item.get("url") or "").strip() for item in artifacts if isinstance(item, dict) ) if status in terminal_failures or (status == "completed" and not has_screenshot): failed_by_keyword.setdefault(keyword, []).append(account_id) if not failed_by_keyword: raise TaskError("NO_FAILED_SCREENSHOT_DEVICES", "当前没有可重新采集的失败设备", 409) dispatched_devices: set[str] = set() dispatched_pairs = 0 errors: list[str] = [] for keyword, account_ids in failed_by_keyword.items(): run_id = keyword_runs.get(keyword) if not run_id: errors.append(f"{keyword}: 缺少关键词运行记录") continue try: result = schedule_xapi_screenshots( [keyword], trigger_source="acceptance", xhs_account_ids=account_ids, ) batches = result.get("batches") or [] batch_ids = [ str(batch.get("batch_id") or "").strip() for batch in batches if str(batch.get("batch_id") or "").strip() ] if not batch_ids: raise RuntimeError("Framework 未返回可回查的 batch_id") _update_run( run_id, status="waiting", result={**result, "batches": batches}, crawler_task_id=",".join(batch_ids), error=None, ) dispatched_devices.update(account_ids) dispatched_pairs += len(account_ids) except Exception as exc: errors.append(f"{keyword}: {exc}") if not dispatched_pairs: raise TaskError("FAILED_DEVICE_RETRY_DISPATCH_FAILED", "；".join(errors)[:2000], 502) with connection() as conn: conn.execute( """UPDATE acceptance_tasks SET status='running',progress_percent=70, error_message=?,updated_at=? WHERE id=?""", ("；".join(errors)[:2000] if errors else None, now_iso(), task["id"]), ) threading.Thread( target=_poll_screenshot_results, args=(public_id,), daemon=True, name=f"acceptance-retry-failed-{public_id}", ).start() result_task = get_acceptance_task(public_id) result_task["retry_dispatched_device_count"] = len(dispatched_devices) result_task["retry_dispatched_pair_count"] = dispatched_pairs result_task["retry_dispatch_errors"] = errors return result_task 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", []))) method = str(data.get("acceptance_method") or "") if method not in METHODS: raise TaskError("INVALID_ACCEPTANCE_METHOD", "验收方式无效", 422) 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)), } 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 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", f"{resource_name}数量需要在 1–{available_count} 之间", 422, ) name = str(data.get("name") or "").strip() or f"项目验收｜{keywords[0] if keywords else '飞书资料'}" now = now_iso() with connection() as conn: public_id = _public_id() 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,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("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)], ) 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) VALUES (?,'cookie',?,?,?)""", [(task_id, resource_id, index, now) for index, resource_id in enumerate(credential_ids)], ) conn.executemany( """INSERT INTO acceptance_feishu_sources (task_id,source_order,source_url,source_type,fetch_status,created_at,updated_at) VALUES (?,?,?,?,'waiting',?,?)""", [(task_id, index, url, _source_type(url), now, now) for index, url in enumerate(links)], ) threading.Thread(target=_fetch_sources, args=(public_id,), daemon=True, name=f"acceptance-{public_id}").start() return get_acceptance_task(public_id) def _normalize_schedule_weekdays(value: Any) -> list[int]: raw = value if isinstance(value, (list, tuple, set)) else [value] try: weekdays = sorted(set(int(item) for item in raw)) except (TypeError, ValueError) as exc: raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422) from exc if not weekdays or any(item not in WEEKDAYS for item in weekdays): raise TaskError("INVALID_WEEKDAY", "采集周期至少选择一个周一至周日", 422) return weekdays def _normalize_schedule_time(value: Any, field_name: str) -> str: text = str(value or "").strip() try: parsed = datetime.strptime(text, "%H:%M") except ValueError as exc: raise TaskError("INVALID_SCHEDULE_TIME", f"{field_name}必须为 HH:MM 格式", 422) from exc return parsed.strftime("%H:%M") def _next_weekly_run( weekdays: list[int] | tuple[int, ...] | set[int] | int, start_time: str = "00:00", *, after: datetime | None = None, ) -> str: normalized_weekdays = _normalize_schedule_weekdays(weekdays) normalized_start = _normalize_schedule_time(start_time, "开始时间") current = after or datetime.now().astimezone() hour, minute = (int(item) for item in normalized_start.split(":")) for offset in range(0, 8): candidate = (current + timedelta(days=offset)).replace( hour=hour, minute=minute, second=0, microsecond=0, ) if candidate.isoweekday() in normalized_weekdays and candidate > current: return candidate.isoformat(timespec="seconds") raise RuntimeError("无法计算下一次周期采集时间") def update_acceptance_schedule( public_id: str, weekdays: Any, start_time: Any = "00:00", end_time: Any = "23:59", ) -> dict[str, Any]: normalized_weekdays = _normalize_schedule_weekdays(weekdays) normalized_start = _normalize_schedule_time(start_time, "开始时间") normalized_end = _normalize_schedule_time(end_time, "结束时间") if normalized_start > normalized_end: raise TaskError("INVALID_SCHEDULE_RANGE", "开始时间不能晚于结束时间", 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=?, schedule_weekdays=?,schedule_start_time=?,schedule_end_time=?, next_run_at=?,schedule_error=NULL,updated_at=? WHERE id=?""", ( normalized_weekdays[0], json.dumps(normalized_weekdays), normalized_start, normalized_end, _next_weekly_run(normalized_weekdays, normalized_start), 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 _acceptance_task_is_cancelled(public_id: str) -> bool: with connection() as conn: row = conn.execute( "SELECT status FROM acceptance_tasks WHERE public_task_id=?", (public_id,), ).fetchone() return not row or row["status"] == "cancelled" def cancel_acceptance_task(public_id: str, reason: str = "") -> dict[str, Any]: """只停止当前批次的处理；该项目已有的周期调度保持不变。""" current_time = now_iso() with connection() as conn: task = conn.execute( "SELECT id,status FROM acceptance_tasks WHERE public_task_id=?", (public_id,), ).fetchone() if not task: raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404) if task["status"] == "cancelled": return get_acceptance_task(public_id) if task["status"] not in {"waiting", "running"}: raise TaskError("ACCEPTANCE_TASK_NOT_CANCELLABLE", "任务状态已变化，无法停止", 409) conn.execute( """UPDATE acceptance_tasks SET status='cancelled', error_message=?,updated_at=? WHERE id=?""", ((reason or "用户主动停止验收任务")[:500], current_time, task["id"]), ) conn.execute( """UPDATE acceptance_keyword_runs SET status='cancelled',error_message=?, completed_at=?,updated_at=? WHERE task_id=? AND status IN ('waiting','running')""", ((reason or "用户主动停止验收任务")[:500], current_time, current_time, task["id"]), ) 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: current = datetime.now().astimezone() now = current.isoformat(timespec="seconds") with connection() as conn: due = conn.execute( """SELECT public_task_id,schedule_weekday,schedule_weekdays, schedule_start_time,schedule_end_time,next_run_at 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() runnable = [] for row in due: weekdays = _decode_json(row["schedule_weekdays"], None) or [row["schedule_weekday"]] start_time = row["schedule_start_time"] or "00:00" end_time = row["schedule_end_time"] or "23:59" due_at = datetime.fromisoformat(row["next_run_at"]) end_hour, end_minute = (int(item) for item in end_time.split(":")) window_end = due_at.replace(hour=end_hour, minute=end_minute, second=59) missed = current > window_end conn.execute( """UPDATE acceptance_tasks SET next_run_at=?,last_run_at=?, schedule_error=?,updated_at=? WHERE public_task_id=?""", ( _next_weekly_run(weekdays, start_time, after=current), None if missed else now, "本次调度已超过设定结束时间，自动顺延" if missed else None, now, row["public_task_id"], ), ) if not missed: runnable.append(row) for row in runnable: 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(runnable) 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() if not task: return sources = conn.execute("SELECT * FROM acceptance_feishu_sources WHERE task_id=? ORDER BY source_order", (task["id"],)).fetchall() initial_keywords = [row["keyword"] for row in conn.execute( "SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (task["id"],), ).fetchall()] if task["status"] == "cancelled": return conn.execute("UPDATE acceptance_tasks SET status='running',progress_percent=5,updated_at=? WHERE id=?", (now_iso(), task["id"])) # 数据分析时，表单中已知的关键词无需等待飞书解析完成，立即并行采集。 analysis_pool = None initial_analysis_future = None 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"], "data-analysis", initial_keywords, task["cookie_count"] if task["acceptance_method"] == "combined" else task["device_count"], ) failures = 0 note_failures = 0 discovered_keywords: list[str] = [] for index, source in enumerate(sources): if _acceptance_task_is_cancelled(public_id): return try: response = requests.get(source["source_url"], timeout=25, allow_redirects=True, headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/124 Safari/537.36", "Accept-Language": "zh-CN,zh;q=0.9", }) response.raise_for_status() parser = _ReadableHTML() parser.feed(response.text) title = parser.meta.get("og:title") or parser.title or "未命名飞书资料" description = parser.meta.get("og:description") or parser.meta.get("description") or "" text = "\n".join(dict.fromkeys(unescape(part) for part in parser.parts)) text = re.sub(r"\n{3,}", "\n\n", text).strip() records: list[dict[str, str]] = [] if source["source_type"] in {"wiki", "sheets", "base"}: rendered_title, records = _extract_sheet_with_browser(source["source_url"]) title = rendered_title or title text = "\n".join( "\t".join(str(value) for value in record.values()) for record in records ) for keyword in _keywords_from_records(records): if keyword not in discovered_keywords: discovered_keywords.append(keyword) payload = { "title": title, "description": description, "text": text, "meta": parser.meta, "records": records, # 保存公开页面原始响应，便于后续针对飞书不同文档类型继续做结构化解析。 "raw_html": response.text[:5_000_000], } with connection() as conn: conn.execute( """UPDATE acceptance_feishu_sources SET final_url=?,title=?,content_text=?,content_json=?, fetch_status='completed',http_status=?,fetched_at=?,updated_at=? WHERE id=?""", (response.url, title[:500], text, json.dumps(payload, ensure_ascii=False), response.status_code, now_iso(), now_iso(), source["id"]), ) note_ids = _store_records_and_notes(task["id"], source["id"], records) progress_floor = 5 + round((index / max(1, len(sources))) * 55) progress_ceiling = 5 + round(((index + 1) / max(1, len(sources))) * 55) last_note_progress = [progress_floor] def update_note_progress(done: int, total: int) -> None: progress_value = progress_floor + round( (progress_ceiling - progress_floor) * done / max(1, total) ) if progress_value <= last_note_progress[0]: return last_note_progress[0] = progress_value with connection() as progress_conn: progress_conn.execute( """UPDATE acceptance_tasks SET progress_percent=?,updated_at=? WHERE public_task_id=?""", (progress_value, now_iso(), public_id), ) _, failed_notes = _parse_notes(note_ids, update_note_progress) note_failures += failed_notes except Exception as exc: failures += 1 with connection() as conn: conn.execute( """UPDATE acceptance_feishu_sources SET fetch_status='failed',error_message=?, fetched_at=?,updated_at=? WHERE id=?""", (str(exc)[:1000], now_iso(), now_iso(), source["id"]), ) progress = 5 + round(((index + 1) / max(1, len(sources))) * 55) with connection() as conn: 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 _acceptance_task_is_cancelled(public_id): return if not sources or failures < len(sources): try: keywords = _sync_acceptance_keywords(task["id"], discovered_keywords) with connection() as conn: conn.execute("UPDATE acceptance_tasks SET progress_percent=65,updated_at=? WHERE id=?", (now_iso(), task["id"])) if initial_analysis_future is not None: run_failures += initial_analysis_future.result() new_keywords = [keyword for keyword in keywords if keyword not in initial_keywords] if new_keywords: run_failures += _execute_acceptance_method( 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( task["id"], task["acceptance_method"], keywords, task["device_count"] ) except Exception as exc: run_failures = 1 run_error = str(exc) finally: if analysis_pool is not None: analysis_pool.shutdown(wait=False) elif initial_analysis_future is not None: try: run_failures = initial_analysis_future.result() except Exception as exc: run_failures = 1 run_error = str(exc) 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: if _acceptance_task_is_cancelled(public_id): return 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, "部分关键词下发或采集失败" if run_failures and not run_error else None, run_error, ])) or None with connection() as conn: conn.execute( """UPDATE acceptance_tasks SET status=?,progress_percent=100,error_message=?,updated_at=? WHERE public_task_id=? AND status!='cancelled'""", (status, error, now_iso(), public_id), ) def resume_acceptance_task(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) if task["acceptance_method"] != "data-analysis": raise TaskError("RESUME_NOT_SUPPORTED", "只有数据分析任务支持 Cookie 补采", 422) if task["status"] not in {"failed", "partial"}: raise TaskError("TASK_NOT_RESUMABLE", "当前任务状态不需要继续采集", 409) conn.execute( """UPDATE acceptance_tasks SET status='running',progress_percent=65, error_message=NULL,updated_at=? WHERE id=?""", (now_iso(), task["id"]), ) threading.Thread( target=_resume_data_analysis_worker, args=(public_id,), daemon=True, name=f"acceptance-resume-{public_id}", ).start() return get_acceptance_task(public_id) def resume_unfinished_acceptance_tasks() -> None: """服务重启后恢复等待中或运行中的验收任务。""" with connection() as conn: 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 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=target, args=(public_id,), daemon=True, name=f"acceptance-recover-{public_id}", ).start() def _resume_data_analysis_worker(public_id: str) -> None: with connection() as conn: task = conn.execute( "SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,), ).fetchone() if not task: return keywords = [row["keyword"] for row in conn.execute( "SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (task["id"],), ).fetchall()] _create_keyword_runs(task["id"], "data-analysis", keywords) with connection() as conn: runs = conn.execute( """SELECT id,keyword FROM acceptance_keyword_runs WHERE task_id=? AND run_type='data-analysis' ORDER BY run_order""", (task["id"],), ).fetchall() failures = 0 for run in runs: if _acceptance_task_is_cancelled(public_id): return try: if not _run_keyword_with_cookie_failover( task["id"], run["id"], run["keyword"], task["device_count"], ): failures += 1 except Exception as exc: failures += 1 _update_run(run["id"], status="failed", error=str(exc)[:2000], completed=True) with connection() as conn: source_failures = conn.execute( """SELECT COUNT(*) AS value FROM acceptance_feishu_sources WHERE task_id=? AND fetch_status='failed'""", (task["id"],), ).fetchone()["value"] note_failures = conn.execute( """SELECT COUNT(*) AS value FROM acceptance_xhs_notes WHERE task_id=? AND parse_status='failed'""", (task["id"],), ).fetchone()["value"] status = "completed" if not failures and not source_failures and not note_failures else "partial" error = None if status == "completed" else "Cookie 补采后仍有部分数据未完成" conn.execute( """UPDATE acceptance_tasks SET status=?,progress_percent=100, error_message=?,updated_at=? WHERE id=? AND status!='cancelled'""", (status, error, now_iso(), task["id"]), ) 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, (SELECT COUNT(*) FROM acceptance_feishu_records r WHERE r.source_id=s.id) AS record_count, (SELECT COUNT(*) FROM acceptance_xhs_notes n WHERE n.source_id=s.id) AS note_count FROM acceptance_feishu_sources s WHERE task_id=? ORDER BY source_order""", (task["id"],), ).fetchall()] note_stats = conn.execute( """SELECT COUNT(*) AS total, SUM(CASE WHEN parse_status='completed' THEN 1 ELSE 0 END) AS completed, SUM(CASE WHEN parse_status='failed' THEN 1 ELSE 0 END) AS failed FROM acceptance_xhs_notes WHERE task_id=?""", (task["id"],), ).fetchone() keyword_runs = [dict(row) for row in conn.execute( """SELECT r.keyword,r.run_type,r.run_order,r.status,r.crawler_task_id, r.error_message,r.started_at,r.completed_at, (SELECT COUNT(*) FROM acceptance_cookie_attempts a WHERE a.keyword_run_id=r.id) AS attempt_count, (SELECT COUNT(DISTINCT credential_id) FROM acceptance_cookie_attempts a WHERE a.keyword_run_id=r.id AND a.status='completed') AS successful_cookie_count FROM acceptance_keyword_runs r WHERE r.task_id=? ORDER BY r.run_order""", (task["id"],), ).fetchall()] task_resources = conn.execute( """SELECT resource_type,resource_id,resource_order FROM acceptance_task_resources WHERE task_id=? ORDER BY resource_order""", (task["id"],), ).fetchall() cookie_attempts = conn.execute( """SELECT credential_id,keyword,status,attempt_number,error_message, started_at,completed_at 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": all_related_runs = [ row for row in xapi_rows if row["xhs_account_id"] == resource["resource_id"] ] latest_by_keyword: dict[str, Any] = {} for row in all_related_runs: latest_by_keyword[str(row["keyword"] or "")] = row related_runs = list(latest_by_keyword.values()) 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(all_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: statuses = [row["status"] for row in related if row["keyword"] == keyword] if "completed" in statuses: completed_keywords += 1 elif "running" in statuses: running_keywords += 1 elif "failed" in statuses: failed_keywords += 1 terminal_keywords = completed_keywords + failed_keywords total_keywords = len(keywords) if completed_keywords >= total_keywords and total_keywords: resource_status = "completed" elif running_keywords or (task["status"] == "running" and terminal_keywords < total_keywords): resource_status = "running" if related else "waiting" elif failed_keywords: resource_status = "failed" else: resource_status = "waiting" resource_progress.append({ "resource_type": resource["resource_type"], "resource_id": resource["resource_id"], "resource_order": resource["resource_order"], "status": resource_status, "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), "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"], "cookie_count": task["cookie_count"], "schedule_enabled": bool(task["schedule_enabled"]), "schedule_weekday": task["schedule_weekday"], "schedule_weekdays": _decode_json(task["schedule_weekdays"], None) or ( [task["schedule_weekday"]] if task["schedule_weekday"] else [] ), "schedule_start_time": task["schedule_start_time"] or "00:00", "schedule_end_time": task["schedule_end_time"] or "23:59", "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"], } def get_acceptance_task(public_id: str) -> 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) return _task_dict(conn, task) def list_acceptance_tasks() -> dict[str, Any]: with connection() as conn: rows = conn.execute("SELECT * FROM acceptance_tasks ORDER BY created_at DESC").fetchall() items = [_task_dict(conn, row) for row in rows] return {"items": items, "total": len(items)} 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() if not task: raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404) rows = conn.execute( """SELECT n.source_row_number,n.source_column_name,n.note_url,n.note_id,n.parse_status, f.title,f.body,f.author_nickname,f.author_avatar,f.likes,f.collects,f.comments, f.images_json,f.tags_json,f.raw_json,n.error_message,n.parsed_at FROM acceptance_xhs_notes n LEFT JOIN feishu_notes f ON (n.note_id IS NOT NULL AND f.note_id=n.note_id) OR f.normalized_url=n.normalized_url WHERE n.task_id=? ORDER BY source_row_number,n.id""", (task["id"],), ).fetchall() items = [] for row in rows: item = dict(row) item["images"] = json.loads(item.pop("images_json")) if item.get("images_json") else [] item["tags"] = json.loads(item.pop("tags_json")) if item.get("tags_json") else [] item["parsed_data"] = json.loads(item.pop("raw_json")) if item.get("raw_json") else None items.append(item) return {"items": items, "total": len(items)} def get_acceptance_link_ai_data(public_id: str) -> dict[str, Any]: """返回 Link AI 所需的笔记、AI 采集结果和逐关键词比对结果。""" 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) task_data = _task_dict(conn, task) rows = conn.execute( """SELECT a.keyword,a.credential_id,a.attempt_number,a.crawler_task_id, a.result_json,a.completed_at FROM acceptance_cookie_attempts a WHERE a.task_id=? AND a.status='completed' AND a.result_json IS NOT NULL ORDER BY a.keyword,a.id""", (task["id"],), ).fetchall() ai_results = [] for row in rows: try: result = json.loads(row["result_json"]) except (TypeError, json.JSONDecodeError): continue ai_results.append({ "keyword": row["keyword"], "credential_id": row["credential_id"], "attempt_number": row["attempt_number"], "crawler_task_id": row["crawler_task_id"], "completed_at": row["completed_at"], "result": result, }) comparisons = [] for keyword in task_data["keywords"]: try: comparisons.append({ "keyword": keyword, "status": "completed", "data": match_feishu_notes(keyword), }) except Exception as exc: comparisons.append({ "keyword": keyword, "status": "failed", "error": str(exc), "data": None, }) return { "task": task_data, "notes": list_acceptance_notes(public_id), "ai_results": {"items": ai_results, "total": len(ai_results)}, "note_matches": {"items": comparisons, "total": len(comparisons)}, "data_scope": "仅包含当前已完成并已入库的数据", }