linyuhang
/
AutoRpa
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add Fujian M5 match and source-priority fixtures 

Offline confirmation for 哈药牌钙铁锌口服液 uses recorded GOODSID 4000109983 and PRIORITY=1 source rows. M5 host/SID are configured; credentials stay local.
This commit is contained in:
Yuuuu 2026-08-31 17:04:48 +08:00
parent 63eec40334
commit 29df92ee17
12 changed files with 548 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored
View File

@ -33,6 +33,12 @@ rpa-chrome-profile/
# Duplicate / scratch
state_center_副本/
# Local M5 credentials
state_center/fixtures/m5/connection.local.json
.env
.env.*
!.env.example
# Media (keep docs, drop recordings and extracted frames)
*.mp4
video_frames_v2/

55
rpa_yisaiqi/runners/mock.py
View File

@ -29,21 +29,48 @@ def run(node_code: str, job: dict[str, Any], client: Any) -> dict[str, Any]:
def run_order_fetch(job: dict[str, Any], client: Any) -> dict[str, Any]:
from pathlib import Path
import json
fixture = (
Path(__file__).resolve().parents[2]
/ "state_center"
/ "fixtures"
/ "orders"
/ "yyc-fujian-20260804.json"
)
order_date = str(job.get("order_date") or "2026-08-12")
order_no = f"DD-{order_date.replace('-', '')}-001"
rows = [
{
"orderNo": order_no,
"rawProductName": "阿卡波糖片",
"platformProductCode": "YP-001",
"spec": "50mg*30片",
"manufacturer": "示例制药",
"approvalNo": "国药准字H00000001",
"qty": 100,
"unitPrice": 25.5,
"sourceRowNo": 1,
}
]
if fixture.is_file():
data = json.loads(fixture.read_text(encoding="utf-8"))
row = data["row"]
order_no = row.get("orderNo") or f"DD-{order_date.replace('-', '')}-001"
rows = [
{
"orderNo": order_no,
"rawProductName": row["rawProductName"],
"platformProductCode": row.get("platformProductCode"),
"spec": row.get("spec"),
"manufacturer": row.get("manufacturer"),
"approvalNo": row.get("approvalNo"),
"qty": row.get("qty") or 1,
"sourceRowNo": row.get("sourceRowNo"),
}
]
else:
order_no = f"DD-{order_date.replace('-', '')}-001"
rows = [
{
"orderNo": order_no,
"rawProductName": "阿卡波糖片",
"platformProductCode": "YP-001",
"spec": "50mg*30片",
"manufacturer": "示例制药",
"approvalNo": "国药准字H00000001",
"qty": 100,
"unitPrice": 25.5,
"sourceRowNo": 1,
}
]
if hasattr(client, "ingest_items"):
client.ingest_items(job["job_id"], rows, replace=True)
return {

8
state_center/app/config.py
View File

@ -18,6 +18,14 @@ class Settings:
ORACLE_USER = os.getenv("ORACLE_USER", "")
ORACLE_PASSWORD = os.getenv("ORACLE_PASSWORD", "")
# M5 货品视图(福建货主对码)。账号密码只走环境变量，不进仓库。
M5_HOST = os.getenv("M5_HOST", "173.1.1.215")
M5_PORT = int(os.getenv("M5_PORT", "1521"))
M5_SERVICE = os.getenv("M5_SERVICE", "m5")
M5_USER = os.getenv("M5_USER", "")
M5_PASSWORD = os.getenv("M5_PASSWORD", "")
M5_SCHEMA = os.getenv("M5_SCHEMA", "m5")
# Oracle 货品主数据表(真实表名/列名拿到后改这里,先按设计文档 A~G 列占位)
ORACLE_GOODS_TABLE = os.getenv("ORACLE_GOODS_TABLE", "GOODS_MASTER")
ORACLE_COL_COMMON_NAME = os.getenv("ORACLE_COL_COMMON_NAME", "COMMON_NAME") # 通用名

174
state_center/app/m5_client.py Normal file
View File

@ -0,0 +1,174 @@
"""M5 货品视图只读查询。账号密码来自环境变量或 connection.local.json。"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
from .config import settings
FIXTURE_DIR = Path(__file__).resolve().parents[1] / "fixtures"
M5_DIR = FIXTURE_DIR / "m5"
ORDER_PATH = FIXTURE_DIR / "orders" / "yyc-fujian-20260804.json"
FUJIAN_PATH = M5_DIR / "fujian.json"
LOCAL_CONN_PATH = M5_DIR / "connection.local.json"
def _load_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def load_connection() -> dict[str, Any]:
local: dict[str, Any] = {}
if LOCAL_CONN_PATH.is_file():
local = _load_json(LOCAL_CONN_PATH)
host = os.getenv("M5_HOST") or local.get("host") or settings.M5_HOST
port = int(os.getenv("M5_PORT") or local.get("port") or settings.M5_PORT)
service = os.getenv("M5_SERVICE") or local.get("serviceName") or settings.M5_SERVICE
user = os.getenv("M5_USER") or local.get("user") or settings.M5_USER
password = os.getenv("M5_PASSWORD") or local.get("password") or settings.M5_PASSWORD
schema = os.getenv("M5_SCHEMA") or local.get("schema") or settings.M5_SCHEMA
return {
"host": host,
"port": port,
"serviceName": service,
"user": user,
"password": password,
"schema": schema,
}
def recorded_source_rows(goods_id: str) -> list[dict[str, Any]] | None:
"""离线夹具：哈药牌这条 SQL 的 P1 结果，不连 M5。"""
params = lookup_params()
if str(goods_id) != params["expectedGoodsId"]:
return None
return [
{
"ENTRYID": params["entryId"],
"PRIORITY": 1,
"GOODSTYPE": params["spec"],
"FACTORYNAME": params["manufacturer"],
}
]
def evaluate_priority_source(
rows: list[dict[str, Any]],
*,
entry_id: int,
spec: str,
manufacturer: str,
) -> dict[str, Any]:
"""PRIORITY=1 且货主/规格/厂家对上，才算货源齐全。"""
priority_rows = [row for row in rows if str(row.get("PRIORITY") or "") == "1"]
hit = next(
(
row
for row in priority_rows
if str(row.get("ENTRYID") or "") == str(entry_id)
and str(row.get("GOODSTYPE") or "").strip() == spec.strip()
and str(row.get("FACTORYNAME") or "").strip() == manufacturer.strip()
),
None,
)
return {
"source_add_needed": hit is None,
"hit": hit,
"priorityRows": priority_rows,
}
def lookup_params(order: dict[str, Any] | None = None, fujian: dict[str, Any] | None = None) -> dict[str, Any]:
order = order or _load_json(ORDER_PATH)
fujian = fujian or _load_json(FUJIAN_PATH)
row = order["row"]
return {
"goodsNameLike": f"%{row['goodsNameLike']}%",
"approvalNo": row["approvalNo"],
"spec": row["spec"],
"manufacturer": row["manufacturer"],
"entryId": int(order["m5"]["entryId"] or fujian["entryId"]),
"expectedGoodsId": str(row["expectedGoodsId"]),
"sql": fujian["sql"],
"sourceSql": fujian["sourceSql"],
"expectedSourceAddNeeded": bool(row.get("expectedSourceAddNeeded", True)),
}
def connect(conn_cfg: dict[str, Any] | None = None):
cfg = conn_cfg or load_connection()
missing = [key for key in ("user", "password") if not cfg.get(key)]
if missing:
raise RuntimeError(
"M5 连接缺少 " + ", ".join(missing) + "。设 M5_USER/M5_PASSWORD，或写 fixtures/m5/connection.local.json"
)
import oracledb
dsn = f"{cfg['host']}:{cfg['port']}/{cfg['serviceName']}"
return oracledb.connect(user=cfg["user"], password=cfg["password"], dsn=dsn)
def lookup_one(conn_cfg: dict[str, Any] | None = None) -> dict[str, Any]:
params = lookup_params()
conn = connect(conn_cfg)
try:
cur = conn.cursor()
cur.execute(
params["sql"],
goodsNameLike=params["goodsNameLike"],
approvalNo=params["approvalNo"],
spec=params["spec"],
manufacturer=params["manufacturer"],
entryId=params["entryId"],
)
rows = _rows_from_cursor(cur)
finally:
conn.close()
goods_ids = [str(row.get("GOODSID") or "") for row in rows]
return {
"ok": len(rows) == 1 and goods_ids[0] == params["expectedGoodsId"],
"expectedGoodsId": params["expectedGoodsId"],
"hits": len(rows),
"goodsIds": goods_ids,
"rows": rows,
}
def _rows_from_cursor(cur) -> list[dict[str, Any]]:
columns = [col[0] for col in cur.description]
return [dict(zip(columns, row)) for row in cur.fetchall()]
def query_priority_source(goods_id: str, conn_cfg: dict[str, Any] | None = None) -> list[dict[str, Any]]:
fujian = _load_json(FUJIAN_PATH)
conn = connect(conn_cfg)
try:
cur = conn.cursor()
cur.execute(fujian["sourceSql"], goodsId=int(goods_id))
return _rows_from_cursor(cur)
finally:
conn.close()
def check_priority_source(conn_cfg: dict[str, Any] | None = None) -> dict[str, Any]:
params = lookup_params()
rows = query_priority_source(params["expectedGoodsId"], conn_cfg)
judged = evaluate_priority_source(
rows,
entry_id=params["entryId"],
spec=params["spec"],
manufacturer=params["manufacturer"],
)
judged["ok"] = judged["source_add_needed"] == params["expectedSourceAddNeeded"]
judged["expectedSourceAddNeeded"] = params["expectedSourceAddNeeded"]
judged["rows"] = rows
return judged
if __name__ == "__main__":
match = lookup_one()
source = check_priority_source() if match.get("ok") else {"skipped": True, "reason": "对码未命中，不查货源"}
print(json.dumps({"match": match, "source": source}, ensure_ascii=False, default=str, indent=2))

87
state_center/app/main.py
View File

@ -353,9 +353,51 @@ def _supplier_has_source(cur, supplier_name: str, goods_id: str, current_job_id:
return cur.fetchone() is not None
def _source_add_needed(cur, job: dict, goods_id: Optional[str]) -> int:
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
@ -934,6 +976,15 @@ def load_fixture_goods(x_agent_token: Optional[str] = Header(default=None)):
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)
@ -1003,7 +1054,9 @@ def match_run(job_id: str, x_agent_token: Optional[str] = Header(default=None)):
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)
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)"
@ -1094,7 +1147,20 @@ def match_confirm(match_id: int, req: MatchConfirmReq,
raise HTTPException(status_code=404, detail="product_match not found")
job_id = pm["job_id"]
job = _load_job(cur, job_id)
source_add_needed = _source_add_needed(cur, job, req.goods_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,"
@ -1188,7 +1254,20 @@ def _submit_candidate_selection(cur, task: dict, job: dict, payload: dict, opera
if not pm or pm.get("job_id") != task["job_id"]:
raise HTTPException(status_code=404, detail="对码结果不存在")
source_add_needed = _source_add_needed(cur, job, goods_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,
goods_id,
(oi or {}).get("spec") or "",
(oi or {}).get("manufacturer") or "",
)
goods_code = payload.get("goodsCode")
base_unit = payload.get("baseUnit")
cur.execute(

20
state_center/fixtures/README.md Normal file
View File

@ -0,0 +1,20 @@
# 对码入参底座（一条，福建货主）
订单明细走这里，不走证据包。本批只有一条：重庆亿昊 / 哈药牌钙铁锌口服液。
## M5
- 地址默认 `173.1.1.215:1521`，Service Name `m5`
- 账号密码可配，不进 git：环境变量 `M5_USER` `M5_PASSWORD`，或复制 `m5/connection.example.json` → `m5/connection.local.json`
- 查询：品名 LIKE + 文号 + 规格 + 厂家 + `ENTRYID=907`
- 产地空或「无」不参与过滤
- 回写 `goods_id` = `GOODSID`（`4000109983`）
- 对码命中后再查货源：`bms_su_supplygoods_doc_v` × `bms_su_supplygoods_dtl_v`，`PRIORITY` 必须是 1，且 `ENTRYID/GOODSTYPE/FACTORYNAME` 对上这条规格。对上则跳过新增货源。
查这一条：
```bash
export M5_USER=...
export M5_PASSWORD=...
python -m state_center.app.m5_client
```

41
state_center/fixtures/ingest_mapping.json Normal file
View File

@ -0,0 +1,41 @@
{
"version": 1,
"comment": "Excel 列 → 订单入参。本批只收一条。条码/医保不进 order_item，也不拿去查 M5。账号密码不写在这份文件里。",
"sheet": "订单明细",
"skipStatuses": ["已取消"],
"jobGroupBy": ["supplierName", "orderDate", "sourceType"],
"columns": {
"父订单号": "parentOrderNo",
"商家": "supplierName",
"订单状态": "orderStatus",
"品名": "rawProductName",
"产品编码": "platformProductCode",
"产品条码": "barcode",
"医保编码": "insuranceCode",
"批准文号": "approvalNo"
},
"orderItem": [
"orderNo",
"rawProductName",
"platformProductCode",
"spec",
"manufacturer",
"approvalNo",
"qty",
"sourceRowNo"
],
"m5Lookup": {
"view": "PUB_ENTRY_GOODS_V",
"goodsIdField": "GOODSID",
"filters": {
"GOODSNAME": { "from": "rawProductName", "op": "like" },
"APPROVEDOCNO": { "from": "approvalNo", "op": "eq" },
"GOODSTYPE": { "from": "spec", "op": "eq" },
"FACTORYNAME": { "from": "manufacturer", "op": "eq" },
"ENTRYID": { "from": "entryId", "op": "eq" }
},
"optionalFilters": {
"PRODAREA": { "from": "prodArea", "skipValues": ["", "无", null] }
}
}
}

9
state_center/fixtures/m5/connection.example.json Normal file
View File

@ -0,0 +1,9 @@
{
"host": "173.1.1.215",
"port": 1521,
"serviceName": "m5",
"user": "",
"password": "",
"schema": "m5",
"comment": "复制为 connection.local.json 填账号密码。local 文件不进 git。也可用环境变量 M5_USER / M5_PASSWORD / M5_SERVICE。"
}

13
state_center/fixtures/m5/fujian.json Normal file
View File

@ -0,0 +1,13 @@
{
"entryName": "福建",
"entryId": 907,
"host": "173.1.1.215",
"port": 1521,
"serviceName": "m5",
"schema": "m5",
"view": "PUB_ENTRY_GOODS_V",
"goodsIdField": "GOODSID",
"requireProdArea": false,
"sql": "SELECT t.FACTORYNAME, t.APPROVEDOCNO, t.GOODSTYPE, t.PRODAREA, t.ENTRYID, t.GOODSID, t.GOODSNAME, t.GOODSUNIT, t.ENTRYGOODSID FROM PUB_ENTRY_GOODS_V t WHERE t.GOODSNAME LIKE :goodsNameLike AND t.APPROVEDOCNO = :approvalNo AND t.GOODSTYPE = :spec AND t.FACTORYNAME = :manufacturer AND t.ENTRYID = :entryId",
"sourceSql": "SELECT t.ENTRYID, a.PRIORITY, t.GOODSTYPE, t.FACTORYNAME FROM bms_su_supplygoods_doc_v t, bms_su_supplygoods_dtl_v a WHERE t.GOODSID = :goodsId AND a.PRIORITY = 1 AND t.SUPPLYGOODSDOCID = a.SUPPLYGOODSDOCID"
}

26
state_center/fixtures/orders/yyc-fujian-20260804.json Normal file
View File

@ -0,0 +1,26 @@
{
"sourceType": "YYC",
"orderDate": "2026-08-04",
"supplierName": "重庆亿昊药业有限公司",
"m5": {
"entryName": "福建",
"entryId": 907
},
"row": {
"orderNo": "YYC-ROW-2",
"parentOrderNo": "",
"rawProductName": "哈药牌钙铁锌口服液",
"goodsNameLike": "哈药牌",
"platformProductCode": "2919482534",
"barcode": "6938780031263",
"approvalNo": "食健备G201923000558",
"spec": "10ml*90支",
"manufacturer": "哈药集团三精制药有限公司",
"prodArea": "无",
"qty": 83,
"orderStatus": "待付款",
"sourceRowNo": 2,
"expectedGoodsId": "4000109983",
"expectedSourceAddNeeded": false
}
}

101
state_center/tests/test_m5_client.py Normal file
View File

@ -0,0 +1,101 @@
from state_center.app.m5_client import (
evaluate_priority_source,
load_connection,
lookup_params,
recorded_source_rows,
)
def test_lookup_params_use_the_single_fujian_row():
params = lookup_params()
assert params["goodsNameLike"] == "%哈药牌%"
assert params["approvalNo"] == "食健备G201923000558"
assert params["spec"] == "10ml*90支"
assert params["manufacturer"] == "哈药集团三精制药有限公司"
assert params["entryId"] == 907
assert params["expectedGoodsId"] == "4000109983"
assert "PUB_ENTRY_GOODS_V" in params["sql"]
assert "PRODAREA" not in params["sql"].split("WHERE", 1)[1]
def test_m5_host_defaults_and_password_stays_out_of_repo(monkeypatch):
monkeypatch.delenv("M5_USER", raising=False)
monkeypatch.delenv("M5_PASSWORD", raising=False)
monkeypatch.delenv("M5_SERVICE", raising=False)
cfg = load_connection()
assert cfg["host"] == "173.1.1.215"
assert cfg["port"] == 1521
assert cfg["serviceName"] == "m5"
assert cfg["password"] == ""
def test_priority_1_matching_spec_skips_source_add():
judged = evaluate_priority_source(
[
{
"ENTRYID": 907,
"PRIORITY": 1,
"GOODSTYPE": "10ml*90支",
"FACTORYNAME": "哈药集团三精制药有限公司",
}
],
entry_id=907,
spec="10ml*90支",
manufacturer="哈药集团三精制药有限公司",
)
assert judged["source_add_needed"] is False
def test_priority_1_other_spec_needs_source_add():
judged = evaluate_priority_source(
[
{
"ENTRYID": 907,
"PRIORITY": 1,
"GOODSTYPE": "10ml*10支",
"FACTORYNAME": "哈药集团三精制药有限公司",
}
],
entry_id=907,
spec="10ml*90支",
manufacturer="哈药集团三精制药有限公司",
)
assert judged["source_add_needed"] is True
def test_lookup_params_include_source_sql():
params = lookup_params()
assert "bms_su_supplygoods_doc_v" in params["sourceSql"]
assert params["expectedSourceAddNeeded"] is False
def test_recorded_source_rows_cover_hayao_without_m5():
rows = recorded_source_rows("4000109983")
assert rows is not None
judged = evaluate_priority_source(
rows,
entry_id=907,
spec="10ml*90支",
manufacturer="哈药集团三精制药有限公司",
)
assert judged["source_add_needed"] is False
assert recorded_source_rows("other") is None
def test_hayao_priority1_sql_row_skips_source_add():
"""第二条 SQL 的真实结果：907 / 1 / 10ml*90支 / 哈药集团三精制药有限公司。"""
judged = evaluate_priority_source(
[
{
"ENTRYID": 907,
"PRIORITY": 1,
"GOODSTYPE": "10ml*90支",
"FACTORYNAME": "哈药集团三精制药有限公司",
}
],
entry_id=907,
spec="10ml*90支",
manufacturer="哈药集团三精制药有限公司",
)
assert judged["source_add_needed"] is False
assert judged["hit"]["ENTRYID"] == 907

26
state_center/tests/test_match_engine.py
View File

@ -50,6 +50,32 @@ def test_none():
assert r.match_status == "NONE"
def test_fujian_hayao_exact_from_m5_row():
eng = MatchEngine()
eng.load_goods(
[
GoodsRecord(
"哈药牌钙铁锌口服液",
"食健备G201923000558",
"10ml*90支",
"盒",
"哈药集团三精制药有限公司",
"4000109983",
"4000109983",
)
]
)
r = eng.match_one(
1,
"哈药牌钙铁锌口服液",
"食健备G201923000558",
"10ml*90支",
"哈药集团三精制药有限公司",
)
assert r.match_status == "EXACT"
assert r.goods_id == "4000109983"
def test_alias_priority():
eng = _engine()
key = build_match_key("阿卡波糖", "", "30片", "") # 平台奇葩写法