""" 状态中心 · FastAPI 入口(第1步骨架)。 提供: - 建单 POST /api/jobs - 节点开始 POST /api/jobs/{job_id}/nodes/{node_code}/start - 节点成功 POST /api/jobs/{job_id}/nodes/{node_code}/complete - 节点失败 POST /api/jobs/{job_id}/nodes/{node_code}/fail - 合同审批查一次 POST /api/jobs/{job_id}/nodes/CONTRACT_APPROVE_WAIT/check-result - 到点待查列表(调度器扫) GET /api/scheduler/due-contract-checks - 对码运行(内部算) POST /api/jobs/{job_id}/match/run - 对码人工确认 POST /api/match/{match_id}/confirm - 装载货品主数据 POST /api/match/reload-goods - 整单查询(画树状图用) GET /api/jobs/{job_id} 鉴权:RPA 回调带 header X-Agent-Token。 节点编排(13 个 = 11 主节点 + 2 条件节点): - 主链 11 步:ORDER_FETCH → PRODUCT_MATCH → CONTRACT_CREATE → CONTRACT_APPROVE_WAIT → REBATE_HEADER → REBATE_IMPORT → REBATE_SUBMIT → M5_CREATE_ORDER → M5_MANUAL_CONFIRM → M5_RECEIPT_NOTICE → ARCHIVE_JOB - 2 条件节点:OCR_EXTRACT(仅线下单)、SOURCE_ADD(仅对码判定缺货源)。 条件不满足则自动跳过,不占流程。 转移不是"数组下一个",是显式转移表 + 条件节点判定(见 _next_node)。 每个节点标了执行方(STATE_CENTER 编排 / RPA 执行 / HUMAN 人工挂起),见 NODE_EXECUTOR。 回调做了幂等(重复 complete 不二次推进)和乱序拒绝(非当前节点回调返回 409)。 """ from __future__ import annotations import json import uuid from datetime import datetime, timedelta from typing import Optional from fastapi import FastAPI, Header, HTTPException from pydantic import BaseModel from .config import settings from .db import get_conn from .match_engine import GoodsRecord, MatchEngine, build_match_key, load_goods_from_oracle from .web_state_machine import build_state_machine_meta, build_timeline app = FastAPI(title="采购主链RPA状态中心", version="0.1") # 全部 13 个节点(11 主 + 2 条件),仅用于展示口径 / 前端画图 / 校验节点名合法。 # 顺序不代表实际转移,实际转移看 TRANSITIONS + _next_node。 NODE_ORDER = [ "ORDER_FETCH", "OCR_EXTRACT", "PRODUCT_MATCH", "SOURCE_ADD", "CONTRACT_CREATE", "CONTRACT_APPROVE_WAIT", "REBATE_HEADER", "REBATE_IMPORT", "REBATE_SUBMIT", "M5_CREATE_ORDER", "M5_MANUAL_CONFIRM", "M5_RECEIPT_NOTICE", "ARCHIVE_JOB", ] # 显式转移表:node -> 下一个"直连"节点(不含条件跳过逻辑,跳过在 _should_skip 处理)。 # None = 整单终点。条件节点(OCR_EXTRACT/SOURCE_ADD)也在链里,但可能被跳过。 TRANSITIONS = { "ORDER_FETCH": "OCR_EXTRACT", "OCR_EXTRACT": "PRODUCT_MATCH", "PRODUCT_MATCH": "SOURCE_ADD", "SOURCE_ADD": "CONTRACT_CREATE", "CONTRACT_CREATE": "CONTRACT_APPROVE_WAIT", "CONTRACT_APPROVE_WAIT": "REBATE_HEADER", "REBATE_HEADER": "REBATE_IMPORT", "REBATE_IMPORT": "REBATE_SUBMIT", "REBATE_SUBMIT": "M5_CREATE_ORDER", "M5_CREATE_ORDER": "M5_MANUAL_CONFIRM", "M5_MANUAL_CONFIRM": "M5_RECEIPT_NOTICE", "M5_RECEIPT_NOTICE": "ARCHIVE_JOB", "ARCHIVE_JOB": None, } # 执行方标注:谁真正干这个节点。 # STATE_CENTER = 状态中心内部算(对码),不派 RPA、不挂人工。 # RPA = 艺赛旗点页面。 # HUMAN = 一期人工做(状态中心只挂待办等回填)。 NODE_EXECUTOR = { "ORDER_FETCH": "RPA", "OCR_EXTRACT": "RPA", # 线下单触发 OCR;线上单跳过 "PRODUCT_MATCH": "STATE_CENTER", # 对码是状态中心内部算 "SOURCE_ADD": "HUMAN", # 缺货源时人工在 M5 新增货源 "CONTRACT_CREATE": "RPA", "CONTRACT_APPROVE_WAIT": "RPA", # 到点由调度器触发 RPA 查一次审批 "REBATE_HEADER": "RPA", "REBATE_IMPORT": "RPA", "REBATE_SUBMIT": "RPA", "M5_CREATE_ORDER": "RPA", "M5_MANUAL_CONFIRM": "HUMAN", # 一期人工检查/保存/签订 "M5_RECEIPT_NOTICE": "HUMAN", # 一期人工生成收货通知单 "ARCHIVE_JOB": "STATE_CENTER", } HUMAN_NODE_ROLES = { "SOURCE_ADD": "PURCHASER", "M5_MANUAL_CONFIRM": "PURCHASER", "M5_RECEIPT_NOTICE": "PURCHASER", } HUMAN_TASK_TITLES = { "M5_MANUAL_CONFIRM": "M5人工确认", "M5_RECEIPT_NOTICE": "生成收货通知单", "CONTRACT_APPROVE_WAIT": "合同审批超时", } HUMAN_TASK_REASONS = { "M5_MANUAL_CONFIRM": "M5采购单已录入待保存，需要人工核对、保存并签订。", "M5_RECEIPT_NOTICE": "M5采购单已签订，需要生成收货通知单并回填单号。", "CONTRACT_APPROVE_WAIT": "合同审批已超过等待上限，需要运维确认已通过或终止本单。", } # 条件节点:满足条件才执行,否则自动跳过下一节点。 CONDITIONAL_NODES = {"OCR_EXTRACT", "SOURCE_ADD"} # 业务编号回写:哪个节点产出哪个 job 字段 BUSINESS_REF_FIELD = { "contractSystemNo": "contract_system_no", "rebateOrderNo": "rebate_order_no", "m5OrderNo": "m5_order_no", "receiptNoticeNo": "receipt_notice_no", } def _now(): return datetime.now() def _check_agent(token: Optional[str]): if token != settings.AGENT_TOKEN: raise HTTPException(status_code=401, detail="bad agent token") def _should_skip(node_code: str, job: dict) -> bool: """条件节点是否跳过。 - OCR_EXTRACT:只有线下单(OFFLINE)才做 OCR,线上单跳过。 - SOURCE_ADD:只有对码判定缺货源(needs_source_add=1)才做,否则跳过。 """ if node_code == "OCR_EXTRACT": return job.get("source_type") != "OFFLINE" if node_code == "SOURCE_ADD": return not job.get("needs_source_add") return False def _next_node(node_code: str, job: dict) -> Optional[str]: """按显式转移表找下一节点;遇到不满足条件的条件节点自动继续往后跳。""" nxt = TRANSITIONS.get(node_code) # 一路跳过不满足条件的条件节点 while nxt is not None and nxt in CONDITIONAL_NODES and _should_skip(nxt, job): nxt = TRANSITIONS.get(nxt) return nxt def _advance(cur, job: dict, node_code: str) -> Optional[str]: """把 job 从 node_code 推进到下一节点,处理终点/合同审批等待/普通三种落法。 返回 next_node(None=整单结束)。调用方需保证 node_code 已记 SUCCESS。""" job_id = job["job_id"] nxt = _next_node(node_code, job) if nxt is None: cur.execute("UPDATE job SET status='SUCCESS',finished_at=%s,updated_at=%s WHERE job_id=%s", (_now(), _now(), job_id)) _log_event(cur, job_id, "JOB_FINISHED", None, "SYSTEM", None) elif nxt == "ARCHIVE_JOB": # 归档是状态中心自己的收口,进入即完成,不派 RPA、不挂人工。 now = _now() cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,started_at,ended_at,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s,%s)" " ON DUPLICATE KEY UPDATE status='SUCCESS',ended_at=VALUES(ended_at)", (job_id, "ARCHIVE_JOB", NODE_EXECUTOR["ARCHIVE_JOB"], 1, "SUCCESS", now, now, now), ) cur.execute( "UPDATE job SET status='SUCCESS',finished_at=%s,updated_at=%s,current_node_code=%s WHERE job_id=%s", (now, now, "ARCHIVE_JOB", job_id), ) _log_event(cur, job_id, "JOB_FINISHED", None, "SYSTEM", None) return None elif nxt == "CONTRACT_APPROVE_WAIT": # 进等待外部态:不立刻派 RPA,挂 WAITING_EXTERNAL,到点由调度器触发查一次。 next_check = _now() + timedelta(minutes=settings.CONTRACT_WAIT_INTERVAL_MIN) cur.execute( "UPDATE job SET current_node_code=%s,status='WAITING_EXTERNAL'," "next_check_at=%s,wait_check_count=0,updated_at=%s WHERE job_id=%s", (nxt, next_check, _now(), job_id), ) _log_event(cur, job_id, "NODE_STARTED", nxt, "SYSTEM", None, {"wait": "CONTRACT_APPROVE", "next_check_at": next_check.isoformat()}) elif nxt in HUMAN_NODE_ROLES: role_code = HUMAN_NODE_ROLES[nxt] _mark_node_waiting_user(cur, job_id, nxt) cur.execute( "UPDATE job SET current_node_code=%s,status='WAITING_USER'," "current_role=%s,updated_at=%s WHERE job_id=%s", (nxt, role_code, _now(), job_id), ) if nxt != "SOURCE_ADD": _create_human_task(cur, job, nxt, role_code) _log_event(cur, job_id, "MANUAL_HANDOFF", nxt, "SYSTEM", None, {"role_code": role_code}) else: cur.execute("UPDATE job SET current_node_code=%s,status='PENDING',updated_at=%s WHERE job_id=%s", (nxt, _now(), job_id)) return nxt def _log_event(cur, job_id, event_type, node_code, operator_type, operator, detail=None): cur.execute( "INSERT INTO job_event(job_id,event_type,node_code,operator_type,operator,detail_json,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s)", (job_id, event_type, node_code, operator_type, operator, json.dumps(detail, ensure_ascii=False) if detail else None, _now()), ) def _serialize_visible_roles(visible_roles: Optional[list[str]]) -> Optional[str]: if visible_roles is None: return None return json.dumps(visible_roles, ensure_ascii=False) def _parse_visible_roles(value): if value in (None, ""): return [] if isinstance(value, list): return [str(item) for item in value] try: parsed = json.loads(value) except (TypeError, ValueError): parsed = None if isinstance(parsed, list): return [str(item) for item in parsed] return [part.strip() for part in str(value).split(",") if part.strip()] def _record_job_artifacts(cur, job_id: str, node_code: str, attempt_no: int, artifacts: Optional[list["NodeArtifactReq"]]): if not artifacts: return for artifact in artifacts: cur.execute( "INSERT INTO job_artifact(job_id,node_code,attempt_no,artifact_type,storage_path," "file_hash,visible_roles,is_masked,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s)", ( job_id, node_code, attempt_no, artifact.artifactType, artifact.storagePath, artifact.fileHash, _serialize_visible_roles(artifact.visibleRoles), 1 if artifact.isMasked else 0, _now(), ), ) def _fetch_job_artifacts(cur, job_id: str) -> list[dict]: cur.execute( "SELECT node_code,attempt_no,artifact_type,storage_path,file_hash,visible_roles," "is_masked,created_at FROM job_artifact WHERE job_id=%s ORDER BY id", (job_id,), ) rows = cur.fetchall() out = [] for row in rows: out.append({ "node_code": row.get("node_code"), "attempt_no": row.get("attempt_no"), "artifact_type": row.get("artifact_type"), "storage_path": row.get("storage_path"), "file_hash": row.get("file_hash"), "visible_roles": _parse_visible_roles(row.get("visible_roles")), "is_masked": bool(row.get("is_masked")), "created_at": row.get("created_at"), }) return out def _mark_node_waiting_user(cur, job_id: str, node_code: str): cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,started_at,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s)" " ON DUPLICATE KEY UPDATE status=VALUES(status),started_at=COALESCE(started_at,VALUES(started_at))", (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), 1, "WAITING_USER", _now(), _now()), ) def _mark_node_success(cur, job_id: str, node_code: str): """把当前人工/内部节点记 SUCCESS。 已有 WAITING_USER 行则原地收口;合同审批超时这类已有 FAILED 检查历史的, 另起 attempt,不覆盖查审批记录。 """ cur.execute( "SELECT attempt_no,status FROM job_node WHERE job_id=%s AND node_code=%s" " ORDER BY attempt_no DESC LIMIT 1", (job_id, node_code), ) latest = cur.fetchone() if latest and latest.get("status") == "WAITING_USER": cur.execute( "UPDATE job_node SET status='SUCCESS',ended_at=%s" " WHERE job_id=%s AND node_code=%s AND attempt_no=%s", (_now(), job_id, node_code, latest["attempt_no"]), ) return next_attempt = int((latest or {}).get("attempt_no") or 0) + 1 now = _now() cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,started_at,ended_at,created_at)" " VALUES(%s,%s,%s,%s,'SUCCESS',%s,%s,%s)", (job_id, node_code, NODE_EXECUTOR.get(node_code, "HUMAN"), next_attempt, now, now, now), ) def _create_human_task(cur, job: dict, node_code: str, role_code: str): job_id = job["job_id"] cur.execute( "DELETE FROM user_task WHERE job_id=%s AND node_code=%s AND status='OPEN'", (job_id, node_code), ) title_prefix = HUMAN_TASK_TITLES.get(node_code, node_code) cur.execute( "INSERT INTO user_task(task_id,job_id,node_code,role_code,title,reason_type," "reason_user_friendly,why_me_text,status,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s,%s,'OPEN',%s)", ( uuid.uuid4().hex, job_id, node_code, role_code, f"{title_prefix}:{job['supplier_name']}", node_code, HUMAN_TASK_REASONS.get(node_code), "该节点一期由人工处理，确认完成后流程继续。", _now(), ), ) def _supplier_has_source(cur, supplier_name: str, goods_id: str, current_job_id: str) -> bool: """用状态中心历史结果判断该供应商是否已有该货品货源。 一期没有 M5 货源 API,所以用历史已确认/已命中的 supplier+goods_id 作为明确触发源。 """ cur.execute( "SELECT 1 FROM job j JOIN product_match pm ON pm.job_id=j.job_id" " WHERE j.supplier_name=%s AND pm.goods_id=%s AND j.job_id<>%s" " AND pm.match_status IN ('EXACT','ALIAS') LIMIT 1", (supplier_name, goods_id, current_job_id), ) return cur.fetchone() is not None def _source_add_needed( cur, job: dict, goods_id: Optional[str], spec: str = "", manufacturer: str = "", ) -> int: if not goods_id: return 0 from .config import settings from .m5_client import ( FUJIAN_PATH, _load_json, evaluate_priority_source, load_connection, query_priority_source, recorded_source_rows, ) if spec and manufacturer: recorded = recorded_source_rows(str(goods_id)) if recorded is not None and settings.DEV_MODE: entry_id = int(_load_json(FUJIAN_PATH)["entryId"]) judged = evaluate_priority_source( recorded, entry_id=entry_id, spec=spec, manufacturer=manufacturer, ) return 1 if judged["source_add_needed"] else 0 cfg = load_connection() if cfg.get("user") and cfg.get("password") and spec and manufacturer: try: entry_id = int(_load_json(FUJIAN_PATH)["entryId"]) rows = query_priority_source(str(goods_id), cfg) judged = evaluate_priority_source( rows, entry_id=entry_id, spec=spec, manufacturer=manufacturer, ) return 1 if judged["source_add_needed"] else 0 except Exception: pass return 0 if _supplier_has_source(cur, job["supplier_name"], goods_id, job["job_id"]) else 1 def _build_source_add_candidates(cur, job_id: str) -> list[dict]: """从 product_match 当前态生成 SOURCE_ADD 待办明细,按 goods_id 去重。""" cur.execute( "SELECT pm.id AS product_match_id,pm.order_item_id,pm.goods_id,pm.goods_code," "pm.base_unit,pm.match_key,oi.raw_product_name,oi.spec,oi.manufacturer" " FROM product_match pm JOIN order_item oi ON oi.id=pm.order_item_id" " WHERE pm.job_id=%s AND pm.source_add_needed=1 AND pm.goods_id IS NOT NULL" " ORDER BY pm.id", (job_id,), ) out = [] seen_goods = set() for row in cur.fetchall(): goods_id = row["goods_id"] if goods_id in seen_goods: continue seen_goods.add(goods_id) out.append({ "goods_id": goods_id, "goods_code": row["goods_code"], "base_unit": row["base_unit"], "raw_product_name": row["raw_product_name"], "spec": row["spec"], "manufacturer": row["manufacturer"], "order_item_id": row["order_item_id"], "product_match_id": row["product_match_id"], "match_key": row["match_key"], }) return out def _sync_source_add_task(cur, job: dict) -> list[dict]: """把 product_match.source_add_needed 同步到 job.needs_source_add 和一条聚合待办。""" job_id = job["job_id"] candidates = _build_source_add_candidates(cur, job_id) cur.execute( "DELETE FROM user_task WHERE job_id=%s AND node_code='SOURCE_ADD' AND status='OPEN'", (job_id,), ) cur.execute( "UPDATE job SET needs_source_add=%s,updated_at=%s WHERE job_id=%s", (1 if candidates else 0, _now(), job_id), ) if not candidates: return [] detail = { "supplier_name": job["supplier_name"], "items": candidates, } cur.execute( "INSERT INTO user_task(task_id,job_id,node_code,role_code,title,reason_type," "reason_user_friendly,why_me_text,comparison_snapshot_json,candidates_json,status,created_at)" " VALUES(%s,%s,'SOURCE_ADD','PURCHASER',%s,'SOURCE_ADD',%s,%s,%s,%s,'OPEN',%s)", ( uuid.uuid4().hex, job_id, f"新增货源:{job['supplier_name']}", "该供应商未维护这些品种的货源,需要在 M5 新增后继续。", "供应商货源由采购/业务在 M5 维护。", json.dumps(detail, ensure_ascii=False), json.dumps(candidates, ensure_ascii=False), _now(), ), ) _log_event(cur, job_id, "TASK_CREATED", "SOURCE_ADD", "SYSTEM", None, detail) return candidates # ---------------- 建单 ---------------- class CreateJobReq(BaseModel): source_type: str # YYC/XYY/OFFLINE order_date: str # YYYY-MM-DD supplier_name: str created_by: str = "system" class OrderItemRowReq(BaseModel): orderNo: str rawProductName: str platformProductCode: Optional[str] = None spec: Optional[str] = None manufacturer: Optional[str] = None approvalNo: Optional[str] = None qty: float unitPrice: Optional[float] = None sourceRowNo: Optional[int] = None class OrderItemsIngestReq(BaseModel): replace: bool = True rows: list[OrderItemRowReq] @app.post("/api/jobs") def create_job(req: CreateJobReq): job_id = f"{req.source_type}-{req.supplier_name}-{req.order_date.replace('-', '')}-{uuid.uuid4().hex[:6]}" business_key = f"{req.source_type}|{req.supplier_name}|{req.order_date}" need_pdf = 1 if req.source_type == "YYC" else 0 first_node = NODE_ORDER[0] conn = get_conn() with conn.cursor() as cur: # 幂等:同 business_key 已有未结束 job 就不重复建 cur.execute( "SELECT job_id FROM job WHERE business_key=%s AND status NOT IN" " ('SUCCESS','CANCELLED','FAILED_MANUAL') LIMIT 1", (business_key,), ) exist = cur.fetchone() if exist: return {"job_id": exist["job_id"], "reused": True} cur.execute( "INSERT INTO job(job_id,business_key,source_type,order_date,supplier_name," "need_pdf_upload,current_node_code,status,auto_save_enabled,created_by,created_at,updated_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s,'PENDING',0,%s,%s,%s)", (job_id, business_key, req.source_type, req.order_date, req.supplier_name, need_pdf, first_node, req.created_by, _now(), _now()), ) _log_event(cur, job_id, "JOB_CREATED", None, "SYSTEM", req.created_by, {"business_key": business_key}) return {"job_id": job_id, "first_node": first_node, "reused": False} @app.post("/api/jobs/{job_id}/items") def ingest_order_items(job_id: str, req: OrderItemsIngestReq, x_agent_token: Optional[str] = Header(default=None)): """开发/接入侧写入订单明细。已有对码结果后禁止改写，避免破坏下游状态。""" _check_agent(x_agent_token) conn = get_conn() with conn.cursor() as cur: _load_job(cur, job_id) cur.execute("SELECT COUNT(*) AS c FROM product_match WHERE job_id=%s", (job_id,)) if cur.fetchone()["c"]: raise HTTPException(status_code=409, detail="已有对码结果，不能覆盖订单明细") if req.replace: cur.execute("DELETE FROM order_item WHERE job_id=%s", (job_id,)) order_nos = [] for row in req.rows: if row.orderNo not in order_nos: order_nos.append(row.orderNo) cur.execute( "INSERT INTO order_item(job_id,order_no,raw_product_name,platform_product_code," "spec,manufacturer,approval_no,qty,unit_price,source_row_no,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", ( job_id, row.orderNo, row.rawProductName, row.platformProductCode, row.spec, row.manufacturer, row.approvalNo, row.qty, row.unitPrice, row.sourceRowNo, _now(), ), ) cur.execute( "UPDATE job SET order_no_list_json=%s,updated_at=%s WHERE job_id=%s", (json.dumps(order_nos, ensure_ascii=False), _now(), job_id), ) _log_event(cur, job_id, "ORDER_ITEMS_INGESTED", "ORDER_FETCH", "ROBOT", None, {"replace": req.replace, "inserted": len(req.rows), "order_nos": order_nos}) return {"ok": True, "inserted": len(req.rows), "order_nos": order_nos} @app.get("/api/jobs") def list_jobs(status: Optional[str] = None, source_type: Optional[str] = None, supplier_name: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None): conn = get_conn() sql = ( "SELECT job_id,supplier_name,order_date,source_type,status,current_node_code," "current_role,created_at,updated_at,finished_at FROM job WHERE 1=1" ) params = [] if status: sql += " AND status=%s" params.append(status) if source_type: sql += " AND source_type=%s" params.append(source_type) if supplier_name: sql += " AND supplier_name LIKE %s" params.append(f"%{supplier_name}%") if date_from: sql += " AND order_date>=%s" params.append(date_from) if date_to: sql += " AND order_date<=%s" params.append(date_to) sql += " ORDER BY created_at DESC" with conn.cursor() as cur: cur.execute(sql, tuple(params)) rows = cur.fetchall() return {"items": rows} @app.get("/api/tasks/my") def list_my_tasks(roleCode: Optional[str] = None, status: Optional[str] = None): conn = get_conn() sql = ( "SELECT task_id,job_id,node_code,order_item_id,product_match_id,role_code,status,title," "reason_type,reason_user_friendly,why_me_text,candidates_json,submitted_action," "submitted_at,created_at FROM user_task WHERE 1=1" ) params = [] if roleCode: sql += " AND role_code=%s" params.append(roleCode) if status: sql += " AND status=%s" params.append(status.upper()) sql += " ORDER BY created_at DESC" with conn.cursor() as cur: cur.execute(sql, tuple(params)) rows = cur.fetchall() return {"items": rows} # ---------------- 节点开始 ---------------- class NodeStartReq(BaseModel): attemptNo: int = 1 executorPlatform: Optional[str] = None operator: Optional[str] = "agent" def _load_job(cur, job_id: str) -> dict: cur.execute("SELECT * FROM job WHERE job_id=%s", (job_id,)) job = cur.fetchone() if not job: raise HTTPException(status_code=404, detail="job not found") return job def _guard_current_node(job: dict, node_code: str): """乱序拒绝:回调的节点必须是该 job 当前节点,否则 409。 避免 RPA 端网络重发/顺序错乱把状态推乱。""" if node_code not in NODE_ORDER: raise HTTPException(status_code=400, detail=f"unknown node {node_code}") if job.get("current_node_code") != node_code: raise HTTPException( status_code=409, detail=f"out-of-order: job 当前在 {job.get('current_node_code')},拒绝 {node_code} 回调", ) @app.post("/api/jobs/{job_id}/nodes/{node_code}/start") def node_start(job_id: str, node_code: str, req: NodeStartReq, x_agent_token: Optional[str] = Header(default=None)): _check_agent(x_agent_token) conn = get_conn() with conn.cursor() as cur: job = _load_job(cur, job_id) _guard_current_node(job, node_code) # 幂等:该节点该次尝试已 RUNNING/SUCCESS,直接回 ok,不重复起点 cur.execute( "SELECT status FROM job_node WHERE job_id=%s AND node_code=%s AND attempt_no=%s", (job_id, node_code, req.attemptNo), ) row = cur.fetchone() if row and row["status"] in ("RUNNING", "SUCCESS"): return {"ok": True, "idempotent": True} cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,executor_platform," "executor_run_id,started_at,created_at) VALUES(%s,%s,%s,%s,'RUNNING',%s,NULL,%s,%s)" " ON DUPLICATE KEY UPDATE status='RUNNING',started_at=VALUES(started_at)", (job_id, node_code, NODE_EXECUTOR.get(node_code, "RPA"), req.attemptNo, req.executorPlatform, _now(), _now()), ) cur.execute( "UPDATE job SET status='RUNNING',updated_at=%s WHERE job_id=%s", (_now(), job_id), ) _log_event(cur, job_id, "NODE_STARTED", node_code, "ROBOT", req.operator) return {"ok": True} # ---------------- 节点成功 ---------------- class NodeArtifactReq(BaseModel): artifactType: str storagePath: str fileHash: Optional[str] = None visibleRoles: Optional[list[str]] = None isMasked: bool = False class NodeCompleteReq(BaseModel): attemptNo: int = 1 executorRunId: Optional[str] = None idempotencyKey: Optional[str] = None # RPA 端生成的回调幂等键,重发同键不二次推进 businessRefs: Optional[dict] = None # {contractSystemNo:..., rebateOrderNo:...} summary: Optional[dict] = None artifacts: Optional[list[NodeArtifactReq]] = None endedAt: Optional[str] = None @app.post("/api/jobs/{job_id}/nodes/{node_code}/complete") def node_complete(job_id: str, node_code: str, req: NodeCompleteReq, x_agent_token: Optional[str] = Header(default=None)): _check_agent(x_agent_token) if NODE_EXECUTOR.get(node_code) == "HUMAN": raise HTTPException(status_code=409, detail="人工节点只能通过待办提交，拒绝 RPA complete") # 约束:要求业务编号却没回传的节点,不许记 SUCCESS required = {"CONTRACT_CREATE": "contractSystemNo", "REBATE_HEADER": "rebateOrderNo", "M5_RECEIPT_NOTICE": "receiptNoticeNo"} if node_code in required: key = required[node_code] if not (req.businessRefs or {}).get(key): raise HTTPException(status_code=400, detail=f"{node_code} 必须回传 {key}") conn = get_conn() with conn.cursor() as cur: job = _load_job(cur, job_id) # 幂等:该节点该次尝试已 SUCCESS,直接回上次算出的 next,不二次推进 cur.execute( "SELECT status,callback_idempotency_key FROM job_node" " WHERE job_id=%s AND node_code=%s AND attempt_no=%s", (job_id, node_code, req.attemptNo), ) prev = cur.fetchone() if prev and prev["status"] == "SUCCESS": return {"ok": True, "next_node": _next_node(node_code, job), "idempotent": True} # 乱序拒绝:非当前节点的 complete 一律 409(幂等重发已在上面放行) _guard_current_node(job, node_code) cur.execute( "UPDATE job_node SET status='SUCCESS',executor_run_id=%s,ended_at=%s," "callback_idempotency_key=%s,output_json=%s" " WHERE job_id=%s AND node_code=%s AND attempt_no=%s", (req.executorRunId, _now(), req.idempotencyKey, json.dumps({"businessRefs": req.businessRefs, "summary": req.summary}, ensure_ascii=False), job_id, node_code, req.attemptNo), ) # 回写业务编号(三把钥匙等) if req.businessRefs: for k, v in req.businessRefs.items(): col = BUSINESS_REF_FIELD.get(k) if col and v: cur.execute(f"UPDATE job SET {col}=%s,updated_at=%s WHERE job_id=%s", (v, _now(), job_id)) _record_job_artifacts(cur, job_id, node_code, req.attemptNo, req.artifacts) _log_event(cur, job_id, "NODE_SUCCESS", node_code, "ROBOT", None, req.summary) # 决定下一步(显式转移表 + 条件节点跳过) nxt = _advance(cur, job, node_code) return {"ok": True, "next_node": nxt} # ---------------- 节点失败 ---------------- class NodeFailReq(BaseModel): attemptNo: int = 1 errorCode: str errorSummary: Optional[str] = None retryable: bool = False routeRole: Optional[str] = None artifacts: Optional[list[NodeArtifactReq]] = None @app.post("/api/jobs/{job_id}/nodes/{node_code}/fail") def node_fail(job_id: str, node_code: str, req: NodeFailReq, x_agent_token: Optional[str] = Header(default=None)): _check_agent(x_agent_token) # M5 质管缺失:不重试、不等质管,直接转人工手工做 force_manual = req.errorCode == "M5_QC_MISSING" conn = get_conn() with conn.cursor() as cur: job = _load_job(cur, job_id) _guard_current_node(job, node_code) # 只有当前节点能报失败,乱序 409 cur.execute( "UPDATE job_node SET status='FAILED',error_code=%s,error_summary=%s," "role_at_failure=%s,ended_at=%s WHERE job_id=%s AND node_code=%s AND attempt_no=%s", (req.errorCode, req.errorSummary, req.routeRole, _now(), job_id, node_code, req.attemptNo), ) _record_job_artifacts(cur, job_id, node_code, req.attemptNo, req.artifacts) if force_manual: job_status = "FAILED_MANUAL" elif req.retryable: job_status = "FAILED_RETRYABLE" else: job_status = "WAITING_USER" # 业务类转人工待办 cur.execute("UPDATE job SET status=%s,current_role=%s,updated_at=%s WHERE job_id=%s", (job_status, req.routeRole, _now(), job_id)) _log_event(cur, job_id, "NODE_FAILED", node_code, "ROBOT", None, {"errorCode": req.errorCode, "retryable": req.retryable}) return {"ok": True, "job_status": job_status} @app.post("/api/jobs/{job_id}/retry") def retry_job(job_id: str, x_agent_token: Optional[str] = Header(default=None)): """把 FAILED_RETRYABLE 的当前节点放回可 start 状态。 不覆盖失败历史:下一轮 start 用 attempt_no+1 另起 job_node 行。 """ _check_agent(x_agent_token) conn = get_conn() with conn.cursor() as cur: job = _load_job(cur, job_id) if job.get("status") != "FAILED_RETRYABLE": raise HTTPException( status_code=409, detail=f"job 不在可重试状态(当前 {job.get('status')})", ) node_code = job["current_node_code"] cur.execute( "SELECT COALESCE(MAX(attempt_no), 0) AS max_attempt FROM job_node" " WHERE job_id=%s AND node_code=%s", (job_id, node_code), ) row = cur.fetchone() or {} next_attempt = int(row.get("max_attempt") or 0) + 1 cur.execute( "UPDATE job SET status='PENDING',updated_at=%s WHERE job_id=%s", (_now(), job_id), ) _log_event( cur, job_id, "NODE_STARTED", node_code, "SYSTEM", None, {"retry": True, "attempt_no": next_attempt}, ) return {"ok": True, "node_code": node_code, "attempt_no": next_attempt} # ---------------- 合同审批等待(定时查一次,不连续轮询) ---------------- @app.get("/api/scheduler/due-contract-checks") def due_contract_checks(x_agent_token: Optional[str] = Header(default=None)): """调度器/RPA 定时扫这个:到点该查审批的 job 列表。 业务方案:送审后每 CONTRACT_WAIT_INTERVAL_MIN 分钟查一次,不死等。""" _check_agent(x_agent_token) conn = get_conn() with conn.cursor() as cur: cur.execute( "SELECT job_id,contract_system_no,wait_check_count,next_check_at FROM job" " WHERE status='WAITING_EXTERNAL' AND current_node_code='CONTRACT_APPROVE_WAIT'" " AND next_check_at IS NOT NULL AND next_check_at<=%s ORDER BY next_check_at", (_now(),), ) rows = cur.fetchall() return {"due": rows, "interval_min": settings.CONTRACT_WAIT_INTERVAL_MIN, "max_checks": settings.CONTRACT_WAIT_MAX_CHECKS} class ContractCheckReq(BaseModel): approved: bool # RPA 查完:审批是否已通过 operator: Optional[str] = "agent" detail: Optional[str] = None @app.post("/api/jobs/{job_id}/nodes/CONTRACT_APPROVE_WAIT/check-result") def contract_check_result(job_id: str, req: ContractCheckReq, x_agent_token: Optional[str] = Header(default=None)): """RPA 到点登录查了一次审批状态,把结果回给状态中心。 - 通过 → 该节点 SUCCESS,推进到 REBATE_HEADER。 - 未通过 → 次数+1;没超上限则排下一次查(next_check_at);超了升级 OPS(WAITING_USER)。 """ _check_agent(x_agent_token) conn = get_conn() with conn.cursor() as cur: job = _load_job(cur, job_id) if job.get("current_node_code") != "CONTRACT_APPROVE_WAIT" or job.get("status") != "WAITING_EXTERNAL": raise HTTPException(status_code=409, detail=f"job 不在合同审批等待态(当前 {job.get('current_node_code')}/{job.get('status')})") # 本次是第几次查(1-based)。每次查都单独落一行 job_node(不覆盖历史), # attempt_no 递增 → 命中 uniq_job_node_attempt 也不冲突,审计轨迹完整。 attempt = int(job.get("wait_check_count") or 0) + 1 if req.approved: cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status," "started_at,ended_at,created_at) VALUES(%s,'CONTRACT_APPROVE_WAIT',%s,%s,'SUCCESS',%s,%s,%s)", (job_id, NODE_EXECUTOR["CONTRACT_APPROVE_WAIT"], attempt, _now(), _now(), _now()), ) nxt = _next_node("CONTRACT_APPROVE_WAIT", job) # REBATE_HEADER cur.execute( "UPDATE job SET current_node_code=%s,status='PENDING',wait_check_count=%s," "next_check_at=NULL,updated_at=%s WHERE job_id=%s", (nxt, attempt, _now(), job_id), ) _log_event(cur, job_id, "NODE_SUCCESS", "CONTRACT_APPROVE_WAIT", "ROBOT", req.operator, {"approved": True, "checks": attempt}) return {"ok": True, "next_node": nxt} # 未通过:本次查也单独落一行(记为一次尝试),再决定重排还是升级 over_limit = attempt >= settings.CONTRACT_WAIT_MAX_CHECKS cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,error_code," "error_summary,role_at_failure,started_at,ended_at,created_at)" " VALUES(%s,'CONTRACT_APPROVE_WAIT',%s,%s,%s,%s,%s,%s,%s,%s,%s)", (job_id, NODE_EXECUTOR["CONTRACT_APPROVE_WAIT"], attempt, "FAILED" if over_limit else "WAITING_EXTERNAL", "CONTRACT_APPROVE_TIMEOUT" if over_limit else None, req.detail, "OPS" if over_limit else None, _now(), _now(), _now()), ) if over_limit: # 超上限:升级 OPS 人工介入,并建一条可提交的待办。 cur.execute( "UPDATE job SET status='WAITING_USER',current_role='OPS'," "wait_check_count=%s,next_check_at=NULL,updated_at=%s WHERE job_id=%s", (attempt, _now(), job_id), ) _create_human_task(cur, job, "CONTRACT_APPROVE_WAIT", "OPS") _log_event(cur, job_id, "MANUAL_HANDOFF", "CONTRACT_APPROVE_WAIT", "SYSTEM", None, {"reason": "approve_timeout", "checks": attempt}) return {"ok": True, "escalated": True, "job_status": "WAITING_USER"} # 没超:排下一次查 next_check = _now() + timedelta(minutes=settings.CONTRACT_WAIT_INTERVAL_MIN) cur.execute( "UPDATE job SET wait_check_count=%s,next_check_at=%s,updated_at=%s WHERE job_id=%s", (attempt, next_check, _now(), job_id), ) _log_event(cur, job_id, "NODE_STARTED", "CONTRACT_APPROVE_WAIT", "ROBOT", req.operator, {"approved": False, "checks": attempt, "next_check_at": next_check.isoformat()}) return {"ok": True, "approved": False, "checks": attempt, "next_check_at": next_check.isoformat()} # ---------------- 对码(PRODUCT_MATCH,状态中心内部算) ---------------- # # 骨架说明:对码是状态中心自己算的节点(executor=STATE_CENTER),不派 RPA。 # 依赖两样东西,拿到后本节点才真正跑得动: # 1) order_item 已落库(建单/ORDER_FETCH 后把订单明细写进来)——Trae 任务; # 2) 货品主数据装载进 MatchEngine(开工时从 Oracle 全量拉)——需 Oracle 真连。 # 当前给出可推进的骨架:有明细就跑 match_engine 落 product_match、判 needs_source_add、 # MULTI/NONE 建 user_task 转人工,全命中则推进下一节点;没明细/引擎没装载则明确报错,不静默卡。 # 进程内对码引擎(开工 reload 一次,后续内存匹配)。多进程部署时各进程各自 reload。 _ENGINE = MatchEngine() _ENGINE_LOADED = False @app.post("/api/match/reload-goods") def reload_goods(x_agent_token: Optional[str] = Header(default=None)): """从 Oracle 全量拉货品主数据装载进内存引擎。需先配好 ORACLE_* 且 pip install oracledb。""" _check_agent(x_agent_token) if not settings.ORACLE_DSN: raise HTTPException(status_code=400, detail="未配置 ORACLE_DSN,无法装载货品主数据") global _ENGINE_LOADED records = load_goods_from_oracle(settings) _ENGINE.load_goods(records) _ENGINE_LOADED = True return {"ok": True, "loaded": len(records)} @app.post("/api/dev/match/load-fixture-goods") def load_fixture_goods(x_agent_token: Optional[str] = Header(default=None)): """仅本地联调用，不依赖 Oracle 装载一组可重复的货品样例。""" _check_agent(x_agent_token) if not settings.DEV_MODE: raise HTTPException(status_code=404, detail="开发接口未启用") records = [ GoodsRecord("阿卡波糖片", "国药准字H00000001", "50mg*30片", "盒", "示例制药", "G-001", "YP-001"), GoodsRecord("阿莫西林胶囊", "国药准字H00000002", "0.25g*24粒", "盒", "示例医药", "G-002", "YP-002"), GoodsRecord("氯化钠注射液", "国药准字H00000003", "100ml", "瓶", "示例药业", "G-003", "YP-003"), GoodsRecord( "哈药牌钙铁锌口服液", "食健备G201923000558", "10ml*90支", "盒", "哈药集团三精制药有限公司", "4000109983", "4000109983", ), ] global _ENGINE_LOADED _ENGINE.load_goods(records) _ENGINE_LOADED = True return {"ok": True, "loaded": len(records)} def _load_aliases_into_engine(cur) -> int: """把 mapping_config(PRODUCT,active)里的累积别名装进引擎 alias_map。 target_value 存的是 {"goods_id":..,"goods_code":..,"base_unit":..} 的 JSON。""" cur.execute( "SELECT config_key,target_value FROM mapping_config" " WHERE config_type='PRODUCT' AND is_active=1" " ORDER BY config_key,updated_at,id", ) alias = {} for row in cur.fetchall(): try: tv = json.loads(row["target_value"]) if row["target_value"] else {} except (TypeError, ValueError): continue gid = tv.get("goods_id") if gid: alias[row["config_key"]] = (gid, tv.get("goods_code", ""), tv.get("base_unit", "")) _ENGINE.load_aliases(alias) return len(alias) @app.post("/api/jobs/{job_id}/match/run") def match_run(job_id: str, x_agent_token: Optional[str] = Header(default=None)): """对该 job 的 order_item 全量跑对码,落 product_match。 - 全部 EXACT/ALIAS(且不缺货源)→ 推进下一节点。 - 有 MULTI/NONE → 逐条建 user_task,job 转 WAITING_USER(单条挂起,整批继续落库)。 """ _check_agent(x_agent_token) if not _ENGINE_LOADED: raise HTTPException(status_code=409, detail="对码引擎未装载货品主数据,先调 /api/match/reload-goods") conn = get_conn() try: conn.begin() with conn.cursor() as cur: job = _load_job(cur, job_id) _guard_current_node(job, "PRODUCT_MATCH") cur.execute( "SELECT id,raw_product_name,approval_no,spec,manufacturer FROM order_item" " WHERE job_id=%s ORDER BY id", (job_id,), ) items = cur.fetchall() if not items: # 明确报错,不静默推进(否则等于跳过对码) raise HTTPException(status_code=400, detail=f"job {job_id} 没有 order_item,无法对码") _load_aliases_into_engine(cur) # 对码结果是当前态,重跑时用 job_id+order_item_id upsert,避免重复 product_match/user_task。 cur.execute( "DELETE FROM user_task WHERE job_id=%s AND node_code IN ('PRODUCT_MATCH','SOURCE_ADD')" " AND status='OPEN'", (job_id,), ) stat = {"EXACT": 0, "ALIAS": 0, "MULTI": 0, "NONE": 0, "SOURCE_ADD": 0} need_match_human = False for it in items: r = _ENGINE.match_one(it["id"], it["raw_product_name"], it["approval_no"] or "", it["spec"] or "", it["manufacturer"] or "") stat[r.match_status] = stat.get(r.match_status, 0) + 1 source_add_needed = _source_add_needed( cur, job, r.goods_id, it.get("spec") or "", it.get("manufacturer") or "" ) cur.execute( "INSERT INTO product_match(job_id,order_item_id,match_key,match_status," "goods_id,goods_code,base_unit,candidates_json,source_add_needed,created_at)" " VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)" " ON DUPLICATE KEY UPDATE match_key=VALUES(match_key),match_status=VALUES(match_status)," "goods_id=VALUES(goods_id),goods_code=VALUES(goods_code),base_unit=VALUES(base_unit)," "candidates_json=VALUES(candidates_json),source_add_needed=VALUES(source_add_needed)," "confirmed_by=NULL,confirmed_at=NULL,id=LAST_INSERT_ID(id)", (job_id, it["id"], r.match_key, r.match_status, r.goods_id, r.goods_code, r.base_unit, json.dumps(r.candidates, ensure_ascii=False) if r.candidates else None, source_add_needed, _now()), ) match_id = cur.lastrowid if r.match_status in ("MULTI", "NONE"): need_match_human = True reason = "MATCH_MULTI" if r.match_status == "MULTI" else "MATCH_NONE" cur.execute( "INSERT INTO user_task(task_id,job_id,node_code,order_item_id,product_match_id," "role_code,title,reason_type,candidates_json,status,created_at)" " VALUES(%s,%s,'PRODUCT_MATCH',%s,%s,'PLANNER',%s,%s,%s,'OPEN',%s)", (uuid.uuid4().hex, job_id, it["id"], match_id, f"对码待确认:{it['raw_product_name']}", reason, json.dumps(r.candidates, ensure_ascii=False) if r.candidates else None, _now()), ) source_add_candidates = _sync_source_add_task(cur, job) stat["SOURCE_ADD"] = len(source_add_candidates) cur.execute( "INSERT INTO job_node(job_id,node_code,executor_type,attempt_no,status,started_at,ended_at,created_at)" " VALUES(%s,'PRODUCT_MATCH',%s,1,%s,%s,%s,%s)" " ON DUPLICATE KEY UPDATE status=VALUES(status),ended_at=VALUES(ended_at)", (job_id, NODE_EXECUTOR["PRODUCT_MATCH"], "WAITING_USER" if need_match_human else "SUCCESS", _now(), _now(), _now()), ) _log_event(cur, job_id, "NODE_SUCCESS" if not need_match_human else "TASK_CREATED", "PRODUCT_MATCH", "SYSTEM", None, stat) if need_match_human: # 有待确认:job 转人工待办,不推进(人工 confirm 后再推进)。 cur.execute( "UPDATE job SET status='WAITING_USER',current_role='PLANNER',updated_at=%s WHERE job_id=%s", (_now(), job_id), ) conn.commit() return {"ok": True, "stat": stat, "waiting_user": True, "source_add_needed": bool(source_add_candidates)} # 全部已明确匹配:按 needs_source_add 决定是否进入 SOURCE_ADD,否则进合同建单。 job = _load_job(cur, job_id) nxt = _advance(cur, job, "PRODUCT_MATCH") conn.commit() return {"ok": True, "stat": stat, "next_node": nxt, "source_add_needed": bool(source_add_candidates)} except Exception: conn.rollback() raise finally: conn.close() class MatchConfirmReq(BaseModel): goods_id: str # 人工选定/新增的货品ID goods_code: Optional[str] = None base_unit: Optional[str] = None confirmed_by: str = "planner" save_as_alias: bool = True # 是否把这次选择沉淀成累积别名(下次自动命中) @app.post("/api/match/{match_id}/confirm") def match_confirm(match_id: int, req: MatchConfirmReq, x_agent_token: Optional[str] = Header(default=None)): """人工确认一条 MULTI/NONE 对码结果: - 回写 product_match(goods_id/编码/单位、确认人)。 - save_as_alias=1 时把 match_key→货品ID 沉淀进 mapping_config(PRODUCT),下次自动 ALIAS 命中。 - 关掉对应 user_task;若该 job 已无待确认,推进 PRODUCT_MATCH 到下一节点。 注意:只写 mapping_config 累积别名,绝不回写 Oracle 货品主数据(主数据只读)。 """ _check_agent(x_agent_token) conn = get_conn() try: conn.begin() with conn.cursor() as cur: cur.execute("SELECT * FROM product_match WHERE id=%s", (match_id,)) pm = cur.fetchone() if not pm: raise HTTPException(status_code=404, detail="product_match not found") job_id = pm["job_id"] job = _load_job(cur, job_id) oi = None if pm.get("order_item_id"): cur.execute( "SELECT spec,manufacturer FROM order_item WHERE id=%s", (pm["order_item_id"],), ) oi = cur.fetchone() source_add_needed = _source_add_needed( cur, job, req.goods_id, (oi or {}).get("spec") or "", (oi or {}).get("manufacturer") or "", ) cur.execute( "UPDATE product_match SET match_status='ALIAS',goods_id=%s,goods_code=%s,base_unit=%s," "source_add_needed=%s,confirmed_by=%s,confirmed_at=%s WHERE id=%s", (req.goods_id, req.goods_code, req.base_unit, source_add_needed, req.confirmed_by, _now(), match_id), ) if req.save_as_alias: target = json.dumps({"goods_id": req.goods_id, "goods_code": req.goods_code, "base_unit": req.base_unit}, ensure_ascii=False) cur.execute( "UPDATE mapping_config SET is_active=0,effective_to=%s,updated_by=%s,updated_at=%s" " WHERE config_type='PRODUCT' AND config_key=%s AND is_active=1", (_now(), req.confirmed_by, _now(), pm["match_key"]), ) cur.execute( "INSERT INTO mapping_config(config_type,config_key,target_value,is_active," "effective_from,updated_by,updated_at)" " VALUES('PRODUCT',%s,%s,1,%s,%s,%s)", (pm["match_key"], target, _now(), req.confirmed_by, _now()), ) # 精确关掉这条对码结果对应的待办(按 product_match_id 关联,不再靠"最早 OPEN"猜) cur.execute( "UPDATE user_task SET status='SUBMITTED',submitted_by=%s,submitted_at=%s" " WHERE product_match_id=%s AND status='OPEN'", (req.confirmed_by, _now(), match_id), ) _log_event(cur, job_id, "TASK_SUBMITTED", "PRODUCT_MATCH", "USER", req.confirmed_by, {"match_id": match_id, "goods_id": req.goods_id, "source_add_needed": bool(source_add_needed)}) source_add_candidates = _sync_source_add_task(cur, job) # 还有没有未解决的对码?(product_match 里仍是 MULTI/NONE 的) cur.execute( "SELECT COUNT(*) AS c FROM product_match WHERE job_id=%s AND match_status IN ('MULTI','NONE')", (job_id,), ) remaining = cur.fetchone()["c"] if remaining == 0: job = _load_job(cur, job_id) if job.get("current_node_code") == "PRODUCT_MATCH": cur.execute( "UPDATE job_node SET status='SUCCESS',ended_at=%s" " WHERE job_id=%s AND node_code='PRODUCT_MATCH' AND attempt_no=1", (_now(), job_id), ) nxt = _advance(cur, job, "PRODUCT_MATCH") conn.commit() return {"ok": True, "all_resolved": True, "next_node": nxt, "source_add_needed": bool(source_add_candidates)} conn.commit() return {"ok": True, "all_resolved": remaining == 0, "remaining": remaining, "source_add_needed": bool(source_add_candidates)} except Exception: conn.rollback() raise finally: conn.close() # ---------------- 人工待办提交(Web/H5 用) ---------------- class TaskSubmitReq(BaseModel): action: str payload: Optional[dict] = None submitToken: str operator: Optional[str] = "user" def _submit_candidate_selection(cur, task: dict, job: dict, payload: dict, operator: str) -> dict: """在当前事务/连接中完成一条对码候选确认。""" match_id = payload.get("productMatchId") goods_id = payload.get("goodsId") if not match_id or not goods_id: raise HTTPException( status_code=400, detail="选择候选货品时必须提供 productMatchId 和 goodsId", ) if task.get("node_code") != "PRODUCT_MATCH": raise HTTPException(status_code=400, detail="该待办不是对码待办") if task.get("product_match_id") and int(match_id) != int(task["product_match_id"]): raise HTTPException(status_code=409, detail="候选货品与当前待办不匹配") if job.get("current_node_code") != "PRODUCT_MATCH": raise HTTPException(status_code=409, detail="当前单据已不在对码节点") cur.execute("SELECT * FROM product_match WHERE id=%s", (match_id,)) pm = cur.fetchone() if not pm or pm.get("job_id") != task["job_id"]: raise HTTPException(status_code=404, detail="对码结果不存在") oi = None if pm.get("order_item_id"): cur.execute( "SELECT spec,manufacturer FROM order_item WHERE id=%s", (pm["order_item_id"],), ) oi = cur.fetchone() source_add_needed = _source_add_needed( cur, job, goods_id, (oi or {}).get("spec") or "", (oi or {}).get("manufacturer") or "", ) goods_code = payload.get("goodsCode") base_unit = payload.get("baseUnit") cur.execute( "UPDATE product_match SET match_status='ALIAS',goods_id=%s,goods_code=%s,base_unit=%s," "source_add_needed=%s,confirmed_by=%s,confirmed_at=%s WHERE id=%s", (goods_id, goods_code, base_unit, source_add_needed, operator, _now(), match_id), ) if payload.get("saveAsAlias", True): target = json.dumps( {"goods_id": goods_id, "goods_code": goods_code, "base_unit": base_unit}, ensure_ascii=False, ) cur.execute( "UPDATE mapping_config SET is_active=0,effective_to=%s,updated_by=%s,updated_at=%s" " WHERE config_type='PRODUCT' AND config_key=%s AND is_active=1", (_now(), operator, _now(), pm["match_key"]), ) cur.execute( "INSERT INTO mapping_config(config_type,config_key,target_value,is_active," "effective_from,updated_by,updated_at) VALUES('PRODUCT',%s,%s,1,%s,%s,%s)", (pm["match_key"], target, _now(), operator, _now()), ) source_add_candidates = _sync_source_add_task(cur, job) cur.execute( "SELECT COUNT(*) AS c FROM product_match WHERE job_id=%s AND match_status IN ('MULTI','NONE')", (task["job_id"],), ) remaining = cur.fetchone()["c"] result = { "ok": True, "all_resolved": remaining == 0, "remaining": remaining, "source_add_needed": bool(source_add_candidates), } if remaining == 0: cur.execute( "UPDATE job_node SET status='SUCCESS',ended_at=%s" " WHERE job_id=%s AND node_code='PRODUCT_MATCH' AND attempt_no=1", (_now(), task["job_id"]), ) result["next_node"] = _advance(cur, job, "PRODUCT_MATCH") return result @app.post("/api/tasks/{task_id}/submit") def task_submit(task_id: str, req: TaskSubmitReq, x_agent_token: Optional[str] = Header(default=None)): _check_agent(x_agent_token) if not req.submitToken or len(req.submitToken) > 64: raise HTTPException(status_code=400, detail="submitToken 最长 64") conn = get_conn() began = False try: begin = getattr(conn, "begin", None) if callable(begin): begin() began = True result = _task_submit_body(conn, task_id, req) if began: conn.commit() return result except Exception: rollback = getattr(conn, "rollback", None) if callable(rollback): rollback() raise finally: close = getattr(conn, "close", None) if callable(close): close() def _task_submit_body(conn, task_id: str, req: TaskSubmitReq): with conn.cursor() as cur: cur.execute("SELECT * FROM user_task WHERE task_id=%s", (task_id,)) task = cur.fetchone() if not task: raise HTTPException(status_code=404, detail="task not found") existing_token = task.get("submit_token") if existing_token: if existing_token != req.submitToken: raise HTTPException(status_code=409, detail="task already submitted with another token") try: result = json.loads(task.get("submitted_result_json") or "{}") except (TypeError, ValueError): result = {"ok": True} result["idempotent"] = True return result if task.get("status") != "OPEN": raise HTTPException(status_code=409, detail=f"task status is {task.get('status')}") job = _load_job(cur, task["job_id"]) node_code = task["node_code"] payload = req.payload or {} if req.action == "SELECT_CANDIDATE": result = _submit_candidate_selection(cur, task, job, payload, req.operator or "user") elif req.action == "CONFIRM_AND_CONTINUE": if job.get("current_node_code") != node_code: raise HTTPException( status_code=409, detail=f"job 当前在 {job.get('current_node_code')},不能提交 {node_code}", ) if node_code == "M5_RECEIPT_NOTICE": receipt_no = payload.get("receiptNoticeNo") if not receipt_no: raise HTTPException(status_code=400, detail="M5_RECEIPT_NOTICE 需要 receiptNoticeNo") cur.execute( "UPDATE job SET receipt_notice_no=%s,updated_at=%s WHERE job_id=%s", (receipt_no, _now(), task["job_id"]), ) if node_code == "SOURCE_ADD": cur.execute( "UPDATE job SET needs_source_add=0,updated_at=%s WHERE job_id=%s", (_now(), task["job_id"]), ) _mark_node_success(cur, task["job_id"], node_code) job = _load_job(cur, task["job_id"]) nxt = _advance(cur, job, node_code) result = {"ok": True, "next_node": nxt} elif req.action == "TERMINATE_JOB": cur.execute( "UPDATE job SET status='CANCELLED',finished_at=%s,updated_at=%s WHERE job_id=%s", (_now(), _now(), task["job_id"]), ) _log_event(cur, task["job_id"], "TASK_SUBMITTED", node_code, "USER", req.operator, {"action": req.action, "payload": payload}) result = {"ok": True, "job_status": "CANCELLED"} elif req.action in ("TRANSFER_MANUAL", "RETRY_CURRENT_NODE"): if req.action == "TRANSFER_MANUAL": target_role = payload.get("targetRole") if target_role: cur.execute( "UPDATE job SET current_role=%s,updated_at=%s WHERE job_id=%s", (target_role, _now(), task["job_id"]), ) _log_event(cur, task["job_id"], "TASK_SUBMITTED", node_code, "USER", req.operator, {"action": req.action, "payload": payload}) result = {"ok": True, "job_status": job.get("status"), "node_code": node_code} else: raise HTTPException(status_code=400, detail=f"unsupported task action {req.action}") result_json = json.dumps(result, ensure_ascii=False) cur.execute( "UPDATE user_task SET status='SUBMITTED',submit_token=%s,submitted_action=%s," "submitted_payload_json=%s,submitted_result_json=%s,submitted_by=%s,submitted_at=%s" " WHERE task_id=%s", ( req.submitToken, req.action, json.dumps(payload, ensure_ascii=False), result_json, req.operator, _now(), task_id, ), ) _log_event(cur, task["job_id"], "TASK_SUBMITTED", node_code, "USER", req.operator, {"action": req.action, "payload": payload, "result": result}) return result # ---------------- 整单查询(前端画树状图用) ---------------- @app.get("/api/jobs/{job_id}") def get_job(job_id: str): conn = get_conn() with conn.cursor() as cur: cur.execute("SELECT * FROM job WHERE job_id=%s", (job_id,)) job = cur.fetchone() if not job: raise HTTPException(status_code=404, detail="job not found") cur.execute("SELECT node_code,attempt_no,status,started_at,ended_at,duration_ms," "error_code,error_summary,output_json FROM job_node WHERE job_id=%s" " ORDER BY id", (job_id,)) nodes = cur.fetchall() cur.execute("SELECT task_id,node_code,role_code,reason_type,status FROM user_task" " WHERE job_id=%s ORDER BY id", (job_id,)) tasks = cur.fetchall() artifacts = _fetch_job_artifacts(cur, job_id) return {"job": job, "nodes": nodes, "tasks": tasks, "artifacts": artifacts, "node_order": NODE_ORDER, "node_executor": NODE_EXECUTOR, "conditional_nodes": sorted(CONDITIONAL_NODES)} @app.get("/api/jobs/{job_id}/timeline") def get_job_timeline(job_id: str): conn = get_conn() with conn.cursor() as cur: cur.execute("SELECT * FROM job WHERE job_id=%s", (job_id,)) job = cur.fetchone() if not job: raise HTTPException(status_code=404, detail="job not found") cur.execute( "SELECT node_code,attempt_no,status,executor_type,started_at,ended_at " "FROM job_node WHERE job_id=%s ORDER BY id", (job_id,), ) nodes = cur.fetchall() return {"job": job, "timeline": build_timeline(job, nodes)} @app.get("/api/jobs/{job_id}/evidence-package") def get_evidence_package(job_id: str): conn = get_conn() with conn.cursor() as cur: cur.execute("SELECT job_id FROM job WHERE job_id=%s", (job_id,)) job = cur.fetchone() if not job: raise HTTPException(status_code=404, detail="job not found") artifacts = _fetch_job_artifacts(cur, job_id) return {"job_id": job_id, "artifacts": artifacts} @app.get("/health") def health(): conn = None try: conn = get_conn() with conn.cursor() as cur: cur.execute("SELECT 1") cur.fetchone() return {"ok": True, "database": "ok"} except Exception as exc: raise HTTPException(status_code=503, detail=f"database unavailable: {exc}") from exc finally: if conn is not None: conn.close() @app.get("/api/state-machine") def get_state_machine(): return build_state_machine_meta()