1910 lines
89 KiB
Python
1910 lines
89 KiB
Python
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,
|
||
create_xapi_device,
|
||
delete_xapi_device,
|
||
fetch_content_detail,
|
||
get_wen_resources,
|
||
list_xhs_wen_xapi_devices,
|
||
match_feishu_notes,
|
||
query_collection,
|
||
query_xapi_batch_runs,
|
||
schedule_xapi_screenshots,
|
||
sync_dashboard_acceptance_project,
|
||
)
|
||
from database import connection, now_iso, upsert_feishu_note
|
||
from task_service import TaskError
|
||
|
||
|
||
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["batch_date"]:
|
||
batch_day = datetime.strptime(task["batch_date"], "%Y-%m-%d").date()
|
||
today = datetime.now().astimezone().date()
|
||
if today > batch_day + timedelta(days=1):
|
||
raise TaskError(
|
||
"ACCEPTANCE_RETRY_WINDOW_EXPIRED",
|
||
f"{task['batch_date']} 批次仅允许在批次当天或次日重采",
|
||
409,
|
||
)
|
||
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)
|
||
active_device_ids = {
|
||
str(item.get("xhs_account_id") or "").strip()
|
||
for item in list_xhs_wen_xapi_devices()
|
||
if str(item.get("xhs_account_id") or "").strip()
|
||
}
|
||
for keyword in keyword_runs:
|
||
collected_ids = {
|
||
account_id for (row_keyword, account_id) in latest if row_keyword == keyword
|
||
}
|
||
missing_ids = active_device_ids - collected_ids
|
||
if missing_ids:
|
||
failed_by_keyword.setdefault(keyword, []).extend(sorted(missing_ids))
|
||
failed_by_keyword = {
|
||
keyword: list(dict.fromkeys(account_ids))
|
||
for keyword, account_ids in failed_by_keyword.items() if account_ids
|
||
}
|
||
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 and not keywords:
|
||
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 '飞书资料'}"
|
||
schedule_requested = data.get("schedule_weekdays") not in (None, [], "")
|
||
schedule_weekdays = _normalize_schedule_weekdays(data.get("schedule_weekdays")) if schedule_requested else []
|
||
schedule_start = _normalize_schedule_time(data.get("schedule_start_time", "00:00"), "开始时间") if schedule_requested else "00:00"
|
||
schedule_end = _normalize_schedule_time(data.get("schedule_end_time", "23:59"), "结束时间") if schedule_requested else "23:59"
|
||
if schedule_requested and schedule_start > schedule_end:
|
||
raise TaskError("INVALID_SCHEDULE_RANGE", "开始时间不能晚于结束时间", 422)
|
||
defer_initial = data.get("defer_initial_collection") is True
|
||
initial_status = "completed" if defer_initial else "waiting"
|
||
initial_progress = 100 if defer_initial else 0
|
||
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,
|
||
schedule_enabled,schedule_weekday,schedule_weekdays,schedule_start_time,
|
||
schedule_end_time,next_run_at,use_all_devices)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(public_id, name, method, initial_status, initial_progress, 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,
|
||
1 if schedule_requested else 0, schedule_weekdays[0] if schedule_weekdays else None,
|
||
json.dumps(schedule_weekdays) if schedule_weekdays else None,
|
||
schedule_start, schedule_end,
|
||
_next_weekly_run(schedule_weekdays, schedule_start) if schedule_weekdays else None,
|
||
1 if method in {"screenshot", "combined"} else 0),
|
||
)
|
||
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)],
|
||
)
|
||
if not defer_initial:
|
||
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()
|
||
current_device_count = task["device_count"]
|
||
if task["acceptance_method"] in {"screenshot", "combined"} and task["use_all_devices"]:
|
||
current_device_count = max(1, int(get_wen_resources().get("device_count") or 0))
|
||
child = create_acceptance_task({
|
||
"name": task["name"],
|
||
"acceptance_method": task["acceptance_method"],
|
||
"feishu_links": links,
|
||
"keywords": keywords,
|
||
"match_keywords": match_keywords,
|
||
"device_count": current_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=?,dashboard_batch_key=?,
|
||
use_all_devices=?,batch_date=? WHERE public_task_id=?""",
|
||
(task["id"], task["dashboard_batch_key"], task["use_all_devices"],
|
||
datetime.now().astimezone().date().isoformat(), 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 collect_acceptance_crawler_now(public_id: str) -> dict[str, Any]:
|
||
"""按项目搜索词同时启动 Cookie 爬虫和截图设备，按自然日归入批次。"""
|
||
with connection() as conn:
|
||
task = conn.execute("SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone()
|
||
if not task:
|
||
raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404)
|
||
root_id = task["scheduled_from_task_id"] or task["id"]
|
||
root = conn.execute("SELECT * FROM acceptance_tasks WHERE id=?", (root_id,)).fetchone()
|
||
keywords = [row["keyword"] for row in conn.execute("SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (root_id,)).fetchall()]
|
||
matches = [row["keyword"] for row in conn.execute("SELECT keyword FROM acceptance_match_keywords WHERE task_id=? ORDER BY keyword_order", (root_id,)).fetchall()]
|
||
child = create_acceptance_task({
|
||
"name": f"{root['name']}｜立即完整采集",
|
||
"acceptance_method": "combined",
|
||
"feishu_links": [],
|
||
"keywords": keywords,
|
||
"match_keywords": matches,
|
||
"created_by_id": root["created_by_id"],
|
||
"created_by_name": root["created_by_name"],
|
||
"trigger_source": "acceptance",
|
||
})
|
||
with connection() as conn:
|
||
conn.execute(
|
||
"""UPDATE acceptance_tasks SET scheduled_from_task_id=?,dashboard_batch_key=?,
|
||
use_all_devices=1,batch_date=?
|
||
WHERE public_task_id=?""",
|
||
(root_id, root["dashboard_batch_key"], datetime.now().astimezone().date().isoformat(), child["task_id"]),
|
||
)
|
||
return get_acceptance_task(child["task_id"])
|
||
|
||
|
||
def run_due_acceptance_schedules() -> int:
|
||
current = datetime.now().astimezone()
|
||
now = current.isoformat(timespec="seconds")
|
||
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:
|
||
# combined 任务的数据分析 Future 已经在运行；先下发截图，二者并行执行。
|
||
if task["acceptance_method"] == "combined":
|
||
run_failures += _execute_acceptance_method(
|
||
task["id"], "screenshot", keywords, task["device_count"],
|
||
)
|
||
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"],
|
||
)
|
||
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,batch_date,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"],
|
||
"dashboard_batch_key": task["dashboard_batch_key"],
|
||
"use_all_devices": bool(task["use_all_devices"]),
|
||
"batch_date": task["batch_date"],
|
||
"retry_deadline": (
|
||
(datetime.strptime(task["batch_date"], "%Y-%m-%d").date() + timedelta(days=1)).isoformat()
|
||
if task["batch_date"] else None
|
||
),
|
||
"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 WHERE scheduled_from_task_id IS NULL ORDER BY created_at DESC"
|
||
).fetchall()
|
||
items = [_task_dict(conn, row) for row in rows]
|
||
return {"items": items, "total": len(items)}
|
||
|
||
|
||
def create_acceptance_device(data: dict[str, Any]) -> dict[str, Any]:
|
||
result = create_xapi_device(data)
|
||
result["resources"] = get_acceptance_resources()
|
||
return result
|
||
|
||
|
||
def remove_acceptance_device(account_id: str) -> dict[str, Any]:
|
||
result = delete_xapi_device(account_id)
|
||
result["resources"] = get_acceptance_resources()
|
||
return result
|
||
|
||
|
||
def update_acceptance_task(public_id: str, data: dict[str, Any]) -> dict[str, Any]:
|
||
name = str(data.get("name") or "").strip()
|
||
keywords = list(dict.fromkeys(str(item).strip() for item in data.get("keywords", []) if str(item).strip()))
|
||
matches = list(dict.fromkeys(str(item).strip() for item in data.get("match_keywords", []) if str(item).strip()))
|
||
if not name or not keywords or not matches:
|
||
raise TaskError("INVALID_ACCEPTANCE_PROJECT", "项目名称、搜索关键词和匹配关键词不能为空", 422)
|
||
with connection() as conn:
|
||
task = conn.execute("SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone()
|
||
if not task:
|
||
raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404)
|
||
root_id = task["scheduled_from_task_id"] or task["id"]
|
||
root = conn.execute("SELECT * FROM acceptance_tasks WHERE id=?", (root_id,)).fetchone()
|
||
conn.execute("UPDATE acceptance_tasks SET name=?,updated_at=? WHERE id=?", (name, now_iso(), root_id))
|
||
conn.execute("DELETE FROM acceptance_keywords WHERE task_id=?", (root_id,))
|
||
conn.execute("DELETE FROM acceptance_match_keywords WHERE task_id=?", (root_id,))
|
||
conn.executemany("INSERT INTO acceptance_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", [(root_id, i, value, now_iso()) for i, value in enumerate(keywords)])
|
||
conn.executemany("INSERT INTO acceptance_match_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)", [(root_id, i, value, now_iso()) for i, value in enumerate(matches)])
|
||
dashboard_key = root["dashboard_batch_key"] or root["public_task_id"]
|
||
sync_dashboard_acceptance_project(dashboard_key, name, keywords, matches)
|
||
return get_acceptance_task(root["public_task_id"])
|
||
|
||
|
||
def list_acceptance_notes(public_id: str) -> dict[str, Any]:
|
||
with connection() as conn:
|
||
task = conn.execute("SELECT id FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone()
|
||
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": "仅包含当前已完成并已入库的数据",
|
||
}
|