diff --git a/acceptance_service.py b/acceptance_service.py index bbbf43e..2cd27b0 100644 --- a/acceptance_service.py +++ b/acceptance_service.py @@ -745,6 +745,8 @@ def _poll_screenshot_results(public_id: str) -> None: 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: @@ -756,22 +758,31 @@ def _poll_screenshot_results(public_id: str) -> None: 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 rows + 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 rows + for row in tracked_rows ) - received = min(len(rows), expected_total) + 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 rows if row["keyword_run_id"] == keyword_run["id"]] + 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() @@ -792,8 +803,13 @@ def _poll_screenshot_results(public_id: str) -> None: 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 rows + str(row["status"] or "").lower() != "completed" + for row in latest_rows.values() ) final_status = "partial" if failed_count else "completed" conn.execute( @@ -820,6 +836,94 @@ def _poll_screenshot_results(public_id: str) -> None: time.sleep(max(5.0, Config.XAPI_RESULT_POLL_INTERVAL_SECONDS)) +def retry_failed_acceptance_devices(public_id: str) -> dict[str, Any]: + """仅对截图失败的“关键词 × 设备”组合重新立即下发。""" + with connection() as conn: + task = conn.execute( + "SELECT * FROM acceptance_tasks WHERE public_task_id=?", (public_id,), + ).fetchone() + if not task: + raise TaskError("ACCEPTANCE_TASK_NOT_FOUND", "验收任务不存在", 404) + if task["acceptance_method"] not in {"screenshot", "combined"}: + raise TaskError("SCREENSHOT_RETRY_NOT_SUPPORTED", "该任务不包含截图采集", 409) + rows = conn.execute( + """SELECT * FROM acceptance_xapi_runs + WHERE task_id=? ORDER BY id""", + (task["id"],), + ).fetchall() + keyword_runs = { + row["keyword"]: row["id"] for row in conn.execute( + """SELECT id,keyword FROM acceptance_keyword_runs + WHERE task_id=? AND run_type='screenshot'""", + (task["id"],), + ).fetchall() + } + latest: dict[tuple[str, str], Any] = {} + for row in rows: + keyword = str(row["keyword"] or "").strip() + account_id = str(row["xhs_account_id"] or "").strip() + if keyword and account_id: + latest[(keyword, account_id)] = row + failed_by_keyword: dict[str, list[str]] = {} + terminal_failures = {"failed", "cancelled", "canceled", "error"} + for (keyword, account_id), row in latest.items(): + status = str(row["status"] or "").lower() + artifacts = _decode_json(row["artifacts_json"], []) or [] + has_screenshot = any( + str(item.get("image_url") or item.get("url") or "").strip() + for item in artifacts if isinstance(item, dict) + ) + if status in terminal_failures or (status == "completed" and not has_screenshot): + failed_by_keyword.setdefault(keyword, []).append(account_id) + if not failed_by_keyword: + raise TaskError("NO_FAILED_SCREENSHOT_DEVICES", "当前没有可重新采集的失败设备", 409) + + dispatched_devices: set[str] = set() + dispatched_pairs = 0 + errors: list[str] = [] + for keyword, account_ids in failed_by_keyword.items(): + run_id = keyword_runs.get(keyword) + if not run_id: + errors.append(f"{keyword}: 缺少关键词运行记录") + continue + try: + result = schedule_xapi_screenshots( + [keyword], trigger_source="acceptance", xhs_account_ids=account_ids, + ) + batches = result.get("batches") or [] + batch_ids = [ + str(batch.get("batch_id") or "").strip() + for batch in batches if str(batch.get("batch_id") or "").strip() + ] + if not batch_ids: + raise RuntimeError("Framework 未返回可回查的 batch_id") + _update_run( + run_id, status="waiting", result={**result, "batches": batches}, + crawler_task_id=",".join(batch_ids), error=None, + ) + dispatched_devices.update(account_ids) + dispatched_pairs += len(account_ids) + except Exception as exc: + errors.append(f"{keyword}: {exc}") + if not dispatched_pairs: + raise TaskError("FAILED_DEVICE_RETRY_DISPATCH_FAILED", "；".join(errors)[:2000], 502) + with connection() as conn: + conn.execute( + """UPDATE acceptance_tasks SET status='running',progress_percent=70, + error_message=?,updated_at=? WHERE id=?""", + ("；".join(errors)[:2000] if errors else None, now_iso(), task["id"]), + ) + threading.Thread( + target=_poll_screenshot_results, args=(public_id,), daemon=True, + name=f"acceptance-retry-failed-{public_id}", + ).start() + result_task = get_acceptance_task(public_id) + result_task["retry_dispatched_device_count"] = len(dispatched_devices) + result_task["retry_dispatched_pair_count"] = dispatched_pairs + result_task["retry_dispatch_errors"] = errors + return result_task + + def create_acceptance_task(data: dict[str, Any]) -> dict[str, Any]: links = list(dict.fromkeys(_validate_url(item) for item in data.get("feishu_links", []))) method = str(data.get("acceptance_method") or "") @@ -918,24 +1022,56 @@ 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]: +def _normalize_schedule_weekdays(value: Any) -> list[int]: + raw = value if isinstance(value, (list, tuple, set)) else [value] try: - weekday = int(weekday) + weekdays = sorted(set(int(item) for item in raw)) except (TypeError, ValueError) as exc: raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422) from exc - if weekday not in WEEKDAYS: - raise TaskError("INVALID_WEEKDAY", "采集周期必须选择周一至周日", 422) + 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,), @@ -944,8 +1080,14 @@ def update_acceptance_schedule(public_id: str, weekday: int) -> dict[str, Any]: 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=?""", - (weekday, _next_weekly_run(weekday), now_iso(), task["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) @@ -962,6 +1104,40 @@ def stop_acceptance_schedule(public_id: str) -> dict[str, Any]: 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( @@ -1012,21 +1188,39 @@ def collect_acceptance_now(public_id: str) -> dict[str, Any]: def run_due_acceptance_schedules() -> int: - now = now_iso() + current = datetime.now().astimezone() + now = current.isoformat(timespec="seconds") with connection() as conn: due = conn.execute( - """SELECT public_task_id,schedule_weekday FROM acceptance_tasks + """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=NULL,updated_at=? WHERE public_task_id=?""", - (_next_weekly_run(row["schedule_weekday"]), now, now, row["public_task_id"]), + 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"], + ), ) - for row in due: + 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: @@ -1035,7 +1229,7 @@ def run_due_acceptance_schedules() -> int: "UPDATE acceptance_tasks SET schedule_error=?,updated_at=? WHERE public_task_id=?", (str(exc)[:2000], now_iso(), row["public_task_id"]), ) - return len(due) + return len(runnable) def start_acceptance_schedule_worker() -> None: @@ -1065,6 +1259,8 @@ def _fetch_sources(public_id: str) -> None: "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 @@ -1080,6 +1276,8 @@ def _fetch_sources(public_id: str) -> None: 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", @@ -1152,6 +1350,8 @@ 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 _acceptance_task_is_cancelled(public_id): + return if not sources or failures < len(sources): try: keywords = _sync_acceptance_keywords(task["id"], discovered_keywords) @@ -1189,6 +1389,8 @@ def _fetch_sources(public_id: str) -> None: 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, @@ -1209,7 +1411,8 @@ def _fetch_sources(public_id: str) -> None: ])) or None with connection() as conn: conn.execute( - "UPDATE acceptance_tasks SET status=?,progress_percent=100,error_message=?,updated_at=? WHERE public_task_id=?", + """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), ) @@ -1282,6 +1485,8 @@ def _resume_data_analysis_worker(public_id: str) -> None: ).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"], @@ -1303,7 +1508,7 @@ def _resume_data_analysis_worker(public_id: str) -> None: error = None if status == "completed" else "Cookie 补采后仍有部分数据未完成" conn.execute( """UPDATE acceptance_tasks SET status=?,progress_percent=100, - error_message=?,updated_at=? WHERE id=?""", + error_message=?,updated_at=? WHERE id=? AND status!='cancelled'""", (status, error, now_iso(), task["id"]), ) @@ -1370,9 +1575,13 @@ def _task_dict(conn, task) -> dict[str, Any]: resource_progress = [] for resource in task_resources: if resource["resource_type"] == "device": - related_runs = [ + 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" @@ -1403,7 +1612,7 @@ def _task_dict(conn, task) -> dict[str, Any]: "keyword_failed": failed_keywords, "keyword_running": running_keywords, "progress_percent": round(terminal_keywords / max(1, total_keywords) * 100), - "attempt_count": len(related_runs), + "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, @@ -1456,6 +1665,11 @@ def _task_dict(conn, task) -> dict[str, Any]: "cookie_count": task["cookie_count"], "schedule_enabled": bool(task["schedule_enabled"]), "schedule_weekday": task["schedule_weekday"], + "schedule_weekdays": _decode_json(task["schedule_weekdays"], None) or ( + [task["schedule_weekday"]] if task["schedule_weekday"] else [] + ), + "schedule_start_time": task["schedule_start_time"] or "00:00", + "schedule_end_time": task["schedule_end_time"] or "23:59", "next_run_at": task["next_run_at"], "last_run_at": task["last_run_at"], "scheduled_from_task_id": task["scheduled_from_task_id"], "schedule_error": task["schedule_error"], diff --git a/apifox-business-tasks.openapi.json b/apifox-business-tasks.openapi.json index 1339af6..54dc773 100644 --- a/apifox-business-tasks.openapi.json +++ b/apifox-business-tasks.openapi.json @@ -2,7 +2,7 @@ "openapi": "3.0.3", "info": { "title": "Holy蟹业务任务接口", - "version": "1.1.0", + "version": "1.2.0", "description": "前端调用的业务后端任务接口。继续采集沿用原任务 ID；重新采集创建新任务。" }, "servers": [ @@ -37,15 +37,69 @@ "responses": {"201": {"description": "新一轮验收任务已创建"}} } }, + "/api/v1/acceptance-tasks/{task_id}/cancel": { + "post": { + "summary": "停止项目验收采集", + "description": "仅停止当前等待中或运行中的采集批次，并阻止后台解析与截图轮询继续反写任务状态；不会修改该项目已设置的周期计划。已经下发到设备的外部指令可能仍会执行。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "requestBody": { + "required": false, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {"reason": {"type": "string"}} + } + } + } + }, + "responses": { + "200": {"description": "验收任务已停止"}, + "409": {"description": "任务已经进入不可停止的终态"} + } + } + }, + "/api/v1/acceptance-tasks/{task_id}/retry-failed-devices": { + "post": { + "summary": "重新采集全部失败截图设备", + "description": "按每个关键词读取最新 XAPI 结果，仅将失败、取消、异常或已完成但无截图的设备精确重新下发。成功设备不会重复执行，新批次继续归入原验收任务。", + "tags": ["项目验收"], + "parameters": [{"$ref": "#/components/parameters/TaskId"}], + "responses": { + "200": {"description": "失败设备重新下发成功"}, + "409": {"description": "没有失败设备或任务不支持截图重采"} + } + } + }, "/api/v1/acceptance-tasks/{task_id}/schedule": { "put": { - "summary": "设置每周采集周期", - "description": "按 Asia/Shanghai 时区，在指定星期的 00:00 创建新一轮采集。", + "summary": "设置多日周期采集", + "description": "按服务所在时区，在选中的每个星期于开始时间创建新一轮采集；结束时间用于限制过期任务，超过区间后自动顺延到下一次。兼容旧版 weekday 单值。", "tags": ["项目验收"], "parameters": [{"$ref": "#/components/parameters/TaskId"}], "requestBody": { "required": true, - "content": {"application/json": {"schema": {"type": "object", "required": ["weekday"], "properties": {"weekday": {"type": "integer", "minimum": 1, "maximum": 7, "description": "1=周一，7=周日"}}}}} + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["weekdays", "start_time", "end_time"], + "properties": { + "weekdays": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"type": "integer", "minimum": 1, "maximum": 7}, + "description": "可多选；1=周一，7=周日" + }, + "start_time": {"type": "string", "pattern": "^([01]\\d|2[0-3]):[0-5]\\d$", "example": "01:00"}, + "end_time": {"type": "string", "pattern": "^([01]\\d|2[0-3]):[0-5]\\d$", "example": "04:30"} + } + }, + "example": {"weekdays": [1, 3, 5], "start_time": "01:00", "end_time": "04:30"} + } + } }, "responses": {"200": {"description": "周期设置成功"}} } diff --git a/app.py b/app.py index 07d28e4..05fc67a 100644 --- a/app.py +++ b/app.py @@ -28,6 +28,7 @@ from ocr_dashboard_service import ( list_results as list_ocr_results, ) from acceptance_service import ( + cancel_acceptance_task, create_acceptance_task, collect_acceptance_now, get_acceptance_resources, @@ -36,6 +37,7 @@ from acceptance_service import ( list_acceptance_notes, list_acceptance_tasks, resume_acceptance_task, + retry_failed_acceptance_devices, resume_unfinished_acceptance_tasks, start_acceptance_schedule_worker, stop_acceptance_schedule, @@ -314,15 +316,35 @@ def create_app() -> Flask: def resume_project_acceptance_task(task_id: str): return jsonify({"data": resume_acceptance_task(task_id), "request_id": g.request_id}) + @app.route("/api/v1/acceptance-tasks//cancel", methods=["POST"]) + def cancel_project_acceptance_task(task_id: str): + payload = request.get_json(silent=True) or {} + return jsonify({ + "data": cancel_acceptance_task(task_id, str(payload.get("reason") or "")), + "request_id": g.request_id, + }) + @app.route("/api/v1/acceptance-tasks//collect-now", methods=["POST"]) def collect_project_acceptance_now(task_id: str): return jsonify({"data": collect_acceptance_now(task_id), "request_id": g.request_id}), 201 + @app.route("/api/v1/acceptance-tasks//retry-failed-devices", methods=["POST"]) + def retry_project_acceptance_failed_devices(task_id: str): + return jsonify({ + "data": retry_failed_acceptance_devices(task_id), + "request_id": g.request_id, + }) + @app.route("/api/v1/acceptance-tasks//schedule", methods=["PUT"]) def configure_project_acceptance_schedule(task_id: str): payload = request.get_json(silent=True) or {} return jsonify({ - "data": update_acceptance_schedule(task_id, payload.get("weekday")), + "data": update_acceptance_schedule( + task_id, + payload.get("weekdays", payload.get("weekday")), + payload.get("start_time", "00:00"), + payload.get("end_time", "23:59"), + ), "request_id": g.request_id, }) diff --git a/config.py b/config.py index 25c657c..974dbee 100644 --- a/config.py +++ b/config.py @@ -32,6 +32,7 @@ class Config: ).rstrip("/") SEARCH_QA_API_KEY = os.getenv("SEARCH_QA_API_KEY", "") SEARCH_QA_TIMEOUT_SECONDS = int(os.getenv("SEARCH_QA_TIMEOUT_SECONDS", "60")) + EVIDENCE_ADMIN_BASE_URL = os.getenv("EVIDENCE_ADMIN_BASE_URL", "http://127.0.0.1:8100").rstrip("/") CORS_ORIGIN = os.getenv("CORS_ORIGIN", "http://localhost:5173") FRONTEND_BASE_URL = os.getenv("FRONTEND_BASE_URL", CORS_ORIGIN).rstrip("/") SESSION_COOKIE_NAME = os.getenv("SESSION_COOKIE_NAME", "hc_session") diff --git a/crawler_client.py b/crawler_client.py index 3f4de72..242a1f3 100644 --- a/crawler_client.py +++ b/crawler_client.py @@ -327,18 +327,27 @@ def fetch_content_detail(note_url: str) -> dict[str, Any]: def schedule_xapi_screenshots( - search_queries: list[str], device_count: int, *, trigger_source: str = "acceptance", + search_queries: list[str], device_count: int | None = None, *, + trigger_source: str = "acceptance", + xhs_account_ids: list[str] | None = None, ) -> dict[str, Any]: """向手机集群立即下发关键词搜索与截图任务。""" queries = normalize_collection_keywords(search_queries) + account_ids = list(dict.fromkeys( + str(item or "").strip() for item in (xhs_account_ids or []) if str(item or "").strip() + )) + payload: dict[str, Any] = { + "keywords": queries, + "execute_now": True, + "trigger_source": trigger_source, + } + if account_ids: + payload["xhs_account_ids"] = account_ids + else: + payload["device_count"] = int(device_count or 0) return _post( "/api/v1/xhs/wen/xapi/schedule", - { - "keywords": queries, - "device_count": device_count, - "execute_now": True, - "trigger_source": trigger_source, - }, + payload, retry_transient=False, ) diff --git a/database.py b/database.py index 3380046..f46a0c8 100644 --- a/database.py +++ b/database.py @@ -377,6 +377,9 @@ def init_database() -> None: acceptance_migrations = { "schedule_enabled": "INTEGER NOT NULL DEFAULT 0", "schedule_weekday": "INTEGER", + "schedule_weekdays": "TEXT", + "schedule_start_time": "TEXT NOT NULL DEFAULT '00:00'", + "schedule_end_time": "TEXT NOT NULL DEFAULT '23:59'", "next_run_at": "TEXT", "last_run_at": "TEXT", "scheduled_from_task_id": "INTEGER", @@ -387,6 +390,15 @@ def init_database() -> None: for column, definition in acceptance_migrations.items(): if column not in acceptance_columns: conn.execute(f"ALTER TABLE acceptance_tasks ADD COLUMN {column} {definition}") + legacy_schedules = conn.execute( + """SELECT id,schedule_weekday FROM acceptance_tasks + WHERE schedule_weekday IS NOT NULL + AND (schedule_weekdays IS NULL OR schedule_weekdays='')""" + ).fetchall() + conn.executemany( + "UPDATE acceptance_tasks SET schedule_weekdays=? WHERE id=?", + [(json.dumps([row["schedule_weekday"]]), row["id"]) for row in legacy_schedules], + ) # 将旧版本保存在 acceptance_xhs_notes.parsed_json 中的成功结果迁移到 # 全局去重的笔记业务表；任务关联仍由 acceptance_xhs_notes 保留。 legacy_rows = conn.execute( diff --git a/ocr_dashboard_service.py b/ocr_dashboard_service.py index 31107dd..511ca37 100644 --- a/ocr_dashboard_service.py +++ b/ocr_dashboard_service.py @@ -102,9 +102,54 @@ def list_evidences(params: dict[str, Any]) -> dict[str, Any]: if not 1 <= limit <= 100: raise TaskError("OCR_EVIDENCE_LIMIT_INVALID", "limit 必须在 1 到 100 之间", 422) query["limit"] = limit - return _get("/xhs/ocr/dashboard/evidences", query) or { + data = _get("/xhs/ocr/dashboard/evidences", query) or { "items": [], "total": 0, "limit": limit, "is_complete": False, } + if Config.EVIDENCE_ADMIN_BASE_URL and query.get("batch_key") and query.get("question_code"): + try: + response = SESSION.get( + f"{Config.EVIDENCE_ADMIN_BASE_URL}/api/evidences", + params={"batch_key": query["batch_key"], "question_code": query["question_code"]}, + timeout=min(Config.CRAWLER_TIMEOUT_SECONDS, 10), + ) + response.raise_for_status() + manual_items = (response.json().get("data") or []) + items = data.setdefault("items", []) + existing_urls = {str(item.get("cos_url") or "") for item in items} + object_indexes = { + str(item.get("cos_object_key")): index for index, item in enumerate(items) + if item.get("cos_object_key") + } + for item in manual_items: + url = str(item.get("cos_url") or "") + object_key = str(item.get("cos_object_key") or "") + if object_key and object_key in object_indexes: + items[object_indexes[object_key]] = item + existing_urls.add(url) + elif url and url not in existing_urls: + items.append(item) + if object_key: + object_indexes[object_key] = len(items) - 1 + existing_urls.add(url) + evidence_type_order = { + "body_ranking": 0, "first_screen": 1, "reference_source": 2, + "before": 3, "after": 3, "before_after": 3, "viral_note": 4, + } + items.sort(key=lambda item: ( + evidence_type_order.get(str(item.get("evidence_type") or ""), 99), + 1 if ( + item.get("image_id") is not None + or "/manual-evidences/" in str(item.get("cos_object_key") or "") + or item.get("upload_source") == "holy-backend" + ) else 0, + str(item.get("created_at") or ""), + int(item.get("sort_order") or 0), + int(item.get("image_id") or item.get("evidence_id") or item.get("id") or 0), + )) + data["total"] = len(data.get("items") or []) + except (requests.RequestException, ValueError): + pass + return data def export_evidences(params: dict[str, Any]) -> tuple[io.BytesIO, str, int]: