liqing/holy-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 验收任务周期采集与设备资源接口

Browse Source 
- 新增 acceptance-tasks 周期采集、立即采集、断点续采与批次管理
- acceptance-resources 返回 xhs_device_accounts 真实设备清单
- 数据看板服务调整及测试更新
This commit is contained in:
李清
2026-08-28 13:43:53 +08:00
parent a021f2fe60
commit e188c50fab
8 changed files with 819 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

626
acceptance_service.py
View File

@@ -1,5 +1,6 @@
import json
import csv
import hashlib
import io
import platform
import re
@@ -8,9 +9,10 @@ import string
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from datetime import datetime, timedelta
from html import unescape
from html.parser import HTMLParser
from http.cookies import SimpleCookie
from typing import Any
from urllib.parse import urlparse
@@ -22,8 +24,10 @@ from crawler_client import (
collect_keyword_sources,
fetch_content_detail,
get_wen_resources,
list_xhs_wen_xapi_devices,
match_feishu_notes,
query_collection,
query_xapi_batch_runs,
schedule_xapi_screenshots,
)
from database import connection, now_iso, upsert_feishu_note
@@ -31,9 +35,11 @@ from task_service import TaskError
ALLOWED_HOST_SUFFIXES = (".feishu.cn", ".larksuite.com")
METHODS = {"screenshot", "data-analysis"}
METHODS = {"screenshot", "data-analysis", "combined"}
WEEKDAYS = set(range(1, 8))
_schedule_started = False
XHS_URL_PATTERN = re.compile(
r"https?://(?:www\.)?(?:xiaohongshu\.com|xhslink\.com)/[^\s\t\r\n<>\"']+",
r"https?://(?:www\.)?(?:xiaohongshu\.com|xhslink\.(?:com|cn))/[^\s\t\r\n<>\"']+",
re.IGNORECASE,
)
@@ -78,6 +84,34 @@ def _public_id() -> str:
return f"HV-{datetime.now():%Y%m%d}-{''.join(secrets.choice(chars) for _ in range(5))}"
def _normalize_custom_cookie(value: Any) -> str:
cookie = str(value or "").strip()
if cookie.lower().startswith("cookie:"):
cookie = cookie.split(":", 1)[1].strip()
if not cookie:
return ""
if len(cookie) > 16384:
raise TaskError("CUSTOM_COOKIE_TOO_LONG", "自定义 Cookie 不能超过 16384 个字符", 422)
parsed = SimpleCookie()
try:
parsed.load(cookie)
except Exception as exc:
raise TaskError("INVALID_CUSTOM_COOKIE", "自定义 Cookie 格式无效", 422) from exc
if not parsed:
raise TaskError("INVALID_CUSTOM_COOKIE", "自定义 Cookie 格式无效", 422)
return cookie
def _custom_cookie_for_task(task_id: int) -> str:
with connection() as conn:
row = conn.execute(
"""SELECT secret_value FROM acceptance_task_secrets
WHERE task_id=? AND secret_type='custom_cookie'""",
(task_id,),
).fetchone()
return str(row["secret_value"] if row else "")
def _validate_url(value: Any) -> str:
url = str(value or "").strip()
parsed = urlparse(url)
@@ -343,11 +377,17 @@ def _update_run(run_id: int, *, status: str, result: dict[str, Any] | None = Non
)
def get_acceptance_resources() -> dict[str, int]:
def get_acceptance_resources() -> dict[str, Any]:
resources = get_wen_resources()
try:
devices = list_xhs_wen_xapi_devices()
except Exception:
devices = []
return {
"cookie_count": max(0, int(resources.get("cookie_count") or 0)),
"device_count": max(0, int(resources.get("device_count") or 0)),
"ocr_questions": resources.get("ocr_questions") or [],
"devices": devices,
}
@@ -418,11 +458,20 @@ def _run_keyword_with_cookie_failover(
task_id: int, run_id: int, keyword: str, target_count: int,
) -> bool:
"""补足目标成功数；成功 Cookie 不重跑，失败时自动尝试池中其他 Cookie。"""
resources = get_wen_resources()
available_ids = list(dict.fromkeys(
str(item).strip() for item in resources.get("credential_ids", [])
if str(item).strip()
))
custom_cookie = _custom_cookie_for_task(task_id)
custom_credential_id = (
f"custom-{hashlib.sha256(custom_cookie.encode('utf-8')).hexdigest()[:10]}"
if custom_cookie else ""
)
resources = get_wen_resources() if not custom_cookie else {}
available_ids = (
[custom_credential_id]
if custom_cookie
else list(dict.fromkeys(
str(item).strip() for item in resources.get("credential_ids", [])
if str(item).strip()
))
)
with connection() as conn:
now = now_iso()
for index, credential_id in enumerate(available_ids[:target_count]):
@@ -459,7 +508,11 @@ def _run_keyword_with_cookie_failover(
)
try:
result, crawler_id, remote_status = _wait_collection_result(
collect_keyword_sources(keyword, credential_id)
collect_keyword_sources(
keyword,
None if custom_cookie else credential_id,
custom_cookie or None,
)
)
if crawler_id:
crawler_ids.append(crawler_id)
@@ -510,12 +563,53 @@ def _execute_acceptance_method(task_id: int, method: str, keywords: list[str], r
failures = 0
if method == "screenshot":
try:
with connection() as conn:
trigger_row = conn.execute(
"SELECT trigger_source FROM acceptance_tasks WHERE id=?", (task_id,),
).fetchone()
trigger_source = str(trigger_row["trigger_source"] if trigger_row else "acceptance")
for run_id in run_ids:
_update_run(run_id, status="running")
result = schedule_xapi_screenshots(keywords, resource_count)
crawler_id = str(result.get("task_id") or result.get("schedule_id") or "") or None
for run_id in run_ids:
_update_run(run_id, status="completed", result=result, crawler_task_id=crawler_id, completed=True)
result = schedule_xapi_screenshots(
keywords, resource_count, trigger_source=trigger_source,
)
if result.get("execution_mode") == "immediate":
plans = {keyword: {"batches": result.get("batches") or []} for keyword in keywords}
else:
plans = {
str(plan.get("search_query") or ""): plan
for plan in result.get("plans") or [] if isinstance(plan, dict)
}
plan_rows = []
for run_id, keyword in zip(run_ids, keywords):
plan = plans.get(keyword) or {}
batch_ids = [
str(batch.get("batch_id") or "").strip()
for batch in plan.get("batches") or []
if str(batch.get("batch_id") or "").strip()
]
_update_run(
run_id, status="waiting", result=plan or result,
crawler_task_id=",".join(batch_ids) or None,
)
plan_rows.append(plan)
with connection() as conn:
resource_order = 0
seen_resources = set()
for plan in plan_rows:
for batch in plan.get("batches") or []:
for item in batch.get("items") or []:
account_id = str(item.get("xhs_account_id") or "").strip()
if not account_id or account_id in seen_resources:
continue
seen_resources.add(account_id)
conn.execute(
"""INSERT OR IGNORE INTO acceptance_task_resources
(task_id,resource_type,resource_id,resource_order,created_at)
VALUES (?,'device',?,?,?)""",
(task_id, account_id, resource_order, now_iso()),
)
resource_order += 1
except Exception as exc:
failures = len(run_ids)
for run_id in run_ids:
@@ -551,25 +645,218 @@ def _execute_acceptance_method(task_id: int, method: str, keywords: list[str], r
return failures
def _decode_json(value: Any, fallback: Any) -> Any:
if not value:
return fallback
if not isinstance(value, str):
return value
try:
return json.loads(value)
except (TypeError, json.JSONDecodeError):
return fallback
def _upsert_acceptance_xapi_runs(
task_id: int, keyword_run_id: int | None, batch_id: str,
runs: list[dict[str, Any]],
) -> None:
now = now_iso()
with connection() as conn:
for run in runs:
xapi_task_id = str(run.get("task_id") or run.get("id") or "").strip()
if not xapi_task_id:
continue
artifacts = run.get("artifacts")
if artifacts is None:
artifacts = _decode_json(run.get("artifacts_json"), [])
feishu_record = run.get("feishu_record")
if feishu_record is None:
feishu_record = _decode_json(run.get("feishu_record_json"), None)
conn.execute(
"""INSERT INTO acceptance_xapi_runs
(task_id,keyword_run_id,batch_id,xapi_task_id,keyword,xhs_account_id,
device_serial,device_label,status,writeback_status,source_base_url,
base_record_id,artifacts_json,feishu_result_text,feishu_record_json,
feishu_sync_status,feishu_sync_error,raw_json,created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(xapi_task_id) DO UPDATE SET
keyword_run_id=excluded.keyword_run_id,batch_id=excluded.batch_id,
keyword=excluded.keyword,xhs_account_id=excluded.xhs_account_id,
device_serial=excluded.device_serial,device_label=excluded.device_label,
status=excluded.status,writeback_status=excluded.writeback_status,
source_base_url=excluded.source_base_url,base_record_id=excluded.base_record_id,
artifacts_json=excluded.artifacts_json,
feishu_result_text=excluded.feishu_result_text,
feishu_record_json=excluded.feishu_record_json,
feishu_sync_status=excluded.feishu_sync_status,
feishu_sync_error=excluded.feishu_sync_error,
raw_json=excluded.raw_json,updated_at=excluded.updated_at""",
(
task_id, keyword_run_id, batch_id, xapi_task_id,
run.get("search_query"), run.get("xhs_account_id"),
run.get("device_serial"), run.get("device_label"), run.get("status"),
run.get("writeback_status"), run.get("source_base_url"),
run.get("base_record_id"), json.dumps(artifacts or [], ensure_ascii=False),
run.get("feishu_result_text"),
json.dumps(feishu_record, ensure_ascii=False) if feishu_record is not None else None,
run.get("feishu_sync_status"), run.get("feishu_sync_error"),
json.dumps(run, ensure_ascii=False, default=str), now, now,
),
)
def _poll_screenshot_results(public_id: str) -> None:
"""轮询已计划批次，持久化截图和飞书结果，并更新前端任务进度。"""
terminal_statuses = {"completed", "failed", "cancelled", "canceled"}
with connection() as conn:
task = conn.execute(
"SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,),
).fetchone()
if not task or task["acceptance_method"] not in {"screenshot", "combined"}:
return
keyword_runs = conn.execute(
"""SELECT id,keyword,crawler_task_id,result_json FROM acceptance_keyword_runs
WHERE task_id=? AND run_type='screenshot' ORDER BY run_order""",
(task["id"],),
).fetchall()
batch_map: dict[str, tuple[int, str, int]] = {}
for keyword_run in keyword_runs:
plan = _decode_json(keyword_run["result_json"], {}) or {}
batches = plan.get("batches") or []
expected_by_id = {
str(batch.get("batch_id") or ""): len(batch.get("items") or [])
for batch in batches if str(batch.get("batch_id") or "").strip()
}
for batch_id in str(keyword_run["crawler_task_id"] or "").split(","):
normalized = batch_id.strip()
if normalized:
batch_map[normalized] = (
keyword_run["id"], keyword_run["keyword"],
max(1, expected_by_id.get(normalized, 1)),
)
if not batch_map:
with connection() as conn:
conn.execute(
"""UPDATE acceptance_tasks SET status='failed',progress_percent=100,
error_message=?,updated_at=? WHERE id=?""",
("截图调度没有返回可回查的 batch_id", now_iso(), task["id"]),
)
return
deadline = time.monotonic() + max(1.0, Config.XAPI_RESULT_MAX_WAIT_HOURS) * 3600
while True:
poll_errors = []
for batch_id, (keyword_run_id, _, _) in batch_map.items():
try:
result = query_xapi_batch_runs(batch_id, refresh=True)
_upsert_acceptance_xapi_runs(
task["id"], keyword_run_id, batch_id, result.get("runs") or [],
)
except Exception as exc:
poll_errors.append(f"{batch_id}: {exc}")
with connection() as conn:
rows = conn.execute(
"SELECT * FROM acceptance_xapi_runs WHERE task_id=?", (task["id"],),
).fetchall()
expected_total = sum(item[2] for item in batch_map.values())
terminal_count = sum(
str(row["status"] or "").lower() in terminal_statuses for row in rows
)
feishu_ready = all(
str(row["status"] or "").lower() != "completed"
or str(row["feishu_sync_status"] or "").lower() == "succeeded"
for row in rows
)
received = min(len(rows), expected_total)
progress = min(99, 70 + round(29 * received / max(1, expected_total)))
for keyword_run in keyword_runs:
related = [row for row in rows if row["keyword_run_id"] == keyword_run["id"]]
related_batch_ids = [
value.strip() for value in str(keyword_run["crawler_task_id"] or "").split(",")
if value.strip()
]
keyword_expected = sum(batch_map[item][2] for item in related_batch_ids)
keyword_terminal = sum(
str(row["status"] or "").lower() in terminal_statuses for row in related
)
if keyword_expected and keyword_terminal >= keyword_expected:
failed = any(str(row["status"] or "").lower() != "completed" for row in related)
conn.execute(
"""UPDATE acceptance_keyword_runs SET status=?,error_message=?,
completed_at=?,updated_at=? WHERE id=?""",
("failed" if failed else "completed",
"部分设备截图失败" if failed else None,
now_iso(), now_iso(), keyword_run["id"]),
)
all_terminal = expected_total > 0 and terminal_count >= expected_total
timed_out = time.monotonic() >= deadline
if all_terminal and feishu_ready:
failed_count = sum(
str(row["status"] or "").lower() != "completed" for row in rows
)
final_status = "partial" if failed_count else "completed"
conn.execute(
"""UPDATE acceptance_tasks SET status=?,progress_percent=100,
error_message=?,updated_at=? WHERE id=?""",
(final_status, "部分设备截图失败" if failed_count else None,
now_iso(), task["id"]),
)
return
if timed_out:
reason = "截图或飞书结果等待超时"
if poll_errors:
reason += "；" + "；".join(poll_errors[:3])
conn.execute(
"""UPDATE acceptance_tasks SET status='partial',progress_percent=100,
error_message=?,updated_at=? WHERE id=?""",
(reason[:2000], now_iso(), task["id"]),
)
return
conn.execute(
"UPDATE acceptance_tasks SET status='running',progress_percent=?,updated_at=? WHERE id=?",
(progress, now_iso(), task["id"]),
)
time.sleep(max(5.0, Config.XAPI_RESULT_POLL_INTERVAL_SECONDS))
def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]:
links = list(dict.fromkeys(_validate_url(item) for item in data.get("feishu_links", [])))
if not links:
raise TaskError("FEISHU_LINK_REQUIRED", "至少需要一个公开飞书链接", 422)
method = str(data.get("acceptance_method") or "")
if method not in METHODS:
raise TaskError("INVALID_ACCEPTANCE_METHOD", "验收方式无效", 422)
keywords = list(dict.fromkeys(str(item).strip() for item in data.get("keywords", []) if str(item).strip()))
custom_cookie = _normalize_custom_cookie(data.get("custom_cookie"))
if custom_cookie and method not in {"data-analysis", "combined"}:
raise TaskError(
"CUSTOM_COOKIE_NOT_SUPPORTED",
"自定义 Cookie 仅支持数据分析验收",
422,
)
raw_resources = get_wen_resources()
keywords = list(dict.fromkeys(
str(item).strip() for item in data.get("keywords", []) if str(item).strip()
))
match_keywords = list(dict.fromkeys(
str(item).strip() for item in data.get("match_keywords", []) if str(item).strip()
))
if len(keywords) > 20:
raise TaskError("KEYWORD_LIMIT_EXCEEDED", "单次最多支持 20 个搜索词", 422)
if method in {"screenshot", "combined"} and not keywords:
raise TaskError("KEYWORD_REQUIRED", "截图取证请至少输入一个搜索词", 422)
if method in {"data-analysis", "combined"} and not links:
raise TaskError("FEISHU_LINK_REQUIRED", "笔记收录请至少输入一个公开飞书链接", 422)
resources = {
"cookie_count": max(0, int(raw_resources.get("cookie_count") or 0)),
"device_count": max(0, int(raw_resources.get("device_count") or 0)),
}
device_count = (
resources["cookie_count"] if method == "data-analysis"
else int(data.get("device_count") or 1)
cookie_count = 1 if custom_cookie else resources["cookie_count"]
device_count = resources["device_count"] if method == "combined" else (
cookie_count if method == "data-analysis" else int(data.get("device_count") or 1)
)
available_count = resources["device_count" if method == "screenshot" else "cookie_count"]
resource_name = "设备" if method == "screenshot" else "Cookie"
available_count = resources["device_count"] if method in {"screenshot", "combined"} else cookie_count
resource_name = "设备" if method in {"screenshot", "combined"} else "Cookie"
if method == "combined" and cookie_count < 1:
raise TaskError("INVALID_RESOURCE_COUNT", "当前没有可用的数据分析 Cookie", 422)
if not 1 <= device_count <= available_count:
raise TaskError(
"INVALID_RESOURCE_COUNT",
@@ -583,22 +870,38 @@ def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]:
cursor = conn.execute(
"""INSERT INTO acceptance_tasks
(public_task_id,name,acceptance_method,status,progress_percent,device_count,
created_by_id,created_by_name,created_at,updated_at)
VALUES (?,?,?,'waiting',0,?,?,?,?,?)""",
created_by_id,created_by_name,created_at,updated_at,trigger_source,cookie_count)
VALUES (?,?,?,'waiting',0,?,?,?,?,?,?,?)""",
(public_id, name, method, device_count, str(data.get("created_by_id") or "local-dev"),
str(data.get("created_by_name") or "本地开发用户"), now, now),
str(data.get("created_by_name") or "本地开发用户"), now, now,
str(data.get("trigger_source") or "acceptance"), cookie_count),
)
task_id = cursor.lastrowid
conn.executemany(
"INSERT INTO acceptance_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)",
[(task_id, index, keyword, now) for index, keyword in enumerate(keywords)],
)
if method == "data-analysis":
credential_ids = [
str(item).strip()
for item in raw_resources.get("credential_ids", [])
if str(item).strip()
][:device_count]
conn.executemany(
"INSERT INTO acceptance_match_keywords(task_id,keyword_order,keyword,created_at) VALUES (?,?,?,?)",
[(task_id, index, keyword, now) for index, keyword in enumerate(match_keywords)],
)
if method in {"data-analysis", "combined"}:
if custom_cookie:
credential_ids = [
f"custom-{hashlib.sha256(custom_cookie.encode('utf-8')).hexdigest()[:10]}"
]
conn.execute(
"""INSERT INTO acceptance_task_secrets
(task_id,secret_type,secret_value,created_at)
VALUES (?,'custom_cookie',?,?)""",
(task_id, custom_cookie, now),
)
else:
credential_ids = [
str(item).strip()
for item in raw_resources.get("credential_ids", [])
if str(item).strip()
][:cookie_count]
conn.executemany(
"""INSERT OR IGNORE INTO acceptance_task_resources
(task_id,resource_type,resource_id,resource_order,created_at)
@@ -615,6 +918,143 @@ def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]:
return get_acceptance_task(public_id)
def _next_weekly_run(weekday: int, *, after: datetime | None = None) -> str:
current = after or datetime.now().astimezone()
days = (weekday - current.isoweekday()) % 7
candidate = (current + timedelta(days=days)).replace(
hour=0, minute=0, second=0, microsecond=0,
)
if candidate <= current:
candidate += timedelta(days=7)
return candidate.isoformat(timespec="seconds")
def update_acceptance_schedule(public_id: str, weekday: int) -> dict[str, Any]:
try:
weekday = int(weekday)
except (TypeError, ValueError) as exc:
raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422) from exc
if weekday not in WEEKDAYS:
raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422)
with connection() as conn:
task = conn.execute(
"SELECT id FROM acceptance_tasks WHERE public_task_id=?", (public_id,),
).fetchone()
if not task:
raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404)
conn.execute(
"""UPDATE acceptance_tasks SET schedule_enabled=1,schedule_weekday=?,
next_run_at=?,schedule_error=NULL,updated_at=? WHERE id=?""",
(weekday, _next_weekly_run(weekday), now_iso(), task["id"]),
)
return get_acceptance_task(public_id)
def stop_acceptance_schedule(public_id: str) -> dict[str, Any]:
with connection() as conn:
result = conn.execute(
"""UPDATE acceptance_tasks SET schedule_enabled=0,next_run_at=NULL,
schedule_error=NULL,updated_at=? WHERE public_task_id=?""",
(now_iso(), public_id),
)
if not result.rowcount:
raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404)
return get_acceptance_task(public_id)
def _copy_acceptance_task(public_id: str, *, trigger_source: str = "acceptance") -> dict[str, Any]:
with connection() as conn:
task = conn.execute(
"SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,),
).fetchone()
if not task:
raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404)
keywords = [row["keyword"] for row in conn.execute(
"SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order",
(task["id"],),
).fetchall()]
match_keywords = [row["keyword"] for row in conn.execute(
"SELECT keyword FROM acceptance_match_keywords WHERE task_id=? ORDER BY keyword_order",
(task["id"],),
).fetchall()]
links = [row["source_url"] for row in conn.execute(
"SELECT source_url FROM acceptance_feishu_sources WHERE task_id=? ORDER BY source_order",
(task["id"],),
).fetchall()]
secret = conn.execute(
"""SELECT secret_value FROM acceptance_task_secrets
WHERE task_id=? AND secret_type='custom_cookie'""",
(task["id"],),
).fetchone()
child = create_acceptance_task({
"name": task["name"],
"acceptance_method": task["acceptance_method"],
"feishu_links": links,
"keywords": keywords,
"match_keywords": match_keywords,
"device_count": task["device_count"],
"created_by_id": task["created_by_id"],
"created_by_name": task["created_by_name"],
"custom_cookie": secret["secret_value"] if secret else None,
"trigger_source": trigger_source,
})
with connection() as conn:
conn.execute(
"UPDATE acceptance_tasks SET scheduled_from_task_id=? WHERE public_task_id=?",
(task["id"], child["task_id"]),
)
return get_acceptance_task(child["task_id"])
def collect_acceptance_now(public_id: str) -> dict[str, Any]:
"""复用任务配置创建一次独立采集，截图任务会立即向 Framework 下发。"""
return _copy_acceptance_task(public_id, trigger_source="acceptance")
def run_due_acceptance_schedules() -> int:
now = now_iso()
with connection() as conn:
due = conn.execute(
"""SELECT public_task_id,schedule_weekday FROM acceptance_tasks
WHERE schedule_enabled=1 AND next_run_at IS NOT NULL AND next_run_at<=?
ORDER BY next_run_at LIMIT 20""",
(now,),
).fetchall()
for row in due:
conn.execute(
"""UPDATE acceptance_tasks SET next_run_at=?,last_run_at=?,
schedule_error=NULL,updated_at=? WHERE public_task_id=?""",
(_next_weekly_run(row["schedule_weekday"]), now, now, row["public_task_id"]),
)
for row in due:
try:
_copy_acceptance_task(row["public_task_id"], trigger_source="acceptance-weekly")
except Exception as exc:
with connection() as conn:
conn.execute(
"UPDATE acceptance_tasks SET schedule_error=?,updated_at=? WHERE public_task_id=?",
(str(exc)[:2000], now_iso(), row["public_task_id"]),
)
return len(due)
def start_acceptance_schedule_worker() -> None:
global _schedule_started
if _schedule_started:
return
_schedule_started = True
def worker() -> None:
while True:
try:
run_due_acceptance_schedules()
except Exception:
pass
time.sleep(30)
threading.Thread(target=worker, daemon=True, name="acceptance-weekly-scheduler").start()
def _fetch_sources(public_id: str) -> None:
with connection() as conn:
task = conn.execute("SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,)).fetchone()
@@ -629,11 +1069,12 @@ def _fetch_sources(public_id: str) -> None:
# 数据分析时，表单中已知的关键词无需等待飞书解析完成，立即并行采集。
analysis_pool = None
initial_analysis_future = None
if task["acceptance_method"] == "data-analysis" and initial_keywords:
if task["acceptance_method"] in {"data-analysis", "combined"} and initial_keywords:
analysis_pool = ThreadPoolExecutor(max_workers=1, thread_name_prefix=f"analysis-{public_id}")
initial_analysis_future = analysis_pool.submit(
_execute_acceptance_method,
task["id"], task["acceptance_method"], initial_keywords, task["device_count"],
task["id"], "data-analysis", initial_keywords,
task["cookie_count"] if task["acceptance_method"] == "combined" else task["device_count"],
)
failures = 0
note_failures = 0
@@ -711,7 +1152,7 @@ def _fetch_sources(public_id: str) -> None:
conn.execute("UPDATE acceptance_tasks SET progress_percent=?,updated_at=? WHERE public_task_id=?", (progress, now_iso(), public_id))
run_failures = 0
run_error = None
if failures < len(sources):
if not sources or failures < len(sources):
try:
keywords = _sync_acceptance_keywords(task["id"], discovered_keywords)
with connection() as conn:
@@ -721,7 +1162,12 @@ def _fetch_sources(public_id: str) -> None:
new_keywords = [keyword for keyword in keywords if keyword not in initial_keywords]
if new_keywords:
run_failures += _execute_acceptance_method(
task["id"], task["acceptance_method"], new_keywords, task["device_count"]
task["id"], "data-analysis", new_keywords,
task["cookie_count"] if task["acceptance_method"] == "combined" else task["device_count"],
)
if task["acceptance_method"] == "combined":
run_failures += _execute_acceptance_method(
task["id"], "screenshot", keywords, task["device_count"],
)
else:
run_failures = _execute_acceptance_method(
@@ -742,6 +1188,19 @@ def _fetch_sources(public_id: str) -> None:
finally:
if analysis_pool is not None:
analysis_pool.shutdown(wait=False)
if task["acceptance_method"] in {"screenshot", "combined"} and not run_error and not run_failures:
with connection() as conn:
conn.execute(
"""UPDATE acceptance_tasks SET status='running',progress_percent=70,
error_message=?,updated_at=? WHERE public_task_id=?""",
("部分飞书资料或小红书笔记解析失败" if failures or note_failures else None,
now_iso(), public_id),
)
threading.Thread(
target=_poll_screenshot_results,
args=(public_id,), daemon=True, name=f"acceptance-xapi-{public_id}",
).start()
return
status = "failed" if failures == len(sources) or run_error else ("partial" if failures or note_failures or run_failures else "completed")
error = "；".join(filter(None, [
"部分飞书资料或小红书笔记解析失败" if failures or note_failures else None,
@@ -782,13 +1241,23 @@ def resume_acceptance_task(public_id: str) -> dict[str, Any]:
def resume_unfinished_acceptance_tasks() -> None:
"""服务重启后恢复等待中或运行中的验收任务。"""
with connection() as conn:
task_ids = [row["public_task_id"] for row in conn.execute(
"""SELECT public_task_id FROM acceptance_tasks
tasks = conn.execute(
"""SELECT t.public_task_id,t.acceptance_method,
EXISTS(SELECT 1 FROM acceptance_keyword_runs r
WHERE r.task_id=t.id AND r.run_type='screenshot'
AND r.crawler_task_id IS NOT NULL) AS has_screenshot_batches
FROM acceptance_tasks t
WHERE status IN ('waiting','running') ORDER BY created_at"""
).fetchall()]
for public_id in task_ids:
).fetchall()
for task in tasks:
public_id = task["public_task_id"]
target = (
_poll_screenshot_results
if task["acceptance_method"] in {"screenshot", "combined"} and task["has_screenshot_batches"]
else _fetch_sources
)
threading.Thread(
target=_fetch_sources, args=(public_id,), daemon=True,
target=target, args=(public_id,), daemon=True,
name=f"acceptance-recover-{public_id}",
).start()
@@ -843,6 +1312,17 @@ def _task_dict(conn, task) -> dict[str, Any]:
keywords = [row["keyword"] for row in conn.execute(
"SELECT keyword FROM acceptance_keywords WHERE task_id=? ORDER BY keyword_order", (task["id"],)
).fetchall()]
match_keywords = [row["keyword"] for row in conn.execute(
"SELECT keyword FROM acceptance_match_keywords WHERE task_id=? ORDER BY keyword_order",
(task["id"],),
).fetchall()]
root_id = task["scheduled_from_task_id"] or task["id"]
batches = [dict(row) for row in conn.execute(
"""SELECT public_task_id,status,progress_percent,created_at,updated_at
FROM acceptance_tasks WHERE id=? OR scheduled_from_task_id=?
ORDER BY created_at DESC""",
(root_id, root_id),
).fetchall()]
sources = [dict(row) for row in conn.execute(
"""SELECT source_url,final_url,source_type,title,content_text,fetch_status,http_status,
error_message,fetched_at,
@@ -879,8 +1359,57 @@ def _task_dict(conn, task) -> dict[str, Any]:
FROM acceptance_cookie_attempts WHERE task_id=? ORDER BY id""",
(task["id"],),
).fetchall()
xapi_rows = conn.execute(
"""SELECT batch_id,xapi_task_id,keyword,xhs_account_id,device_serial,
device_label,status,writeback_status,source_base_url,base_record_id,
artifacts_json,feishu_result_text,feishu_record_json,
feishu_sync_status,feishu_sync_error,updated_at
FROM acceptance_xapi_runs WHERE task_id=? ORDER BY id""",
(task["id"],),
).fetchall()
resource_progress = []
for resource in task_resources:
if resource["resource_type"] == "device":
related_runs = [
row for row in xapi_rows if row["xhs_account_id"] == resource["resource_id"]
]
completed_keywords = len({
row["keyword"] for row in related_runs
if str(row["status"] or "").lower() == "completed"
and str(row["feishu_sync_status"] or "").lower() == "succeeded"
})
failed_keywords = len({
row["keyword"] for row in related_runs
if str(row["status"] or "").lower() in {"failed", "cancelled", "canceled"}
})
running_keywords = len({
row["keyword"] for row in related_runs
if str(row["status"] or "").lower() not in {"completed", "failed", "cancelled", "canceled"}
})
total_keywords = len(keywords)
terminal_keywords = completed_keywords + failed_keywords
resource_progress.append({
"resource_type": "device",
"resource_id": resource["resource_id"],
"resource_order": resource["resource_order"],
"status": (
"failed" if failed_keywords
else "completed" if total_keywords and completed_keywords >= total_keywords
else "running" if related_runs
else "waiting"
),
"keyword_total": total_keywords,
"keyword_completed": completed_keywords,
"keyword_failed": failed_keywords,
"keyword_running": running_keywords,
"progress_percent": round(terminal_keywords / max(1, total_keywords) * 100),
"attempt_count": len(related_runs),
"error_message": next(
(row["feishu_sync_error"] for row in reversed(related_runs) if row["feishu_sync_error"]),
None,
),
})
continue
related = [row for row in cookie_attempts if row["credential_id"] == resource["resource_id"]]
completed_keywords = failed_keywords = running_keywords = 0
for keyword in keywords:
@@ -914,14 +1443,29 @@ def _task_dict(conn, task) -> dict[str, Any]:
"attempt_count": len(related),
"error_message": next((row["error_message"] for row in reversed(related) if row["error_message"]), None),
})
xapi_runs = []
for row in xapi_rows:
item = dict(row)
item["artifacts"] = _decode_json(item.pop("artifacts_json"), [])
item["feishu_record"] = _decode_json(item.pop("feishu_record_json"), None)
xapi_runs.append(item)
return {
"task_id": task["public_task_id"], "name": task["name"],
"acceptance_method": task["acceptance_method"], "status": task["status"],
"progress_percent": task["progress_percent"], "device_count": task["device_count"],
"keywords": keywords, "feishu_links": [item["source_url"] for item in sources],
"cookie_count": task["cookie_count"],
"schedule_enabled": bool(task["schedule_enabled"]),
"schedule_weekday": task["schedule_weekday"],
"next_run_at": task["next_run_at"], "last_run_at": task["last_run_at"],
"scheduled_from_task_id": task["scheduled_from_task_id"],
"schedule_error": task["schedule_error"],
"trigger_source": task["trigger_source"],
"keywords": keywords, "match_keywords": match_keywords,
"batches": batches, "feishu_links": [item["source_url"] for item in sources],
"feishu_sources": sources, "created_by": {"id": task["created_by_id"], "name": task["created_by_name"]},
"note_stats": {"total": note_stats["total"] or 0, "completed": note_stats["completed"] or 0, "failed": note_stats["failed"] or 0},
"keyword_runs": keyword_runs,
"xapi_runs": xapi_runs,
"resource_progress": resource_progress,
"error_message": task["error_message"], "created_at": task["created_at"], "updated_at": task["updated_at"],
}