1923 lines
80 KiB
Python
1923 lines
80 KiB
Python
"""PDD 任务的 SQLite Repository。
|
||
|
||
Repository 是数据库访问入口。界面和自动化代码不应自行拼接任务 SQL。
|
||
"""
|
||
|
||
import json
|
||
import sqlite3
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Dict, Iterable, List, Optional, Tuple, Union
|
||
from uuid import uuid4
|
||
|
||
from .db import initialize_database, open_database
|
||
from .task_models import (
|
||
NewClaimedTask,
|
||
OutboxEventRecord,
|
||
OutboxEventType,
|
||
OutboxStatus,
|
||
RunStatus,
|
||
StartedTaskRun,
|
||
TaskDetail,
|
||
TaskFilters,
|
||
TaskRerunPlan,
|
||
TaskStatus,
|
||
TaskSummary,
|
||
TaskType,
|
||
TaskRunRecord,
|
||
)
|
||
|
||
|
||
PathValue = Union[str, Path]
|
||
MAX_PAGE_SIZE = 500
|
||
|
||
|
||
class DuplicateTaskError(ValueError):
|
||
"""相同远程任务编号已经存在，不能覆盖。"""
|
||
|
||
|
||
class CollectRerunError(ValueError):
|
||
"""当前任务不满足重新采集条件。"""
|
||
|
||
|
||
class PurchaseRerunError(ValueError):
|
||
"""当前任务不满足安全重新采购条件。"""
|
||
|
||
|
||
class TaskRemovalError(ValueError):
|
||
"""勾选任务不满足从普通列表移除的安全条件。"""
|
||
|
||
|
||
def utc_now_iso() -> str:
|
||
"""返回精确到秒的 UTC ISO 8601 时间。"""
|
||
|
||
return datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
|
||
"+00:00", "Z"
|
||
)
|
||
|
||
|
||
class TaskRepository:
|
||
"""保存和查询本机已经领取的 PDD 任务。"""
|
||
|
||
def __init__(self, db_path: Optional[PathValue] = None):
|
||
self._db_path = initialize_database(db_path)
|
||
|
||
def add_claimed_task(
|
||
self, task: NewClaimedTask, received_at: Optional[str] = None
|
||
) -> int:
|
||
"""写入一条新领取任务，返回本地自增编号。
|
||
|
||
相同 ``remote_task_id`` 已存在时抛出 ``DuplicateTaskError``，
|
||
不覆盖已经保存的本地状态。
|
||
"""
|
||
|
||
now = received_at or utc_now_iso()
|
||
payload = json.dumps(task.admin_payload, ensure_ascii=False)
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
try:
|
||
with connection:
|
||
cursor = connection.execute(
|
||
"INSERT INTO pdd_tasks ("
|
||
" remote_task_id, task_type, execution_mode, goods_id,"
|
||
" goods_url, title,"
|
||
" target_color, target_size, price_cent, quantity, status,"
|
||
" priority, version, admin_payload, received_at, created_at,"
|
||
" updated_at"
|
||
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(
|
||
task.remote_task_id.strip(),
|
||
task.task_type.value,
|
||
task.execution_mode,
|
||
task.goods_id,
|
||
task.goods_url.strip(),
|
||
task.title,
|
||
task.target_color,
|
||
task.target_size,
|
||
task.price_cent,
|
||
task.quantity,
|
||
TaskStatus.CLAIMED.value,
|
||
task.priority,
|
||
task.version,
|
||
payload,
|
||
now,
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
return int(cursor.lastrowid)
|
||
except sqlite3.IntegrityError as exc:
|
||
if "pdd_tasks.remote_task_id" in str(exc):
|
||
raise DuplicateTaskError(
|
||
f"任务 {task.remote_task_id} 已经存在"
|
||
) from exc
|
||
raise
|
||
finally:
|
||
connection.close()
|
||
|
||
def list_tasks(
|
||
self,
|
||
filters: Optional[TaskFilters] = None,
|
||
limit: int = 50,
|
||
offset: int = 0,
|
||
) -> List[TaskSummary]:
|
||
"""分页查询任务摘要，默认按更新时间和本地编号倒序。"""
|
||
|
||
self._validate_page(limit, offset)
|
||
where_sql, parameters = self._build_where(filters or TaskFilters())
|
||
parameters.extend((limit, offset))
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
rows = connection.execute(
|
||
"SELECT pdd_tasks.id, remote_task_id, task_type, goods_id, title,"
|
||
" shop_name, target_color, target_size, price_cent, quantity,"
|
||
" status, pdd_tasks.updated_at, latest_run.run_status AS latest_run_status,"
|
||
" CASE WHEN latest_run.run_status = 'running' THEN NULL"
|
||
" WHEN latest_run.started_at IS NOT NULL AND latest_run.finished_at IS NOT NULL"
|
||
" THEN MAX(0, CAST(strftime('%s', latest_run.finished_at)"
|
||
" - strftime('%s', latest_run.started_at) AS INTEGER))"
|
||
" ELSE NULL END AS duration_seconds,"
|
||
" CASE WHEN pdd_tasks.pdd_data IS NOT NULL"
|
||
" AND json_valid(pdd_tasks.pdd_data)"
|
||
" THEN CAST(json_extract(pdd_tasks.pdd_data,"
|
||
" '$.purchase.order_no') AS TEXT)"
|
||
" ELSE NULL END AS order_no"
|
||
" FROM pdd_tasks"
|
||
" LEFT JOIN task_runs AS latest_run ON latest_run.id = ("
|
||
" SELECT id FROM task_runs WHERE task_id = pdd_tasks.id"
|
||
" ORDER BY attempt_no DESC LIMIT 1)"
|
||
f"{where_sql}"
|
||
" ORDER BY pdd_tasks.updated_at DESC, pdd_tasks.id DESC"
|
||
" LIMIT ? OFFSET ?",
|
||
parameters,
|
||
).fetchall()
|
||
finally:
|
||
connection.close()
|
||
return [self._to_summary(row) for row in rows]
|
||
|
||
def count_tasks(self, filters: Optional[TaskFilters] = None) -> int:
|
||
"""返回符合筛选条件的任务总数。"""
|
||
|
||
where_sql, parameters = self._build_where(filters or TaskFilters())
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
f"SELECT COUNT(*) FROM pdd_tasks{where_sql}", parameters
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return int(row[0])
|
||
|
||
def remove_tasks_from_list(self, remote_task_ids: Iterable[str]) -> int:
|
||
"""把符合安全条件的终态任务从普通列表中软移除。
|
||
|
||
全部任务会在同一个事务中完成校验和更新；任一任务不安全时，
|
||
所有任务都保持原样。执行记录和 Outbox 不会被删除或修改。
|
||
"""
|
||
|
||
task_ids = tuple(
|
||
dict.fromkeys(
|
||
str(value).strip()
|
||
for value in remote_task_ids
|
||
if value is not None and str(value).strip()
|
||
)
|
||
)
|
||
if not task_ids:
|
||
raise TaskRemovalError("没有可删除的任务，请重新勾选。")
|
||
|
||
placeholders = ", ".join("?" for _ in task_ids)
|
||
terminal_statuses = {
|
||
TaskStatus.SUCCEEDED.value,
|
||
TaskStatus.FAILED.value,
|
||
TaskStatus.CANCELLED.value,
|
||
}
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
rows = connection.execute(
|
||
"SELECT id, remote_task_id, status, removed_at FROM pdd_tasks"
|
||
f" WHERE remote_task_id IN ({placeholders})",
|
||
task_ids,
|
||
).fetchall()
|
||
found = {row["remote_task_id"]: row for row in rows}
|
||
missing = [task_id for task_id in task_ids if task_id not in found]
|
||
if missing:
|
||
raise TaskRemovalError(
|
||
f"本地找不到任务 {missing[0]}，请刷新列表后重试。"
|
||
)
|
||
removed = next(
|
||
(row for row in rows if row["removed_at"] is not None), None
|
||
)
|
||
if removed is not None:
|
||
raise TaskRemovalError(
|
||
f"任务 {removed['remote_task_id']} 已不在普通列表，请刷新后重试。"
|
||
)
|
||
non_terminal = next(
|
||
(row for row in rows if row["status"] not in terminal_statuses),
|
||
None,
|
||
)
|
||
if non_terminal is not None:
|
||
raise TaskRemovalError(
|
||
f"任务 {non_terminal['remote_task_id']} 尚未结束，不能删除。"
|
||
)
|
||
|
||
irreversible = connection.execute(
|
||
"SELECT t.remote_task_id FROM task_runs r"
|
||
" JOIN pdd_tasks t ON t.id = r.task_id"
|
||
f" WHERE t.remote_task_id IN ({placeholders})"
|
||
" AND r.irreversible_action_at IS NOT NULL LIMIT 1",
|
||
task_ids,
|
||
).fetchone()
|
||
if irreversible is not None:
|
||
raise TaskRemovalError(
|
||
f"任务 {irreversible['remote_task_id']} 已进入不可逆阶段，"
|
||
"必须保留在列表中核对订单。"
|
||
)
|
||
|
||
unsent = connection.execute(
|
||
"SELECT t.remote_task_id, o.status FROM outbox_events o"
|
||
" JOIN pdd_tasks t ON t.id = o.task_id"
|
||
f" WHERE t.remote_task_id IN ({placeholders})"
|
||
" AND o.status <> 'sent' LIMIT 1",
|
||
task_ids,
|
||
).fetchone()
|
||
if unsent is not None:
|
||
raise TaskRemovalError(
|
||
f"任务 {unsent['remote_task_id']} 还有未发送完成的数据，"
|
||
"请先重新上报。"
|
||
)
|
||
|
||
cursor = connection.execute(
|
||
"UPDATE pdd_tasks SET removed_at = ?, updated_at = ?"
|
||
f" WHERE remote_task_id IN ({placeholders})"
|
||
" AND removed_at IS NULL",
|
||
(now, now, *task_ids),
|
||
)
|
||
if cursor.rowcount != len(task_ids):
|
||
raise TaskRemovalError("任务列表已发生变化，请刷新后重试。")
|
||
finally:
|
||
connection.close()
|
||
return len(task_ids)
|
||
|
||
def get_task(self, remote_task_id: str) -> Optional[TaskDetail]:
|
||
"""按稳定远程编号读取完整任务；不存在时返回 None。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row) if row is not None else None
|
||
|
||
def recover_interrupted_work(self) -> None:
|
||
"""恢复上次异常退出留下的任务，不猜测内存状态。"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
connection.execute(
|
||
"UPDATE outbox_events SET status = 'pending', updated_at = ?"
|
||
" WHERE status = 'sending'",
|
||
(now,),
|
||
)
|
||
# #35 之前，Admin 的失败响应缺少 result_id。Admin 实际已经
|
||
# 接收，但旧 Client 把这类 2xx 响应误标为永久失败。恢复为
|
||
# pending 后使用原幂等键重试，不会重复写入业务结果。
|
||
connection.execute(
|
||
"UPDATE outbox_events SET status = 'pending', updated_at = ?"
|
||
" WHERE status = 'failed'"
|
||
" AND last_error = 'Admin 提交响应字段不完整'",
|
||
(now,),
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'retry_wait',"
|
||
" current_step = 'interrupted', retry_count = retry_count + 1,"
|
||
" last_error_code = 'CLIENT_INTERRUPTED',"
|
||
" last_error_message = '客户端上次执行期间退出', updated_at = ?"
|
||
" WHERE status = 'running' AND task_type = 'collect'",
|
||
(now,),
|
||
)
|
||
connection.execute(
|
||
"UPDATE task_runs SET run_status = 'failed',"
|
||
" error_code = 'CLIENT_INTERRUPTED',"
|
||
" error_message = '客户端上次执行期间退出',"
|
||
" finished_at = ?, updated_at = ?"
|
||
" WHERE run_status = 'running' AND irreversible_action_at IS NULL"
|
||
" AND task_id IN (SELECT id FROM pdd_tasks WHERE task_type = 'collect')",
|
||
(now, now),
|
||
)
|
||
purchase_runs = connection.execute(
|
||
"SELECT t.id AS task_id, r.irreversible_action_at"
|
||
" FROM pdd_tasks t JOIN task_runs r ON r.task_id = t.id"
|
||
" WHERE t.task_type = 'purchase' AND t.status = 'running'"
|
||
" AND r.run_status = 'running'"
|
||
" ORDER BY r.attempt_no DESC"
|
||
).fetchall()
|
||
recovered_task_ids = set()
|
||
for run in purchase_runs:
|
||
task_id = int(run["task_id"])
|
||
if task_id in recovered_task_ids:
|
||
continue
|
||
recovered_task_ids.add(task_id)
|
||
irreversible = connection.execute(
|
||
"SELECT 1 FROM task_runs WHERE task_id = ?"
|
||
" AND run_status = 'running'"
|
||
" AND irreversible_action_at IS NOT NULL LIMIT 1",
|
||
(task_id,),
|
||
).fetchone()
|
||
if irreversible is not None:
|
||
message = (
|
||
"上次采购在不可逆阶段中断，"
|
||
"只允许核对订单"
|
||
)
|
||
connection.execute(
|
||
"UPDATE task_runs SET run_status = 'manual_review',"
|
||
" current_step = 'reconcile_purchase',"
|
||
" error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
|
||
" error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND run_status = 'running'",
|
||
(message, now, now, task_id),
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'manual_review',"
|
||
" current_step = 'reconcile_purchase',"
|
||
" last_error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
|
||
" last_error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(message, now, now, task_id),
|
||
)
|
||
else:
|
||
message = "采购演练上次执行中断，已关闭旧执行记录"
|
||
connection.execute(
|
||
"UPDATE task_runs SET run_status = 'failed',"
|
||
" error_code = 'CLIENT_INTERRUPTED',"
|
||
" error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND run_status = 'running'",
|
||
(message, now, now, task_id),
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'claimed',"
|
||
" current_step = 'purchase_recovery_ready',"
|
||
" retry_count = retry_count + 1,"
|
||
" last_error_code = 'CLIENT_INTERRUPTED',"
|
||
" last_error_message = ?, finished_at = NULL,"
|
||
" updated_at = ? WHERE id = ?",
|
||
(message, now, task_id),
|
||
)
|
||
finally:
|
||
connection.close()
|
||
|
||
def next_collect_task(self) -> Optional[TaskDetail]:
|
||
"""返回最早的本地待执行采集任务。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks"
|
||
" WHERE task_type = 'collect' AND status = 'claimed'"
|
||
" ORDER BY received_at ASC, id ASC LIMIT 1"
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row) if row is not None else None
|
||
|
||
def next_purchase_task(self) -> Optional[TaskDetail]:
|
||
"""返回最早的一条本地待执行采购任务。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks"
|
||
" WHERE task_type = 'purchase' AND status = 'claimed'"
|
||
" ORDER BY received_at ASC, id ASC LIMIT 1"
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row) if row is not None else None
|
||
|
||
def next_purchase_reconcile_task(self) -> Optional[TaskDetail]:
|
||
"""返回最早一条只允许读取核对的采购任务。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE task_type = 'purchase'"
|
||
" AND status = 'manual_review'"
|
||
" AND current_step = 'reconcile_purchase'"
|
||
" ORDER BY received_at ASC, id ASC LIMIT 1"
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row) if row is not None else None
|
||
|
||
def unresolved_irreversible_purchase(self) -> Optional[TaskDetail]:
|
||
"""查找仍在运行且已有不可逆标记的采购，防止继续控制设备。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT t.* FROM pdd_tasks t JOIN task_runs r ON r.task_id = t.id"
|
||
" WHERE t.task_type = 'purchase' AND t.status = 'running'"
|
||
" AND r.run_status = 'running'"
|
||
" AND r.irreversible_action_at IS NOT NULL"
|
||
" ORDER BY r.attempt_no DESC LIMIT 1"
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row) if row is not None else None
|
||
|
||
def latest_task_run(
|
||
self, remote_task_id: str
|
||
) -> Optional[TaskRunRecord]:
|
||
"""返回任务最新的执行记录。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT r.* FROM task_runs r"
|
||
" JOIN pdd_tasks t ON t.id = r.task_id"
|
||
" WHERE t.remote_task_id = ?"
|
||
" ORDER BY r.attempt_no DESC LIMIT 1",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_task_run(row) if row is not None else None
|
||
|
||
def next_runnable_task(
|
||
self, *, include_purchase: bool
|
||
) -> Optional[TaskDetail]:
|
||
"""按领取顺序返回当前执行器能够处理的最早本地任务。"""
|
||
|
||
if include_purchase:
|
||
where = (
|
||
"((task_type = 'collect' AND status = 'claimed')"
|
||
" OR (task_type = 'purchase' AND status = 'claimed'))"
|
||
)
|
||
else:
|
||
where = "task_type = 'collect' AND status = 'claimed'"
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
f"SELECT * FROM pdd_tasks WHERE {where}"
|
||
" ORDER BY received_at ASC, id ASC LIMIT 1"
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row) if row is not None else None
|
||
|
||
def validate_collect_rerun(self, remote_task_id: str) -> TaskDetail:
|
||
"""校验任务能否重新采集，成功时返回任务详情。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
raise CollectRerunError(f"任务 {remote_task_id} 不存在")
|
||
self._check_collect_rerun(connection, row)
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row)
|
||
|
||
def plan_rerun_batch(
|
||
self, remote_task_ids: Iterable[str], target_type: TaskType
|
||
) -> TaskRerunPlan:
|
||
"""批量预检并分类：可执行、类型不符、被安全条件阻止。"""
|
||
|
||
stable_ids = tuple(dict.fromkeys(str(value) for value in remote_task_ids if value))
|
||
eligible: List[str] = []
|
||
filtered: List[str] = []
|
||
blocked: List[Tuple[str, str]] = []
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
for task_id in stable_ids:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?", (task_id,)
|
||
).fetchone()
|
||
if row is None:
|
||
blocked.append((task_id, "任务不存在"))
|
||
continue
|
||
if row["task_type"] != target_type.value:
|
||
filtered.append(task_id)
|
||
continue
|
||
try:
|
||
if target_type is TaskType.COLLECT:
|
||
self._check_collect_rerun(connection, row)
|
||
else:
|
||
self._check_purchase_rerun(connection, row)
|
||
except (CollectRerunError, PurchaseRerunError) as exc:
|
||
blocked.append((task_id, str(exc)))
|
||
else:
|
||
eligible.append(task_id)
|
||
finally:
|
||
connection.close()
|
||
return TaskRerunPlan(
|
||
target_type, len(stable_ids), tuple(eligible), tuple(filtered), tuple(blocked)
|
||
)
|
||
|
||
def prepare_collect_rerun(self, remote_task_id: str) -> TaskDetail:
|
||
"""事务内把已结束的采集任务恢复为待执行，保留旧结果。"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
raise CollectRerunError(f"任务 {remote_task_id} 不存在")
|
||
self._check_collect_rerun(connection, row)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'claimed',"
|
||
" current_step = 'rerun_requested', finished_at = NULL,"
|
||
" last_error_code = NULL, last_error_message = NULL,"
|
||
" updated_at = ? WHERE id = ?",
|
||
(now, row["id"]),
|
||
)
|
||
finally:
|
||
connection.close()
|
||
task = self.get_task(remote_task_id)
|
||
assert task is not None
|
||
return task
|
||
|
||
@staticmethod
|
||
def _check_collect_rerun(
|
||
connection: sqlite3.Connection, row: sqlite3.Row
|
||
) -> None:
|
||
"""检查重新采集的类型、终态和 Outbox 约束。"""
|
||
|
||
if row["task_type"] != TaskType.COLLECT.value:
|
||
raise CollectRerunError("采购任务不能重新执行，以免重复下单")
|
||
allowed_statuses = {
|
||
TaskStatus.CLAIMED.value,
|
||
TaskStatus.RETRY_WAIT.value,
|
||
TaskStatus.SUCCEEDED.value,
|
||
TaskStatus.FAILED.value,
|
||
TaskStatus.CANCELLED.value,
|
||
TaskStatus.MANUAL_REVIEW.value,
|
||
}
|
||
if row["status"] not in allowed_statuses:
|
||
status_name = {
|
||
TaskStatus.CLAIMED.value: "待执行",
|
||
TaskStatus.RUNNING.value: "执行中",
|
||
TaskStatus.RESULT_PENDING.value: "结果待提交",
|
||
}.get(row["status"], row["status"])
|
||
raise CollectRerunError(f"任务当前为“{status_name}”，不能重新采集")
|
||
unsent = connection.execute(
|
||
"SELECT event_type, status, last_error FROM outbox_events"
|
||
" WHERE task_id = ? AND status != 'sent'"
|
||
" ORDER BY id DESC LIMIT 1",
|
||
(row["id"],),
|
||
).fetchone()
|
||
if unsent is not None and unsent["status"] == OutboxStatus.FAILED.value:
|
||
reason = unsent["last_error"] or "Admin 未接收上次结果"
|
||
event_name = (
|
||
"失败信息"
|
||
if unsent["event_type"] == OutboxEventType.TASK_FAILURE.value
|
||
else "结果"
|
||
)
|
||
raise CollectRerunError(
|
||
f"任务上次{event_name}提交失败：{reason}；"
|
||
"请先勾选任务点击“重新上报”"
|
||
)
|
||
if unsent is not None:
|
||
if unsent["event_type"] == OutboxEventType.TASK_FAILURE.value:
|
||
raise CollectRerunError(
|
||
"任务仍有未上报的失败信息，请先勾选任务点击“重新上报”"
|
||
)
|
||
raise CollectRerunError(
|
||
"任务仍有未发送的结果，请先勾选任务点击“重新上报”"
|
||
)
|
||
|
||
def validate_purchase_rerun(self, remote_task_id: str) -> TaskDetail:
|
||
"""校验采购任务能否重新采购；历史上已下单的任务永久拒绝。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?", (remote_task_id,)
|
||
).fetchone()
|
||
if row is None:
|
||
raise PurchaseRerunError(f"任务 {remote_task_id} 不存在")
|
||
self._check_purchase_rerun(connection, row)
|
||
finally:
|
||
connection.close()
|
||
return self._to_detail(row)
|
||
|
||
def prepare_purchase_rerun(self, remote_task_id: str) -> TaskDetail:
|
||
"""事务内恢复安全采购任务；不会清除任何历史执行记录。"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?", (remote_task_id,)
|
||
).fetchone()
|
||
if row is None:
|
||
raise PurchaseRerunError(f"任务 {remote_task_id} 不存在")
|
||
self._check_purchase_rerun(connection, row)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'claimed',"
|
||
" current_step = 'purchase_rerun_requested', finished_at = NULL,"
|
||
" last_error_code = NULL, last_error_message = NULL, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(now, row["id"]),
|
||
)
|
||
finally:
|
||
connection.close()
|
||
task = self.get_task(remote_task_id)
|
||
assert task is not None
|
||
return task
|
||
|
||
@staticmethod
|
||
def _check_purchase_rerun(
|
||
connection: sqlite3.Connection, row: sqlite3.Row
|
||
) -> None:
|
||
if row["task_type"] != TaskType.PURCHASE.value:
|
||
raise PurchaseRerunError("采集任务不能重新采购")
|
||
irreversible = connection.execute(
|
||
"SELECT 1 FROM task_runs WHERE task_id = ?"
|
||
" AND irreversible_action_at IS NOT NULL LIMIT 1",
|
||
(row["id"],),
|
||
).fetchone()
|
||
if irreversible is not None:
|
||
raise PurchaseRerunError("任务历史上已进入下单阶段，只准核对订单，绝不重新下单")
|
||
allowed = {
|
||
TaskStatus.CLAIMED.value,
|
||
TaskStatus.RETRY_WAIT.value,
|
||
TaskStatus.SUCCEEDED.value,
|
||
TaskStatus.FAILED.value,
|
||
TaskStatus.CANCELLED.value,
|
||
TaskStatus.MANUAL_REVIEW.value,
|
||
}
|
||
if row["status"] not in allowed:
|
||
raise PurchaseRerunError(f"任务当前状态 {row['status']} 不能重新采购")
|
||
unsent = connection.execute(
|
||
"SELECT 1 FROM outbox_events WHERE task_id = ? AND status != 'sent' LIMIT 1",
|
||
(row["id"],),
|
||
).fetchone()
|
||
if unsent is not None:
|
||
raise PurchaseRerunError("任务仍有未发送数据，请先点击“重新上报”")
|
||
|
||
def start_collect_run(
|
||
self, remote_task_id: str, device_address: str
|
||
) -> StartedTaskRun:
|
||
"""原子地把待执行任务改为执行中，并创建一次运行记录。"""
|
||
|
||
now = utc_now_iso()
|
||
attempt_id = str(uuid4())
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if row["task_type"] != TaskType.COLLECT.value:
|
||
raise ValueError("当前只能执行采集任务")
|
||
if row["status"] not in {
|
||
TaskStatus.CLAIMED.value,
|
||
TaskStatus.RETRY_WAIT.value,
|
||
}:
|
||
raise ValueError(f"任务状态 {row['status']} 不能开始采集")
|
||
attempt_no = int(
|
||
connection.execute(
|
||
"SELECT COALESCE(MAX(attempt_no), 0) + 1"
|
||
" FROM task_runs WHERE task_id = ?",
|
||
(row["id"],),
|
||
).fetchone()[0]
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'running',"
|
||
" current_step = 'collecting', started_at = COALESCE(started_at, ?),"
|
||
" last_error_code = NULL, last_error_message = NULL, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(now, now, row["id"]),
|
||
)
|
||
connection.execute(
|
||
"INSERT INTO task_runs (task_id, attempt_id, attempt_no,"
|
||
" device_address, run_status, current_step, started_at,"
|
||
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(
|
||
row["id"], attempt_id, attempt_no, device_address,
|
||
RunStatus.RUNNING.value, "collecting", now, now, now,
|
||
),
|
||
)
|
||
task = self.get_task(remote_task_id)
|
||
assert task is not None
|
||
return StartedTaskRun(task, attempt_id, attempt_no)
|
||
finally:
|
||
connection.close()
|
||
|
||
def start_purchase_run(
|
||
self, remote_task_id: str, device_address: str
|
||
) -> StartedTaskRun:
|
||
"""原子地开始一次采购演练并创建独立运行记录。"""
|
||
|
||
now = utc_now_iso()
|
||
attempt_id = str(uuid4())
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
row = connection.execute(
|
||
"SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if row["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
if row["status"] != TaskStatus.CLAIMED.value:
|
||
raise ValueError(
|
||
f"采购任务状态 {row['status']} 不能开始演练"
|
||
)
|
||
attempt_no = int(
|
||
connection.execute(
|
||
"SELECT COALESCE(MAX(attempt_no), 0) + 1"
|
||
" FROM task_runs WHERE task_id = ?",
|
||
(row["id"],),
|
||
).fetchone()[0]
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'running',"
|
||
" current_step = 'purchase_prepare',"
|
||
" started_at = COALESCE(started_at, ?),"
|
||
" last_error_code = NULL, last_error_message = NULL,"
|
||
" updated_at = ? WHERE id = ?",
|
||
(now, now, row["id"]),
|
||
)
|
||
connection.execute(
|
||
"INSERT INTO task_runs (task_id, attempt_id, attempt_no,"
|
||
" device_address, run_status, current_step, started_at,"
|
||
" created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(
|
||
row["id"],
|
||
attempt_id,
|
||
attempt_no,
|
||
device_address,
|
||
RunStatus.RUNNING.value,
|
||
"purchase_prepare",
|
||
now,
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
task = self.get_task(remote_task_id)
|
||
assert task is not None
|
||
return StartedTaskRun(task, attempt_id, attempt_no)
|
||
finally:
|
||
connection.close()
|
||
|
||
def update_purchase_step(
|
||
self, remote_task_id: str, attempt_id: str, step: str
|
||
) -> None:
|
||
"""在进入采购关键步骤前，同时持久化任务和本次运行的步骤。"""
|
||
|
||
checked_step = str(step or "").strip()
|
||
if not checked_step:
|
||
raise ValueError("采购步骤不能为空")
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, task_type, status FROM pdd_tasks"
|
||
" WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
if task["status"] != TaskStatus.RUNNING.value:
|
||
raise ValueError("采购任务当前不在执行中")
|
||
cursor = connection.execute(
|
||
"UPDATE task_runs SET current_step = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND attempt_id = ?"
|
||
" AND run_status = 'running'",
|
||
(checked_step, now, task["id"], attempt_id),
|
||
)
|
||
if cursor.rowcount != 1:
|
||
raise ValueError("采购执行记录不存在或已经结束")
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET current_step = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(checked_step, now, task["id"]),
|
||
)
|
||
finally:
|
||
connection.close()
|
||
|
||
def mark_purchase_irreversible(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
final_confirmation: Optional[Dict[str, object]] = None,
|
||
) -> str:
|
||
"""同一事务保存最终确认快照和不可逆时间，再允许提交订单。"""
|
||
|
||
now = utc_now_iso()
|
||
step = "purchase_irreversible_step_entered"
|
||
diagnostics_json = json.dumps(
|
||
{"final_confirmation": dict(final_confirmation or {})},
|
||
ensure_ascii=False,
|
||
)
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, task_type, status, execution_mode FROM pdd_tasks"
|
||
" WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
if task["execution_mode"] != "live":
|
||
raise ValueError("演练任务不能进入不可逆阶段")
|
||
if task["status"] != TaskStatus.RUNNING.value:
|
||
raise ValueError("采购任务当前不在执行中")
|
||
cursor = connection.execute(
|
||
"UPDATE task_runs SET irreversible_action_at = ?,"
|
||
" current_step = ?, diagnostics_json = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND attempt_id = ?"
|
||
" AND run_status = 'running'"
|
||
" AND irreversible_action_at IS NULL",
|
||
(
|
||
now,
|
||
step,
|
||
diagnostics_json,
|
||
now,
|
||
task["id"],
|
||
attempt_id,
|
||
),
|
||
)
|
||
if cursor.rowcount != 1:
|
||
raise ValueError("不可逆标记写入失败或已经存在")
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET current_step = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(step, now, task["id"]),
|
||
)
|
||
return now
|
||
finally:
|
||
connection.close()
|
||
|
||
def move_purchase_to_reconcile(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
*,
|
||
order_submitted_at: Optional[str] = None,
|
||
message: str = "订单提交结果待核对，绝不重新下单",
|
||
) -> None:
|
||
"""不可逆点击后转入只读核单状态，不创建可重试下单路径。"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, task_type, status FROM pdd_tasks"
|
||
" WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
cursor = connection.execute(
|
||
"UPDATE task_runs SET run_status = 'manual_review',"
|
||
" current_step = 'reconcile_purchase',"
|
||
" order_submitted_at = COALESCE(order_submitted_at, ?),"
|
||
" error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
|
||
" error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND attempt_id = ?"
|
||
" AND run_status = 'running'"
|
||
" AND irreversible_action_at IS NOT NULL",
|
||
(
|
||
order_submitted_at,
|
||
message,
|
||
now,
|
||
now,
|
||
task["id"],
|
||
attempt_id,
|
||
),
|
||
)
|
||
if cursor.rowcount != 1:
|
||
raise ValueError("不可逆采购执行记录不存在或已经结束")
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'manual_review',"
|
||
" current_step = 'reconcile_purchase',"
|
||
" last_error_code = 'PURCHASE_OUTCOME_UNKNOWN',"
|
||
" last_error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(message, now, now, task["id"]),
|
||
)
|
||
finally:
|
||
connection.close()
|
||
|
||
def save_purchase_reconciliation(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
match_status: str,
|
||
diagnostics: Dict[str, object],
|
||
) -> None:
|
||
"""保存一次只读订单核对结果，任务仍留给人工确认。"""
|
||
|
||
allowed = {"matched", "not_found", "ambiguous", "unknown"}
|
||
if match_status not in allowed:
|
||
raise ValueError("采购核对结果无效")
|
||
now = utc_now_iso()
|
||
step = (
|
||
"reconcile_completed"
|
||
if match_status == "matched"
|
||
else "reconcile_manual_review"
|
||
)
|
||
messages = {
|
||
"matched": "只读核对发现唯一候选订单，请人工确认",
|
||
"not_found": "只读核对未找到订单，不得重新下单",
|
||
"ambiguous": "只读核对发现多个候选订单，请人工确认",
|
||
"unknown": "订单字段不完整、不一致或读取失败，不得重新下单",
|
||
}
|
||
error_codes = {
|
||
"matched": "PURCHASE_RECONCILED",
|
||
"not_found": "ORDER_NOT_FOUND",
|
||
"ambiguous": "AMBIGUOUS_ORDER_MATCH",
|
||
"unknown": "ORDER_MATCH_UNCERTAIN",
|
||
}
|
||
message = messages[match_status]
|
||
error_code = error_codes[match_status]
|
||
saved_diagnostics = dict(diagnostics)
|
||
saved_diagnostics["match_status"] = match_status
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, task_type, status, current_step FROM pdd_tasks"
|
||
" WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
if (
|
||
task["status"] != TaskStatus.MANUAL_REVIEW.value
|
||
or task["current_step"] != "reconcile_purchase"
|
||
):
|
||
raise ValueError("采购任务当前不在待核对状态")
|
||
run = connection.execute(
|
||
"SELECT irreversible_action_at, run_status, current_step,"
|
||
" diagnostics_json"
|
||
" FROM task_runs"
|
||
" WHERE task_id = ? AND attempt_id = ?",
|
||
(task["id"], attempt_id),
|
||
).fetchone()
|
||
if run is None or run["irreversible_action_at"] is None:
|
||
raise ValueError("只有已进入不可逆阶段的运行才能核对")
|
||
if (
|
||
run["run_status"] != RunStatus.MANUAL_REVIEW.value
|
||
or run["current_step"] != "reconcile_purchase"
|
||
):
|
||
raise ValueError("采购执行记录已经核对或状态已变更")
|
||
run_diagnostics = (
|
||
self._load_json_object(run["diagnostics_json"])
|
||
if run["diagnostics_json"]
|
||
else {}
|
||
)
|
||
run_diagnostics["reconciliation"] = saved_diagnostics
|
||
run_cursor = connection.execute(
|
||
"UPDATE task_runs SET run_status = 'manual_review',"
|
||
" current_step = ?, error_code = ?,"
|
||
" error_message = ?, diagnostics_json = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND attempt_id = ?",
|
||
(
|
||
step,
|
||
error_code,
|
||
message,
|
||
json.dumps(run_diagnostics, ensure_ascii=False),
|
||
now,
|
||
task["id"],
|
||
attempt_id,
|
||
),
|
||
)
|
||
if run_cursor.rowcount != 1:
|
||
raise ValueError("采购执行记录核对保存失败")
|
||
task_cursor = connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'manual_review',"
|
||
" current_step = ?, last_error_code = ?,"
|
||
" last_error_message = ?, updated_at = ? WHERE id = ?"
|
||
" AND status = 'manual_review'"
|
||
" AND current_step = 'reconcile_purchase'",
|
||
(step, error_code, message, now, task["id"]),
|
||
)
|
||
if task_cursor.rowcount != 1:
|
||
raise ValueError("采购任务状态已变更，核对结果未保存")
|
||
finally:
|
||
connection.close()
|
||
|
||
def save_matched_purchase_reconciliation(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
pdd_data: Dict[str, object],
|
||
diagnostics: Dict[str, object],
|
||
) -> OutboxEventRecord:
|
||
"""唯一未付款订单核对成功后，原子保存结果并创建幂等 Outbox。"""
|
||
|
||
purchase = pdd_data.get("purchase")
|
||
if not isinstance(purchase, dict) or (
|
||
purchase.get("mode") != "live"
|
||
or purchase.get("match_status") != "matched"
|
||
or purchase.get("order_submitted") is not True
|
||
or purchase.get("payment_attempted") is not False
|
||
or purchase.get("payment_status") != "unpaid"
|
||
or not str(purchase.get("order_no") or "").strip()
|
||
):
|
||
raise ValueError("真实采购核对结果结构无效")
|
||
|
||
now = utc_now_iso()
|
||
idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1"
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
existing = connection.execute(
|
||
"SELECT id FROM outbox_events WHERE idempotency_key = ?",
|
||
(idempotency_key,),
|
||
).fetchone()
|
||
if existing is not None:
|
||
event_id = int(existing["id"])
|
||
else:
|
||
task = connection.execute(
|
||
"SELECT id, version, task_type, status, current_step"
|
||
" FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
if (
|
||
task["status"] != TaskStatus.MANUAL_REVIEW.value
|
||
or task["current_step"] != "reconcile_purchase"
|
||
):
|
||
raise ValueError("采购任务当前不在待核对状态")
|
||
run = connection.execute(
|
||
"SELECT run_status, current_step, diagnostics_json,"
|
||
" irreversible_action_at FROM task_runs"
|
||
" WHERE task_id = ? AND attempt_id = ?",
|
||
(task["id"], attempt_id),
|
||
).fetchone()
|
||
if (
|
||
run is None
|
||
or run["irreversible_action_at"] is None
|
||
or run["run_status"] != RunStatus.MANUAL_REVIEW.value
|
||
or run["current_step"] != "reconcile_purchase"
|
||
):
|
||
raise ValueError("不可逆采购执行记录不在待核对状态")
|
||
|
||
run_diagnostics = (
|
||
self._load_json_object(run["diagnostics_json"])
|
||
if run["diagnostics_json"]
|
||
else {}
|
||
)
|
||
run_diagnostics["reconciliation"] = dict(diagnostics)
|
||
result_json = json.dumps(pdd_data, ensure_ascii=False)
|
||
payload = {
|
||
"task_version": task["version"],
|
||
"attempt_id": attempt_id,
|
||
"result_type": "purchase",
|
||
"completed_at": now,
|
||
"pdd_data": pdd_data,
|
||
}
|
||
task_cursor = connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'result_pending',"
|
||
" current_step = 'purchase_order_matched_pending_report',"
|
||
" pdd_data = ?, shop_name = ?,"
|
||
" price_cent = ?, last_error_code = NULL,"
|
||
" last_error_message = NULL, finished_at = ?, updated_at = ?"
|
||
" WHERE id = ? AND status = 'manual_review'"
|
||
" AND current_step = 'reconcile_purchase'",
|
||
(
|
||
result_json,
|
||
self._summary_shop_name(pdd_data),
|
||
self._purchase_result_price(pdd_data),
|
||
now,
|
||
now,
|
||
task["id"],
|
||
),
|
||
)
|
||
if task_cursor.rowcount != 1:
|
||
raise ValueError("采购任务状态已变化，未保存核对结果")
|
||
run_cursor = connection.execute(
|
||
"UPDATE task_runs SET run_status = 'succeeded',"
|
||
" current_step = 'purchase_order_matched_pending_report',"
|
||
" result_data = ?,"
|
||
" diagnostics_json = ?, error_code = NULL,"
|
||
" error_message = NULL, finished_at = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND attempt_id = ?"
|
||
" AND run_status = 'manual_review'"
|
||
" AND current_step = 'reconcile_purchase'"
|
||
" AND irreversible_action_at IS NOT NULL",
|
||
(
|
||
result_json,
|
||
json.dumps(run_diagnostics, ensure_ascii=False),
|
||
now,
|
||
now,
|
||
task["id"],
|
||
attempt_id,
|
||
),
|
||
)
|
||
if run_cursor.rowcount != 1:
|
||
raise ValueError("采购执行记录状态已变化，未保存核对结果")
|
||
event_cursor = connection.execute(
|
||
"INSERT INTO outbox_events (task_id, event_type,"
|
||
" idempotency_key, payload_json, status, created_at,"
|
||
" updated_at)"
|
||
" VALUES (?, 'purchase_result', ?, ?, 'pending', ?, ?)",
|
||
(
|
||
task["id"],
|
||
idempotency_key,
|
||
json.dumps(payload, ensure_ascii=False),
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
event_id = int(event_cursor.lastrowid)
|
||
event = self.get_outbox_event(event_id)
|
||
assert event is not None
|
||
return event
|
||
finally:
|
||
connection.close()
|
||
|
||
def save_purchase_result(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
pdd_data: Dict[str, object],
|
||
) -> OutboxEventRecord:
|
||
"""原子保存采购演练结果，并创建采购结果 Outbox。"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, version, task_type FROM pdd_tasks"
|
||
" WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
payload = {
|
||
"task_version": task["version"],
|
||
"attempt_id": attempt_id,
|
||
"result_type": "purchase",
|
||
"completed_at": now,
|
||
"pdd_data": pdd_data,
|
||
}
|
||
result_json = json.dumps(pdd_data, ensure_ascii=False)
|
||
idempotency_key = (
|
||
f"{remote_task_id}:{attempt_id}:result-v1"
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'result_pending',"
|
||
" current_step = 'submit_result', pdd_data = ?, shop_name = ?,"
|
||
" price_cent = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(
|
||
result_json,
|
||
self._summary_shop_name(pdd_data),
|
||
self._purchase_result_price(pdd_data),
|
||
now,
|
||
now,
|
||
task["id"],
|
||
),
|
||
)
|
||
cursor = connection.execute(
|
||
"UPDATE task_runs SET run_status = 'succeeded',"
|
||
" current_step = 'submit_result', result_data = ?,"
|
||
" finished_at = ?, updated_at = ?"
|
||
" WHERE task_id = ? AND attempt_id = ?"
|
||
" AND run_status = 'running'",
|
||
(
|
||
result_json,
|
||
now,
|
||
now,
|
||
task["id"],
|
||
attempt_id,
|
||
),
|
||
)
|
||
if cursor.rowcount != 1:
|
||
raise ValueError("采购执行记录不存在或已经结束")
|
||
event_cursor = connection.execute(
|
||
"INSERT INTO outbox_events (task_id, event_type,"
|
||
" idempotency_key, payload_json, status, created_at, updated_at)"
|
||
" VALUES (?, 'purchase_result', ?, ?, 'pending', ?, ?)",
|
||
(
|
||
task["id"],
|
||
idempotency_key,
|
||
json.dumps(payload, ensure_ascii=False),
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
event_id = int(event_cursor.lastrowid)
|
||
event = self.get_outbox_event(event_id)
|
||
assert event is not None
|
||
return event
|
||
finally:
|
||
connection.close()
|
||
|
||
def save_purchase_failure(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
status: TaskStatus,
|
||
error_code: str,
|
||
error_message: str,
|
||
retryable: bool,
|
||
step: str,
|
||
diagnostics: Optional[Dict[str, object]] = None,
|
||
) -> OutboxEventRecord:
|
||
"""原子保存采购演练失败，并创建失败 Outbox。"""
|
||
|
||
if status not in {
|
||
TaskStatus.RETRY_WAIT,
|
||
TaskStatus.MANUAL_REVIEW,
|
||
TaskStatus.FAILED,
|
||
TaskStatus.CANCELLED,
|
||
}:
|
||
raise ValueError("失败状态无效")
|
||
checked_step = str(step or "purchase_prepare").strip()
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, version, task_type FROM pdd_tasks"
|
||
" WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
if task["task_type"] != TaskType.PURCHASE.value:
|
||
raise ValueError("当前任务不是采购任务")
|
||
payload = {
|
||
"task_version": task["version"],
|
||
"attempt_id": attempt_id,
|
||
"status": status.value,
|
||
"error": {
|
||
"code": error_code,
|
||
"message": error_message,
|
||
"retryable": retryable,
|
||
"step": checked_step,
|
||
},
|
||
"diagnostics": diagnostics or {"artifacts": []},
|
||
"reported_at": now,
|
||
}
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = ?, current_step = ?,"
|
||
" retry_count = retry_count + ?, last_error_code = ?,"
|
||
" last_error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(
|
||
status.value,
|
||
"failed",
|
||
1 if status is TaskStatus.RETRY_WAIT else 0,
|
||
error_code,
|
||
error_message,
|
||
now,
|
||
now,
|
||
task["id"],
|
||
),
|
||
)
|
||
run_status = {
|
||
TaskStatus.CANCELLED: RunStatus.CANCELLED,
|
||
TaskStatus.MANUAL_REVIEW: RunStatus.MANUAL_REVIEW,
|
||
}.get(status, RunStatus.FAILED)
|
||
cursor = connection.execute(
|
||
"UPDATE task_runs SET run_status = ?, current_step = ?,"
|
||
" error_code = ?, error_message = ?, diagnostics_json = ?,"
|
||
" finished_at = ?,"
|
||
" updated_at = ? WHERE task_id = ? AND attempt_id = ?"
|
||
" AND run_status = 'running'",
|
||
(
|
||
run_status.value,
|
||
checked_step,
|
||
error_code,
|
||
error_message,
|
||
json.dumps(diagnostics or {}, ensure_ascii=False),
|
||
now,
|
||
now,
|
||
task["id"],
|
||
attempt_id,
|
||
),
|
||
)
|
||
if cursor.rowcount != 1:
|
||
raise ValueError("采购执行记录不存在或已经结束")
|
||
idempotency_key = (
|
||
f"{remote_task_id}:{attempt_id}:failure-v1"
|
||
)
|
||
event_cursor = connection.execute(
|
||
"INSERT INTO outbox_events (task_id, event_type,"
|
||
" idempotency_key, payload_json, status, created_at, updated_at)"
|
||
" VALUES (?, 'task_failure', ?, ?, 'pending', ?, ?)",
|
||
(
|
||
task["id"],
|
||
idempotency_key,
|
||
json.dumps(payload, ensure_ascii=False),
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
event_id = int(event_cursor.lastrowid)
|
||
event = self.get_outbox_event(event_id)
|
||
assert event is not None
|
||
return event
|
||
finally:
|
||
connection.close()
|
||
|
||
def save_collect_result(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
pdd_data: Dict[str, object],
|
||
) -> OutboxEventRecord:
|
||
"""在一个事务中保存采集结果并创建待提交事件。"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
payload = {
|
||
"task_version": task["version"],
|
||
"attempt_id": attempt_id,
|
||
"result_type": "collect",
|
||
"completed_at": now,
|
||
"pdd_data": pdd_data,
|
||
}
|
||
idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1"
|
||
result_json = json.dumps(pdd_data, ensure_ascii=False)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'result_pending',"
|
||
" current_step = 'submit_result', pdd_data = ?, goods_id = ?,"
|
||
" title = ?, shop_name = ?, price_cent = ?,"
|
||
" finished_at = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(
|
||
result_json,
|
||
pdd_data.get("goods_id"), pdd_data.get("title"),
|
||
self._summary_shop_name(pdd_data),
|
||
self._summary_price(pdd_data), now, now, task["id"],
|
||
),
|
||
)
|
||
connection.execute(
|
||
"UPDATE task_runs SET run_status = 'succeeded',"
|
||
" current_step = 'submit_result', result_data = ?,"
|
||
" finished_at = ?, updated_at = ?"
|
||
" WHERE attempt_id = ?",
|
||
(result_json, now, now, attempt_id),
|
||
)
|
||
cursor = connection.execute(
|
||
"INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
|
||
" payload_json, status, created_at, updated_at)"
|
||
" VALUES (?, 'collect_result', ?, ?, 'pending', ?, ?)",
|
||
(
|
||
task["id"], idempotency_key,
|
||
json.dumps(payload, ensure_ascii=False), now, now,
|
||
),
|
||
)
|
||
event_id = int(cursor.lastrowid)
|
||
event = self.get_outbox_event(event_id)
|
||
assert event is not None
|
||
return event
|
||
finally:
|
||
connection.close()
|
||
|
||
def save_collect_failure(
|
||
self,
|
||
remote_task_id: str,
|
||
attempt_id: str,
|
||
status: TaskStatus,
|
||
error_code: str,
|
||
error_message: str,
|
||
retryable: bool,
|
||
diagnostics: Optional[Dict[str, object]] = None,
|
||
) -> OutboxEventRecord:
|
||
"""保存结构化失败，并可靠排队提交 Admin。"""
|
||
|
||
if status not in {
|
||
TaskStatus.RETRY_WAIT, TaskStatus.MANUAL_REVIEW,
|
||
TaskStatus.FAILED, TaskStatus.CANCELLED,
|
||
}:
|
||
raise ValueError("失败状态无效")
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
task = connection.execute(
|
||
"SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if task is None:
|
||
raise ValueError(f"任务 {remote_task_id} 不存在")
|
||
payload = {
|
||
"task_version": task["version"],
|
||
"attempt_id": attempt_id,
|
||
"status": status.value,
|
||
"error": {
|
||
"code": error_code,
|
||
"message": error_message,
|
||
"retryable": retryable,
|
||
"step": "collecting",
|
||
},
|
||
"diagnostics": diagnostics or {"artifacts": []},
|
||
"reported_at": now,
|
||
}
|
||
idempotency_key = f"{remote_task_id}:{attempt_id}:failure-v1"
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = ?, current_step = 'failed',"
|
||
" retry_count = retry_count + ?, last_error_code = ?,"
|
||
" last_error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE id = ?",
|
||
(
|
||
status.value, 1 if status is TaskStatus.RETRY_WAIT else 0,
|
||
error_code, error_message, now, now, task["id"],
|
||
),
|
||
)
|
||
run_status = {
|
||
TaskStatus.CANCELLED: RunStatus.CANCELLED,
|
||
TaskStatus.MANUAL_REVIEW: RunStatus.MANUAL_REVIEW,
|
||
}.get(status, RunStatus.FAILED)
|
||
connection.execute(
|
||
"UPDATE task_runs SET run_status = ?, current_step = 'failed',"
|
||
" error_code = ?, error_message = ?, finished_at = ?, updated_at = ?"
|
||
" WHERE attempt_id = ?",
|
||
(run_status.value, error_code, error_message, now, now, attempt_id),
|
||
)
|
||
cursor = connection.execute(
|
||
"INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
|
||
" payload_json, status, created_at, updated_at)"
|
||
" VALUES (?, 'task_failure', ?, ?, 'pending', ?, ?)",
|
||
(
|
||
task["id"], idempotency_key,
|
||
json.dumps(payload, ensure_ascii=False), now, now,
|
||
),
|
||
)
|
||
event_id = int(cursor.lastrowid)
|
||
event = self.get_outbox_event(event_id)
|
||
assert event is not None
|
||
return event
|
||
finally:
|
||
connection.close()
|
||
|
||
def next_pending_outbox(self) -> Optional[OutboxEventRecord]:
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT o.* FROM outbox_events o JOIN pdd_tasks t ON t.id = o.task_id"
|
||
" WHERE o.status = 'pending'"
|
||
" AND (o.next_retry_at IS NULL OR o.next_retry_at <= ?)"
|
||
" ORDER BY o.id ASC LIMIT 1",
|
||
(utc_now_iso(),),
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_outbox(row) if row is not None else None
|
||
|
||
def latest_result_outbox(
|
||
self, remote_task_id: str
|
||
) -> Optional[OutboxEventRecord]:
|
||
"""读取任务最新的成功结果事件，不返回失败上报事件。"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT o.* FROM outbox_events o"
|
||
" JOIN pdd_tasks t ON t.id = o.task_id"
|
||
" WHERE t.remote_task_id = ?"
|
||
" AND o.event_type IN ('collect_result', 'purchase_result')"
|
||
" ORDER BY o.id DESC LIMIT 1",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_outbox(row) if row is not None else None
|
||
|
||
def outbox_for_resubmit(
|
||
self, remote_task_id: str
|
||
) -> Optional[OutboxEventRecord]:
|
||
"""返回手工重新上报应发送的事件。
|
||
|
||
先返回最早一条尚未发送的事件，保证失败信息不会被旧结果跳过；
|
||
全部事件都已发送时，才返回最新的成功结果用于再次确认。
|
||
"""
|
||
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT o.* FROM outbox_events o"
|
||
" JOIN pdd_tasks t ON t.id = o.task_id"
|
||
" WHERE t.remote_task_id = ? AND o.status != 'sent'"
|
||
" ORDER BY o.id ASC LIMIT 1",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
row = connection.execute(
|
||
"SELECT o.* FROM outbox_events o"
|
||
" JOIN pdd_tasks t ON t.id = o.task_id"
|
||
" WHERE t.remote_task_id = ?"
|
||
" AND o.event_type IN ('collect_result', 'purchase_result')"
|
||
" ORDER BY o.id DESC LIMIT 1",
|
||
(remote_task_id,),
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_outbox(row) if row is not None else None
|
||
|
||
def get_outbox_event(self, event_id: int) -> Optional[OutboxEventRecord]:
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT * FROM outbox_events WHERE id = ?", (event_id,)
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
return self._to_outbox(row) if row is not None else None
|
||
|
||
def outbox_task_id(self, event_id: int) -> str:
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
row = connection.execute(
|
||
"SELECT t.remote_task_id FROM outbox_events o"
|
||
" JOIN pdd_tasks t ON t.id = o.task_id WHERE o.id = ?",
|
||
(event_id,),
|
||
).fetchone()
|
||
finally:
|
||
connection.close()
|
||
if row is None:
|
||
raise ValueError(f"Outbox {event_id} 不存在")
|
||
return str(row[0])
|
||
|
||
def mark_outbox_sending(self, event_id: int) -> None:
|
||
self._update_outbox(event_id, "sending", None)
|
||
|
||
def mark_outbox_retry(self, event_id: int, message: str) -> None:
|
||
self._update_outbox(event_id, "pending", message, increment=True)
|
||
|
||
def mark_outbox_failed(self, event_id: int, message: str) -> None:
|
||
self._update_outbox(event_id, "failed", message, increment=True)
|
||
|
||
def mark_outbox_sent(self, event_id: int) -> None:
|
||
"""标记事件已发送，并只按当前最新执行更新任务状态。
|
||
|
||
已经发送过的历史结果再次上报时，不得覆盖任务的最新状态。
|
||
最新失败信息发送成功时会恢复对应失败状态，用于修复旧版本
|
||
产生的“表格显示已完成、实际最新执行失败”记录。
|
||
"""
|
||
|
||
now = utc_now_iso()
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
row = connection.execute(
|
||
"SELECT task_id, event_type, payload_json, sent_at"
|
||
" FROM outbox_events WHERE id = ?",
|
||
(event_id,),
|
||
).fetchone()
|
||
if row is None:
|
||
raise ValueError(f"Outbox {event_id} 不存在")
|
||
was_already_sent = row["sent_at"] is not None
|
||
connection.execute(
|
||
"UPDATE outbox_events SET status = 'sent', sent_at = ?,"
|
||
" updated_at = ?, attempt_count = attempt_count + 1"
|
||
" WHERE id = ?",
|
||
(now, now, event_id),
|
||
)
|
||
payload = self._load_outbox_payload(row["payload_json"])
|
||
if row["event_type"] == OutboxEventType.TASK_FAILURE.value:
|
||
self._restore_current_failure_state(
|
||
connection, row["task_id"], payload, now
|
||
)
|
||
elif not was_already_sent and row["event_type"] in {
|
||
OutboxEventType.COLLECT_RESULT.value,
|
||
OutboxEventType.PURCHASE_RESULT.value,
|
||
}:
|
||
self._complete_current_result(
|
||
connection, row["task_id"], payload, now
|
||
)
|
||
finally:
|
||
connection.close()
|
||
|
||
@staticmethod
|
||
def _load_outbox_payload(payload_json: str) -> Dict[str, object]:
|
||
"""读取本地 Outbox JSON；损坏时返回空对象，不猜测任务状态。"""
|
||
|
||
try:
|
||
payload = json.loads(payload_json)
|
||
except (TypeError, ValueError):
|
||
return {}
|
||
return payload if isinstance(payload, dict) else {}
|
||
|
||
@staticmethod
|
||
def _latest_run_for_task(
|
||
connection: sqlite3.Connection, task_id: int
|
||
) -> Optional[sqlite3.Row]:
|
||
return connection.execute(
|
||
"SELECT attempt_id, finished_at FROM task_runs"
|
||
" WHERE task_id = ? ORDER BY attempt_no DESC LIMIT 1",
|
||
(task_id,),
|
||
).fetchone()
|
||
|
||
@classmethod
|
||
def _complete_current_result(
|
||
cls,
|
||
connection: sqlite3.Connection,
|
||
task_id: int,
|
||
payload: Dict[str, object],
|
||
now: str,
|
||
) -> None:
|
||
"""只有最新执行的待提交结果可以把任务改成已完成。"""
|
||
|
||
latest_run = cls._latest_run_for_task(connection, task_id)
|
||
if (
|
||
latest_run is None
|
||
or payload.get("attempt_id") != latest_run["attempt_id"]
|
||
):
|
||
return
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = 'succeeded',"
|
||
" current_step = 'completed', updated_at = ?"
|
||
" WHERE id = ? AND status = 'result_pending'",
|
||
(now, task_id),
|
||
)
|
||
|
||
@classmethod
|
||
def _restore_current_failure_state(
|
||
cls,
|
||
connection: sqlite3.Connection,
|
||
task_id: int,
|
||
payload: Dict[str, object],
|
||
now: str,
|
||
) -> None:
|
||
"""最新失败事件上报后恢复任务失败状态，旧事件不覆盖新状态。"""
|
||
|
||
latest_run = cls._latest_run_for_task(connection, task_id)
|
||
if (
|
||
latest_run is None
|
||
or payload.get("attempt_id") != latest_run["attempt_id"]
|
||
):
|
||
return
|
||
status = payload.get("status")
|
||
allowed_statuses = {
|
||
TaskStatus.RETRY_WAIT.value,
|
||
TaskStatus.MANUAL_REVIEW.value,
|
||
TaskStatus.FAILED.value,
|
||
TaskStatus.CANCELLED.value,
|
||
}
|
||
if status not in allowed_statuses:
|
||
return
|
||
error = payload.get("error")
|
||
error = error if isinstance(error, dict) else {}
|
||
finished_at = (
|
||
latest_run["finished_at"] or payload.get("reported_at") or now
|
||
)
|
||
connection.execute(
|
||
"UPDATE pdd_tasks SET status = ?, current_step = 'failed',"
|
||
" last_error_code = ?, last_error_message = ?,"
|
||
" finished_at = ?, updated_at = ? WHERE id = ?",
|
||
(
|
||
status,
|
||
error.get("code"),
|
||
error.get("message"),
|
||
finished_at,
|
||
now,
|
||
task_id,
|
||
),
|
||
)
|
||
|
||
def _update_outbox(
|
||
self, event_id: int, status: str, message: Optional[str], increment: bool = False
|
||
) -> None:
|
||
connection = open_database(self._db_path)
|
||
try:
|
||
with connection:
|
||
cursor = connection.execute(
|
||
"UPDATE outbox_events SET status = ?, last_error = ?, updated_at = ?,"
|
||
f" attempt_count = attempt_count + {1 if increment else 0} WHERE id = ?",
|
||
(status, message, utc_now_iso(), event_id),
|
||
)
|
||
if cursor.rowcount != 1:
|
||
raise ValueError(f"Outbox {event_id} 不存在")
|
||
finally:
|
||
connection.close()
|
||
|
||
@staticmethod
|
||
def _summary_price(pdd_data: Dict[str, object]) -> Optional[int]:
|
||
skus = pdd_data.get("skus")
|
||
if not isinstance(skus, list):
|
||
return None
|
||
prices = [
|
||
item.get("price_cent") for item in skus
|
||
if isinstance(item, dict) and isinstance(item.get("price_cent"), int)
|
||
]
|
||
return min(prices) if prices else None
|
||
|
||
@staticmethod
|
||
def _purchase_result_price(
|
||
pdd_data: Dict[str, object]
|
||
) -> Optional[int]:
|
||
purchase = pdd_data.get("purchase")
|
||
if not isinstance(purchase, dict):
|
||
return None
|
||
confirmed = purchase.get("confirmed")
|
||
if not isinstance(confirmed, dict):
|
||
return None
|
||
price = confirmed.get("unit_price_cent")
|
||
if isinstance(price, bool) or not isinstance(price, int):
|
||
return None
|
||
return price
|
||
|
||
@staticmethod
|
||
def _to_outbox(row: sqlite3.Row) -> OutboxEventRecord:
|
||
return OutboxEventRecord(
|
||
id=row["id"], task_id=row["task_id"],
|
||
event_type=OutboxEventType(row["event_type"]),
|
||
idempotency_key=row["idempotency_key"],
|
||
payload_json=TaskRepository._load_json_object(row["payload_json"]),
|
||
status=OutboxStatus(row["status"]), attempt_count=row["attempt_count"],
|
||
next_retry_at=row["next_retry_at"], last_error=row["last_error"],
|
||
created_at=row["created_at"], updated_at=row["updated_at"],
|
||
sent_at=row["sent_at"],
|
||
)
|
||
|
||
@staticmethod
|
||
def _to_task_run(row: sqlite3.Row) -> TaskRunRecord:
|
||
diagnostics = (
|
||
TaskRepository._load_json_object(row["diagnostics_json"])
|
||
if row["diagnostics_json"] is not None
|
||
else None
|
||
)
|
||
result_data = (
|
||
TaskRepository._load_json_object(row["result_data"])
|
||
if row["result_data"] is not None
|
||
else None
|
||
)
|
||
return TaskRunRecord(
|
||
id=row["id"],
|
||
task_id=row["task_id"],
|
||
attempt_id=row["attempt_id"],
|
||
attempt_no=row["attempt_no"],
|
||
device_address=row["device_address"],
|
||
run_status=RunStatus(row["run_status"]),
|
||
current_step=row["current_step"],
|
||
started_at=row["started_at"],
|
||
finished_at=row["finished_at"],
|
||
irreversible_action_at=row["irreversible_action_at"],
|
||
order_submitted_at=row["order_submitted_at"],
|
||
error_code=row["error_code"],
|
||
error_message=row["error_message"],
|
||
diagnostics_json=diagnostics,
|
||
result_data=result_data,
|
||
artifact_directory=row["artifact_directory"],
|
||
created_at=row["created_at"],
|
||
updated_at=row["updated_at"],
|
||
)
|
||
|
||
@staticmethod
|
||
def _validate_page(limit: int, offset: int) -> None:
|
||
if not 1 <= limit <= MAX_PAGE_SIZE:
|
||
raise ValueError(f"limit 必须在 1 到 {MAX_PAGE_SIZE} 之间")
|
||
if offset < 0:
|
||
raise ValueError("offset 不能小于 0")
|
||
|
||
@staticmethod
|
||
def _build_where(filters: TaskFilters) -> Tuple[str, List[object]]:
|
||
clauses = ["removed_at IS NULL"]
|
||
parameters: List[object] = []
|
||
if filters.task_type is not None:
|
||
clauses.append("task_type = ?")
|
||
parameters.append(filters.task_type.value)
|
||
if filters.status is not None:
|
||
clauses.append("status = ?")
|
||
parameters.append(filters.status.value)
|
||
if filters.keyword.strip():
|
||
keyword = TaskRepository._escape_like(filters.keyword.strip())
|
||
pattern = f"%{keyword}%"
|
||
clauses.append(
|
||
"(remote_task_id LIKE ? ESCAPE '\\'"
|
||
" OR COALESCE(goods_id, '') LIKE ? ESCAPE '\\'"
|
||
" OR COALESCE(title, '') LIKE ? ESCAPE '\\'"
|
||
" OR COALESCE(CASE WHEN pdd_data IS NOT NULL"
|
||
" AND json_valid(pdd_data)"
|
||
" THEN json_extract(pdd_data, '$.purchase.order_no')"
|
||
" END, '') LIKE ? ESCAPE '\\')"
|
||
)
|
||
parameters.extend((pattern, pattern, pattern, pattern))
|
||
return (" WHERE " + " AND ".join(clauses) if clauses else "", parameters)
|
||
|
||
@staticmethod
|
||
def _escape_like(value: str) -> str:
|
||
return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||
|
||
@staticmethod
|
||
def _to_summary(row: sqlite3.Row) -> TaskSummary:
|
||
return TaskSummary(
|
||
id=row["id"],
|
||
remote_task_id=row["remote_task_id"],
|
||
task_type=TaskType(row["task_type"]),
|
||
goods_id=row["goods_id"],
|
||
title=row["title"],
|
||
shop_name=row["shop_name"],
|
||
target_color=row["target_color"],
|
||
target_size=row["target_size"],
|
||
price_cent=row["price_cent"],
|
||
quantity=row["quantity"],
|
||
status=TaskStatus(row["status"]),
|
||
latest_run_status=(
|
||
RunStatus(row["latest_run_status"])
|
||
if row["latest_run_status"] is not None
|
||
else None
|
||
),
|
||
duration_seconds=row["duration_seconds"],
|
||
order_no=row["order_no"],
|
||
updated_at=row["updated_at"],
|
||
)
|
||
|
||
@staticmethod
|
||
def _summary_shop_name(pdd_data: Dict[str, object]) -> Optional[str]:
|
||
"""提取列表需要的店铺名，空值不写入摘要列。"""
|
||
|
||
value = pdd_data.get("shop_name")
|
||
if not isinstance(value, str):
|
||
return None
|
||
value = value.strip()
|
||
return value or None
|
||
|
||
@staticmethod
|
||
def _to_detail(row: sqlite3.Row) -> TaskDetail:
|
||
admin_payload = TaskRepository._load_json_object(row["admin_payload"])
|
||
pdd_data = (
|
||
TaskRepository._load_json_object(row["pdd_data"])
|
||
if row["pdd_data"] is not None
|
||
else None
|
||
)
|
||
return TaskDetail(
|
||
id=row["id"],
|
||
remote_task_id=row["remote_task_id"],
|
||
task_type=TaskType(row["task_type"]),
|
||
execution_mode=row["execution_mode"],
|
||
goods_id=row["goods_id"],
|
||
goods_url=row["goods_url"],
|
||
title=row["title"],
|
||
target_color=row["target_color"],
|
||
target_size=row["target_size"],
|
||
price_cent=row["price_cent"],
|
||
quantity=row["quantity"],
|
||
status=TaskStatus(row["status"]),
|
||
current_step=row["current_step"],
|
||
priority=row["priority"],
|
||
version=row["version"],
|
||
admin_payload=admin_payload,
|
||
pdd_data=pdd_data,
|
||
retry_count=row["retry_count"],
|
||
last_error_code=row["last_error_code"],
|
||
last_error_message=row["last_error_message"],
|
||
received_at=row["received_at"],
|
||
started_at=row["started_at"],
|
||
finished_at=row["finished_at"],
|
||
created_at=row["created_at"],
|
||
updated_at=row["updated_at"],
|
||
)
|
||
|
||
@staticmethod
|
||
def _load_json_object(text: str) -> Dict[str, object]:
|
||
value = json.loads(text)
|
||
if not isinstance(value, dict):
|
||
raise ValueError("数据库 JSON 字段必须是对象")
|
||
return value
|