feat: 完善真实采购订单核对与恢复 (#100)
This commit is contained in:
chengma
+172
-10
@@ -602,12 +602,19 @@ class TaskRepository:
|
||||
connection.close()
|
||||
|
||||
def mark_purchase_irreversible(
|
||||
self, remote_task_id: str, attempt_id: str
|
||||
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:
|
||||
@@ -626,11 +633,18 @@ class TaskRepository:
|
||||
raise ValueError("采购任务当前不在执行中")
|
||||
cursor = connection.execute(
|
||||
"UPDATE task_runs SET irreversible_action_at = ?,"
|
||||
" current_step = ?, updated_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, now, task["id"], attempt_id),
|
||||
(
|
||||
now,
|
||||
step,
|
||||
diagnostics_json,
|
||||
now,
|
||||
task["id"],
|
||||
attempt_id,
|
||||
),
|
||||
)
|
||||
if cursor.rowcount != 1:
|
||||
raise ValueError("不可逆标记写入失败或已经存在")
|
||||
@@ -719,9 +733,16 @@ class TaskRepository:
|
||||
"matched": "只读核对发现唯一候选订单，请人工确认",
|
||||
"not_found": "只读核对未找到订单，不得重新下单",
|
||||
"ambiguous": "只读核对发现多个候选订单，请人工确认",
|
||||
"unknown": "无法确定采购结果，不得重新下单",
|
||||
"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)
|
||||
@@ -742,7 +763,8 @@ class TaskRepository:
|
||||
):
|
||||
raise ValueError("采购任务当前不在待核对状态")
|
||||
run = connection.execute(
|
||||
"SELECT irreversible_action_at, run_status, current_step"
|
||||
"SELECT irreversible_action_at, run_status, current_step,"
|
||||
" diagnostics_json"
|
||||
" FROM task_runs"
|
||||
" WHERE task_id = ? AND attempt_id = ?",
|
||||
(task["id"], attempt_id),
|
||||
@@ -754,15 +776,22 @@ class TaskRepository:
|
||||
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 = 'PURCHASE_RECONCILED',"
|
||||
" current_step = ?, error_code = ?,"
|
||||
" error_message = ?, diagnostics_json = ?, updated_at = ?"
|
||||
" WHERE task_id = ? AND attempt_id = ?",
|
||||
(
|
||||
step,
|
||||
error_code,
|
||||
message,
|
||||
json.dumps(saved_diagnostics, ensure_ascii=False),
|
||||
json.dumps(run_diagnostics, ensure_ascii=False),
|
||||
now,
|
||||
task["id"],
|
||||
attempt_id,
|
||||
@@ -772,17 +801,150 @@ class TaskRepository:
|
||||
raise ValueError("采购执行记录核对保存失败")
|
||||
task_cursor = connection.execute(
|
||||
"UPDATE pdd_tasks SET status = 'manual_review',"
|
||||
" current_step = ?, last_error_code = 'PURCHASE_RECONCILED',"
|
||||
" current_step = ?, last_error_code = ?,"
|
||||
" last_error_message = ?, updated_at = ? WHERE id = ?"
|
||||
" AND status = 'manual_review'"
|
||||
" AND current_step = 'reconcile_purchase'",
|
||||
(step, message, now, task["id"]),
|
||||
(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 = ?,"
|
||||
" 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._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,
|
||||
|
||||
Reference in New Issue
Block a user