125 lines
6.1 KiB
Markdown
125 lines
6.1 KiB
Markdown
# 状态中心(采购主链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 只需要把激活命令换成下一段对应写法，接口和浏览器端完全相同。
|
||
|
||
```bash
|
||
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：
|
||
|
||
```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
|
||
```
|
||
|
||
浏览器端另开一个终端：
|
||
|
||
```bash
|
||
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步(不接真实系统,假回调走全链)
|
||
|
||
```bash
|
||
# 建单
|
||
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(...)`,之后内存匹配。
|
||
|
||
## 对码单测
|
||
|
||
```bash
|
||
python3 state_center/tests/test_match_engine.py # 6/6 passed
|
||
```
|