feat: 实现采购任务安全演练 (#71)
This commit is contained in:
chengma
@@ -210,6 +210,20 @@ class TaskRepository:
|
||||
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 validate_collect_rerun(self, remote_task_id: str) -> TaskDetail:
|
||||
"""校验任务能否重新采集，成功时返回任务详情。"""
|
||||
|
||||
@@ -341,6 +355,291 @@ class TaskRepository:
|
||||
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 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 = ?,"
|
||||
" price_cent = ?, finished_at = ?, updated_at = ?"
|
||||
" WHERE id = ?",
|
||||
(
|
||||
result_json,
|
||||
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 = ?, finished_at = ?,"
|
||||
" updated_at = ? WHERE task_id = ? AND attempt_id = ?"
|
||||
" AND run_status = 'running'",
|
||||
(
|
||||
run_status.value,
|
||||
checked_step,
|
||||
error_code,
|
||||
error_message,
|
||||
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,
|
||||
@@ -543,7 +842,10 @@ class TaskRepository:
|
||||
" WHERE id = ?",
|
||||
(now, now, event_id),
|
||||
)
|
||||
if row["event_type"] == OutboxEventType.COLLECT_RESULT.value:
|
||||
if row["event_type"] in {
|
||||
OutboxEventType.COLLECT_RESULT.value,
|
||||
OutboxEventType.PURCHASE_RESULT.value,
|
||||
}:
|
||||
connection.execute(
|
||||
"UPDATE pdd_tasks SET status = 'succeeded',"
|
||||
" current_step = 'completed', updated_at = ? WHERE id = ?",
|
||||
@@ -579,6 +881,21 @@ class TaskRepository:
|
||||
]
|
||||
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(
|
||||
|
||||
Reference in New Issue
Block a user