AutoRpa/state_center

状态中心(采购主链RPA一期 · 第1步骨架)

Python + FastAPI + MySQL。库里只存流程状态 + 对码累积别名 + 证据;对码货品主数据在 Oracle,直连查(不复制)。

目录

state_center/
├── sql/schema.sql          建库脚本(job 那套 + mapping_config + 证据)
├── app/
│   ├── config.py           配置(MySQL/Oracle 连接、表名列名,从环境变量读)
│   ├── db.py               MySQL 访问(pymysql)
│   ├── match_engine.py     对码引擎(两级命中 + 规格标准化 + Oracle 取数)
│   └── main.py             FastAPI:建单 + 回调(含幂等/乱序拒绝)+ 合同审批等待 + 整单查询
├── tests/test_match_engine.py  对码单测(不依赖 DB/Oracle)
└── requirements.txt

起步

以下命令在 macOS/Linux 的终端执行；Windows PowerShell 只需要把激活命令换成下一段对应写法，接口和浏览器端完全相同。

python -m pip install -r state_center/requirements.txt

# 1) 建库(先起个 MySQL 8)
mysql -uroot -p < state_center/sql/schema.sql

# 2) 配连接(默认 root/root@127.0.0.1)
export MYSQL_DSN="mysql+pymysql://root:root@127.0.0.1:3306/rpa_center?charset=utf8mb4"

# 3) 起服务
uvicorn state_center.app.main:app --reload --port 8000
# 打开 http://127.0.0.1:8000/docs 有全部接口可点

Windows PowerShell：

py -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r state_center\requirements.txt
$env:DEV_MODE = "true"
$env:MYSQL_DSN = "mysql+pymysql://root:root@127.0.0.1:3306/rpa_center?charset=utf8mb4"
python -m uvicorn state_center.app.main:app --reload --port 8000

浏览器端另开一个终端：

cd state_center_web
npm install
npm run dev

然后打开 http://localhost:5173/。Windows 和 macOS 均使用标准浏览器访问，不需要 Electron 或系统专用脚本。DEV_MODE=true 只用于本地样例货品和节点回调模拟，生产环境不要开启。

本地样例联调顺序：建单 → 写入订单明细(POST /api/jobs/{job_id}/items) → 开发接口装载样例货品(POST /api/dev/match/load-fixture-goods) → 运行对码。真实环境把样例接口替换为 Oracle 货品装载即可。

验证第1步(不接真实系统,假回调走全链)

# 建单
curl -X POST http://127.0.0.1:8000/api/jobs -H 'Content-Type: application/json' \
  -d '{"source_type":"YYC","order_date":"2026-08-04","supplier_name":"重庆亿昊药业"}'
# 拿到 job_id 后,模拟艺赛旗回调(带 token):
curl -X POST http://127.0.0.1:8000/api/jobs/<JOB_ID>/nodes/ORDER_FETCH/start \
  -H 'X-Agent-Token: dev-agent-token' -H 'Content-Type: application/json' -d '{"attemptNo":1}'
curl -X POST http://127.0.0.1:8000/api/jobs/<JOB_ID>/nodes/ORDER_FETCH/complete \
  -H 'X-Agent-Token: dev-agent-token' -H 'Content-Type: application/json' \
  -d '{"attemptNo":1,"summary":{"orderCount":12}}'
# 反复 start/complete 走完各节点,最后 GET 看整单轨迹:
curl http://127.0.0.1:8000/api/jobs/<JOB_ID>

节点编排(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_type=OFFLINE)	线上单(医药城/小药药)
SOURCE_ADD	对码判定缺货源(needs_source_add=1)	货源齐全

转移不是"数组下一个",是 main.py 里的显式转移表 TRANSITIONS + 条件跳过。 每个节点标了执行方 NODE_EXECUTOR:STATE_CENTER(对码,内部算)/ RPA(艺赛旗点页面)/ HUMAN(一期人工)。 所以文档里"11 个节点"和代码里"13 个"是同一套:11 主 + 2 条件 = 13。

关键设计点(会审确认)

• 回调幂等:同节点同 attempt 重复 complete 不会二次推进(靠 job_node 状态 + callback_idempotency_key)。
• 乱序拒绝:回调节点必须是 job 当前节点,否则 409(避免网络重发把状态推乱)。
• 合同审批等待(CONTRACT_APPROVE_WAIT)不连续轮询:进 WAITING_EXTERNAL 态,每 CONTRACT_WAIT_INTERVAL_MIN(默认 30 分)由调度器扫 GET /api/scheduler/due-contract-checks 触发 RPA 查一次,结果回 POST .../CONTRACT_APPROVE_WAIT/check-result;查满 CONTRACT_WAIT_MAX_CHECKS(默认 8 次)还没过就升级 OPS。
• 对码状态只有 4 种:EXACT / ALIAS / MULTI / NONE(无 FUZZY;ALIAS=累积别名命中,非当次人工)。
• PRODUCT_MATCH 是状态中心自己算的节点(executor=STATE_CENTER),不走 RPA 三回调,靠 POST /api/jobs/{id}/match/run 推进:跑 order_item → 落 product_match → MULTI/NONE 建 待办转 WAITING_USER,全命中自动往下走;POST /api/match/{id}/confirm 人工确认后沉淀 mapping_config 别名并推进。前置:order_item 已落库 + POST /api/match/reload-goods 从 Oracle 装载过货品主数据(这两样是 Trae 任务2/3 的接真数据部分)。
• SOURCE_ADD 触发源明确:对码拿到 goods_id 后,状态中心按历史 supplier_name + goods_id 判断是否已有货源;没历史则写 product_match.source_add_needed=1、同步 job.needs_source_add=1 并创建一条聚合的 SOURCE_ADD 待办。
• 对码重跑幂等:product_match 以 (job_id, order_item_id) 为当前态唯一键, match/run 重跑时 upsert 结果并重建 OPEN 待办,不重复造结果/待办。
• alias 当前态唯一:match/confirm 会先失效同一 config_key 的旧 active 记录,再插入新 active; 装载 alias 时按 updated_at/id 固定顺序读取,历史脏数据也有确定结果。

对码接 Oracle(拿到连接串后)

1. pip install oracledb
2. 在 config.py / 环境变量填 ORACLE_DSN/USER/PASSWORD,把 ORACLE_GOODS_TABLE 和各列名改成真实的。
3. 开工时 load_goods_from_oracle(settings) 拉全量 → MatchEngine.load_goods(...),之后内存匹配。

对码单测

python3 state_center/tests/test_match_engine.py   # 6/6 passed