diff --git a/.env.example b/.env.example index 1e78487..4f38828 100644 --- a/.env.example +++ b/.env.example @@ -7,4 +7,6 @@ CRAWLER_RETRY_TIMES=3 CRAWLER_RETRY_DELAY_SECONDS=2 CRAWLER_TIMEOUT_SECONDS=120 CRAWLER_POLL_INTERVAL_SECONDS=5 +ACCEPTANCE_BROWSER_TIMEOUT_SECONDS=60 +ACCEPTANCE_NOTE_WORKERS=4 CORS_ORIGIN=http://localhost:5173 diff --git a/README.md b/README.md index 1452c1a..dc75937 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ cd "/Users/xiaoti/Downloads/Holy蟹搜索监测平台-后端服务-20260723" python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt +playwright install chromium cp .env.example .env python app.py ``` diff --git a/acceptance_service.py b/acceptance_service.py new file mode 100644 index 0000000..807fa39 --- /dev/null +++ b/acceptance_service.py @@ -0,0 +1,967 @@ +import json +import csv +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 +from html import unescape +from html.parser import HTMLParser +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, + query_collection, + 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"} +XHS_URL_PATTERN = re.compile( + r"https?://(?:www\.)?(?:xiaohongshu\.com|xhslink\.com)/[^\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 _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, int]: + resources = get_wen_resources() + return { + "cookie_count": max(0, int(resources.get("cookie_count") or 0)), + "device_count": max(0, int(resources.get("device_count") or 0)), + } + + +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。""" + resources = get_wen_resources() + available_ids = 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, credential_id) + ) + 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: + 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) + 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 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())) + raw_resources = get_wen_resources() + 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) + ) + available_count = resources["device_count" if method == "screenshot" else "cookie_count"] + resource_name = "设备" if method == "screenshot" else "Cookie" + 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) + 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), + ) + 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 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 _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()] + 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"] == "data-analysis" 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"], + ) + failures = 0 + note_failures = 0 + discovered_keywords: list[str] = [] + for index, source in enumerate(sources): + 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 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"], task["acceptance_method"], new_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) + 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=?", + (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: + task_ids = [row["public_task_id"] for row in conn.execute( + """SELECT public_task_id FROM acceptance_tasks + WHERE status IN ('waiting','running') ORDER BY created_at""" + ).fetchall()] + for public_id in task_ids: + threading.Thread( + target=_fetch_sources, 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: + 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=?""", + (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()] + 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() + resource_progress = [] + for resource in task_resources: + 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), + }) + 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], + "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, + "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)} diff --git a/app.py b/app.py index 19d1af2..222442d 100644 --- a/app.py +++ b/app.py @@ -13,6 +13,15 @@ from auth_service import ( ) from config import Config from database import init_database +from acceptance_service import ( + create_acceptance_task, + get_acceptance_resources, + get_acceptance_task, + list_acceptance_notes, + list_acceptance_tasks, + resume_acceptance_task, + resume_unfinished_acceptance_tasks, +) from rule_service import ( delete_rule_document, get_rule_document, @@ -37,6 +46,18 @@ def create_app() -> Flask: app = Flask(__name__) init_database() + def local_dev_user(): + if not Config.LOCAL_AUTH_BYPASS or request.remote_addr not in {"127.0.0.1", "::1"}: + return None + return { + "id": "local-dev", + "name": "本地开发用户", + "account": "local-dev", + "avatar": "", + "union_id": "", + "open_id": "", + } + @app.before_request def assign_request_id(): g.request_id = request.headers.get("X-Request-Id") or f"req_{secrets.token_hex(10)}" @@ -45,7 +66,7 @@ def create_app() -> Flask: open_paths = ("/health", "/api/v1/auth/", "/api/v1/skill/") if request.path == "/health" or any(request.path.startswith(path) for path in open_paths): return None - user = get_user_by_session_token(request.cookies.get(Config.SESSION_COOKIE_NAME)) + user = local_dev_user() or get_user_by_session_token(request.cookies.get(Config.SESSION_COOKIE_NAME)) if user is None and request.path.startswith("/api/"): return jsonify({ "error": {"code": "UNAUTHORIZED", "message": "请先通过钉钉授权登录"}, @@ -190,7 +211,7 @@ def create_app() -> Flask: @app.route("/api/v1/auth/me", methods=["GET"]) def auth_me(): - user = get_user_by_session_token(request.cookies.get(Config.SESSION_COOKIE_NAME)) + user = local_dev_user() or get_user_by_session_token(request.cookies.get(Config.SESSION_COOKIE_NAME)) if not user: raise TaskError("UNAUTHORIZED", "请先通过钉钉授权登录", 401) return jsonify({"data": {"user": user}, "request_id": g.request_id}) @@ -212,6 +233,35 @@ def create_app() -> Flask: task = create_task(payload) return jsonify({"data": task, "request_id": g.request_id}), 201 + @app.route("/api/v1/acceptance-tasks", methods=["POST"]) + def create_project_acceptance_task(): + payload = request.get_json(silent=True) or {} + current_user = getattr(g, "current_user", None) or {} + payload["created_by_id"] = current_user.get("id") or payload.get("created_by_id") + payload["created_by_name"] = current_user.get("name") or payload.get("created_by_name") + task = create_acceptance_task(payload) + return jsonify({"data": task, "request_id": g.request_id}), 201 + + @app.route("/api/v1/acceptance-resources", methods=["GET"]) + def query_project_acceptance_resources(): + return jsonify({"data": get_acceptance_resources(), "request_id": g.request_id}) + + @app.route("/api/v1/acceptance-tasks", methods=["GET"]) + def query_project_acceptance_tasks(): + return jsonify({"data": list_acceptance_tasks(), "request_id": g.request_id}) + + @app.route("/api/v1/acceptance-tasks/", methods=["GET"]) + def get_project_acceptance_task(task_id: str): + return jsonify({"data": get_acceptance_task(task_id), "request_id": g.request_id}) + + @app.route("/api/v1/acceptance-tasks//notes", methods=["GET"]) + def get_project_acceptance_notes(task_id: str): + return jsonify({"data": list_acceptance_notes(task_id), "request_id": g.request_id}) + + @app.route("/api/v1/acceptance-tasks//resume", methods=["POST"]) + 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/tasks", methods=["GET"]) def query_research_tasks(): result = list_tasks( @@ -286,6 +336,7 @@ def create_app() -> Flask: return jsonify({"data": delete_rule_document(), "request_id": g.request_id}) resume_unfinished_tasks() + resume_unfinished_acceptance_tasks() return app diff --git a/config.py b/config.py index c598330..26dda81 100644 --- a/config.py +++ b/config.py @@ -23,12 +23,16 @@ 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")) + ACCEPTANCE_BROWSER_TIMEOUT_SECONDS = int(os.getenv("ACCEPTANCE_BROWSER_TIMEOUT_SECONDS", "60")) + ACCEPTANCE_NOTE_WORKERS = int(os.getenv("ACCEPTANCE_NOTE_WORKERS", "4")) CORS_ORIGIN = os.getenv("CORS_ORIGIN", "http://localhost:5173") FRONTEND_BASE_URL = os.getenv("FRONTEND_BASE_URL", CORS_ORIGIN).rstrip("/") SESSION_COOKIE_NAME = os.getenv("SESSION_COOKIE_NAME", "hc_session") SESSION_COOKIE_SECURE = os.getenv("SESSION_COOKIE_SECURE", "true").lower() in ("1", "true", "yes", "on") SESSION_COOKIE_SAMESITE = os.getenv("SESSION_COOKIE_SAMESITE", "Lax") SESSION_EXPIRE_DAYS = int(os.getenv("SESSION_EXPIRE_DAYS", "14")) + # 仅供本机联调使用。生产环境不要设置或必须保持 false。 + LOCAL_AUTH_BYPASS = os.getenv("LOCAL_AUTH_BYPASS", "false").lower() in ("1", "true", "yes", "on") DINGTALK_CLIENT_ID = os.getenv("DINGTALK_CLIENT_ID", os.getenv("DINGTALK_APP_KEY", "")) DINGTALK_CLIENT_SECRET = os.getenv("DINGTALK_CLIENT_SECRET", os.getenv("DINGTALK_APP_SECRET", "")) DINGTALK_REDIRECT_URI = os.getenv( diff --git a/crawler_client.py b/crawler_client.py index 07c628a..d163757 100644 --- a/crawler_client.py +++ b/crawler_client.py @@ -78,6 +78,29 @@ def _post( return data +def _get(path: str) -> dict[str, Any]: + url = f"{Config.CRAWLER_BASE_URL}{path}" + try: + response = CRAWLER_SESSION.get(url, timeout=Config.CRAWLER_TIMEOUT_SECONDS) + response.raise_for_status() + payload = response.json() + except requests.RequestException as exc: + raise CrawlerError(f"爬虫服务请求失败:{exc}") from exc + except ValueError as exc: + raise CrawlerError("爬虫服务返回的不是有效 JSON") from exc + if payload.get("code") != 200 or not payload.get("success", False): + raise CrawlerError(payload.get("msg") or "爬虫服务返回失败") + data = payload.get("data") + if not isinstance(data, dict): + raise CrawlerError("爬虫服务响应缺少 data") + return data + + +def get_wen_resources() -> dict[str, Any]: + """查询当前可用 Cookie 和可立即下发截图的设备。""" + return _get("/api/v1/xhs/wen/resources") + + def start_collection(mode: str, keywords: list[str]) -> dict[str, Any]: """深度任务走 sources,轻度/常规任务走 overview。""" @@ -113,3 +136,38 @@ def resume_collection(crawler_task_id: str) -> dict[str, Any]: """沿用原爬虫任务 ID,从 framework 保存的检查点继续采集。""" return _post(f"/api/v1/xhs/wen/tasks/{crawler_task_id}/resume", {}) + + +def fetch_content_detail(note_url: str) -> dict[str, Any]: + """按小红书笔记链接读取完整笔记详情。""" + + return _post( + "/api/v1/fetch_content_detail", + {"plant": "xhs", "link": note_url}, + retry_transient=True, + ) + + +def schedule_xapi_screenshots(search_queries: list[str], device_count: int) -> dict[str, Any]: + """向手机集群下发问一问关键词搜索与截图任务。""" + + queries = normalize_collection_keywords(search_queries) + return _post( + "/api/v1/xhs/wen/xapi/schedule", + {"keywords": queries, "device_count": device_count}, + retry_transient=False, + ) + + +def collect_keyword_sources(keyword: str, credential_id: str | None = None) -> dict[str, Any]: + """使用 sources 的完整 Cookie 池采集单个关键词。""" + + normalized = normalize_collection_keywords([keyword])[0] + body: dict[str, Any] = {"keyword": normalized} + if credential_id: + body["credential_id"] = credential_id + return _post( + "/api/v1/xhs/wen/sources", + body, + retry_transient=False, + ) diff --git a/database.py b/database.py index bde8493..7f23510 100644 --- a/database.py +++ b/database.py @@ -133,8 +133,243 @@ def init_database() -> None: ); CREATE INDEX IF NOT EXISTS idx_auth_sessions_user ON auth_sessions(user_id, expires_at); + + CREATE TABLE IF NOT EXISTS acceptance_tasks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + public_task_id TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + acceptance_method TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'waiting', + progress_percent INTEGER NOT NULL DEFAULT 0, + device_count INTEGER NOT NULL DEFAULT 1, + created_by_id TEXT NOT NULL, + created_by_name TEXT NOT NULL, + error_message TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_acceptance_tasks_created + ON acceptance_tasks(created_at DESC); + + CREATE TABLE IF NOT EXISTS acceptance_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), + 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, + source_order INTEGER NOT NULL, + source_url TEXT NOT NULL, + final_url TEXT, + source_type TEXT, + title TEXT, + content_text TEXT, + content_json TEXT, + fetch_status TEXT NOT NULL DEFAULT 'waiting', + http_status INTEGER, + error_message TEXT, + fetched_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(task_id, source_order), + UNIQUE(task_id, source_url), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS acceptance_feishu_records ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_id INTEGER NOT NULL, + row_number INTEGER NOT NULL, + row_json TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(source_id, row_number), + FOREIGN KEY(source_id) REFERENCES acceptance_feishu_sources(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS acceptance_xhs_notes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + source_id INTEGER NOT NULL, + source_row_number INTEGER, + source_column_name TEXT, + note_url TEXT NOT NULL, + normalized_url TEXT NOT NULL, + note_id TEXT, + parse_status TEXT NOT NULL DEFAULT 'waiting', + parsed_json TEXT, + error_message TEXT, + parsed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(task_id, normalized_url), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE, + FOREIGN KEY(source_id) REFERENCES acceptance_feishu_sources(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_acceptance_notes_task_status + ON acceptance_xhs_notes(task_id, parse_status); + + CREATE TABLE IF NOT EXISTS feishu_notes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + note_url TEXT NOT NULL, + normalized_url TEXT NOT NULL UNIQUE, + note_id TEXT UNIQUE, + title TEXT, + body TEXT, + author_nickname TEXT, + author_avatar TEXT, + likes INTEGER, + collects INTEGER, + comments INTEGER, + images_json TEXT, + tags_json TEXT, + raw_json TEXT NOT NULL, + parsed_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_feishu_notes_note_id + ON feishu_notes(note_id); + + CREATE TABLE IF NOT EXISTS acceptance_keyword_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + keyword TEXT NOT NULL, + run_type TEXT NOT NULL, + run_order INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'waiting', + crawler_task_id TEXT, + result_json TEXT, + error_message TEXT, + started_at TEXT, + completed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(task_id, run_type, keyword), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_acceptance_keyword_runs_task + ON acceptance_keyword_runs(task_id, status); + + CREATE TABLE IF NOT EXISTS acceptance_cookie_attempts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + keyword_run_id INTEGER NOT NULL, + keyword TEXT NOT NULL, + credential_id TEXT NOT NULL, + attempt_number INTEGER NOT NULL, + status TEXT NOT NULL, + crawler_task_id TEXT, + result_json TEXT, + error_message TEXT, + started_at TEXT NOT NULL, + completed_at TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(keyword_run_id, credential_id, attempt_number), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE, + FOREIGN KEY(keyword_run_id) REFERENCES acceptance_keyword_runs(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_cookie_attempts_run_status + ON acceptance_cookie_attempts(keyword_run_id, status); + + CREATE TABLE IF NOT EXISTS acceptance_task_resources ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + resource_order INTEGER NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(task_id, resource_type, resource_id), + FOREIGN KEY(task_id) REFERENCES acceptance_tasks(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_acceptance_task_resources + ON acceptance_task_resources(task_id, resource_order); """ ) + # 将旧版本保存在 acceptance_xhs_notes.parsed_json 中的成功结果迁移到 + # 全局去重的笔记业务表;任务关联仍由 acceptance_xhs_notes 保留。 + legacy_rows = conn.execute( + """SELECT n.* FROM acceptance_xhs_notes n + LEFT JOIN feishu_notes f ON f.normalized_url=n.normalized_url + WHERE n.parse_status='completed' AND n.parsed_json IS NOT NULL AND f.id IS NULL""" + ).fetchall() + for row in legacy_rows: + try: + payload = json.loads(row["parsed_json"]) + except (TypeError, json.JSONDecodeError): + continue + upsert_feishu_note(conn, row, payload) + # 为升级前已有的 Cookie 尝试补建任务资源记录,详情页可立即展示设备进度。 + legacy_resources = conn.execute( + """SELECT task_id,credential_id,MIN(id) AS first_id + FROM acceptance_cookie_attempts + GROUP BY task_id,credential_id ORDER BY task_id,first_id""" + ).fetchall() + resource_orders: dict[int, int] = {} + for row in legacy_resources: + order = resource_orders.get(row["task_id"], 0) + conn.execute( + """INSERT OR IGNORE INTO acceptance_task_resources + (task_id,resource_type,resource_id,resource_order,created_at) + VALUES (?,'cookie',?,?,?)""", + (row["task_id"], row["credential_id"], order, now_iso()), + ) + resource_orders[row["task_id"]] = order + 1 + + +def _integer_or_none(value: Any) -> int | None: + if value is None or value == "": + return None + try: + return int(str(value).replace(",", "")) + except (TypeError, ValueError): + return None + + +def upsert_feishu_note(conn: sqlite3.Connection, note_row: sqlite3.Row, payload: dict[str, Any]) -> None: + """按规范化链接更新或新增笔记;任务关联保存在 acceptance_xhs_notes。""" + + now = now_iso() + note_id = payload.get("noteId") or payload.get("note_id") or note_row["note_id"] + existing = conn.execute( + """SELECT id FROM feishu_notes + WHERE normalized_url=? OR (? IS NOT NULL AND note_id=?) + ORDER BY CASE WHEN note_id=? THEN 0 ELSE 1 END LIMIT 1""", + (note_row["normalized_url"], note_id, note_id, note_id), + ).fetchone() + values = ( + note_row["note_url"], note_row["normalized_url"], note_id, + payload.get("title"), payload.get("body") or payload.get("content"), + payload.get("nickname") or payload.get("author_nickname"), + payload.get("avatar") or payload.get("author_avatar"), + _integer_or_none(payload.get("likes")), _integer_or_none(payload.get("collects")), + _integer_or_none(payload.get("comments")), + json.dumps(payload.get("images") or payload.get("images_list") or [], ensure_ascii=False), + json.dumps(payload.get("tags") or [], ensure_ascii=False), + json.dumps(payload, ensure_ascii=False), now, now, + ) + if existing: + conn.execute( + """UPDATE feishu_notes SET + note_url=?,normalized_url=?,note_id=?,title=?,body=?,author_nickname=?,author_avatar=?, + likes=?,collects=?,comments=?,images_json=?,tags_json=?,raw_json=?,parsed_at=?,updated_at=? + WHERE id=?""", + (*values, existing["id"]), + ) + else: + conn.execute( + """INSERT INTO feishu_notes + (note_url,normalized_url,note_id,title,body,author_nickname,author_avatar, + likes,collects,comments,images_json,tags_json,raw_json,parsed_at,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, + (*values, now), + ) def json_dumps(value: Any) -> str: diff --git a/requirements.txt b/requirements.txt index ac5769e..4fb9e89 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ Flask==3.1.1 requests==2.32.4 python-dotenv==1.1.1 gunicorn==23.0.0 +playwright==1.58.0