216 lines
7.3 KiB
Python
216 lines
7.3 KiB
Python
"""
|
||
对码引擎(核心难点)。
|
||
|
||
数据源:
|
||
- 货品主数据在 Oracle(只读)。开工时可全量 load 进内存,后续内存匹配。
|
||
- mapping_config(PRODUCT):平台奇葩写法→货品ID 的人工确认累积,存在自己 MySQL。
|
||
|
||
查找顺序(命中率越用越高):
|
||
1. 先查 mapping_config 累积别名 → 命中直接出货品ID(ALIAS,累积别名命中,非人工介入)
|
||
2. 再查货品主数据(内存/Oracle)四要素精确键 → EXACT
|
||
3. 规格/厂家容错后同商品名多条 → MULTI(出候选)
|
||
4. 都没有 → NONE(转人工)
|
||
|
||
对码结果状态只有 4 种:EXACT / ALIAS / MULTI / NONE。
|
||
(无 FUZZY:本引擎不产出模糊命中档,拿不准就走 MULTI 出候选让人选。)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
|
||
# ---------- 规范化:对付"15×2 vs 30片""干一瓶 vs ×1瓶""进口英文厂家" ----------
|
||
|
||
_FULL2HALF = {ord(" "): " ", ord("（"): "(", ord("）"): ")", ord("＊"): "*", ord("×"): "*"}
|
||
|
||
|
||
def _norm_text(s: Optional[str]) -> str:
|
||
"""通用规范化:去空格、全半角、大小写、常见标点。"""
|
||
if not s:
|
||
return ""
|
||
s = s.translate(_FULL2HALF)
|
||
s = s.lower().strip()
|
||
s = re.sub(r"[\s,，。.、；;:：'\"“”‘’]+", "", s)
|
||
return s
|
||
|
||
|
||
def _norm_manufacturer(s: Optional[str]) -> str:
|
||
"""厂家名容错:进口药英文名统一大小写、去标点(B.V. / S.r.l. 之类)。"""
|
||
if not s:
|
||
return ""
|
||
s = s.lower().strip()
|
||
s = re.sub(r"[.\s,]+", "", s) # 去点、空格、逗号
|
||
return s
|
||
|
||
|
||
def _norm_spec(s: Optional[str]) -> str:
|
||
"""
|
||
规格标准化。把 '15×2' '15*2' 展开成乘积;'30片' '×1瓶' 抽出数字。
|
||
返回一个可比较的规范串,如 '30' 或 '30片'。
|
||
这是痛点,规则先覆盖会议里点名的几种,后续可加。
|
||
"""
|
||
if not s:
|
||
return ""
|
||
raw = s.translate(_FULL2HALF).lower().strip()
|
||
raw = re.sub(r"\s+", "", raw)
|
||
|
||
# 形如 15*2 / 15x2 → 乘积 30
|
||
m = re.search(r"(\d+(?:\.\d+)?)\*(\d+(?:\.\d+)?)", raw)
|
||
if m:
|
||
product = float(m.group(1)) * float(m.group(2))
|
||
product = int(product) if product == int(product) else product
|
||
# 保留单位后缀(如 片/瓶),便于跟 '30片' 对齐
|
||
unit = re.sub(r"[\d*.]+", "", raw)
|
||
return f"{product}{unit}"
|
||
|
||
return raw
|
||
|
||
|
||
def build_match_key(common_name: str, approval_no: str, spec: str, manufacturer: str) -> str:
|
||
"""四要素拼接键:通用名 + 批准文号 + 规格 + 厂家(全部规范化)。"""
|
||
return "|".join([
|
||
_norm_text(common_name),
|
||
_norm_text(approval_no),
|
||
_norm_spec(spec),
|
||
_norm_manufacturer(manufacturer),
|
||
])
|
||
|
||
|
||
# ---------- 货品主数据(Oracle 镜像到内存) ----------
|
||
|
||
@dataclass
|
||
class GoodsRecord:
|
||
common_name: str
|
||
approval_no: str
|
||
spec: str
|
||
base_unit: str
|
||
manufacturer: str
|
||
goods_id: str
|
||
goods_code: str
|
||
|
||
|
||
@dataclass
|
||
class MatchResult:
|
||
order_item_id: int
|
||
match_key: str
|
||
match_status: str # EXACT / ALIAS / MULTI / NONE
|
||
goods_id: Optional[str] = None
|
||
goods_code: Optional[str] = None
|
||
base_unit: Optional[str] = None
|
||
candidates: list = field(default_factory=list)
|
||
|
||
|
||
class MatchEngine:
|
||
"""
|
||
内存索引 + 两级查找。
|
||
goods_index: 四要素键 -> [GoodsRecord] (可能一键多货品,即一对多)
|
||
name_index: 规范通用名 -> [GoodsRecord] (用于同商品名多规格出候选)
|
||
alias_map: mapping_config 累积别名 config_key -> (goods_id, goods_code, base_unit)
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.goods_index: dict[str, list[GoodsRecord]] = {}
|
||
self.name_index: dict[str, list[GoodsRecord]] = {}
|
||
self.alias_map: dict[str, tuple[str, str, str]] = {}
|
||
|
||
# -- 装载 --
|
||
def load_goods(self, records: list[GoodsRecord]) -> None:
|
||
self.goods_index.clear()
|
||
self.name_index.clear()
|
||
for r in records:
|
||
key = build_match_key(r.common_name, r.approval_no, r.spec, r.manufacturer)
|
||
self.goods_index.setdefault(key, []).append(r)
|
||
self.name_index.setdefault(_norm_text(r.common_name), []).append(r)
|
||
|
||
def load_aliases(self, aliases: dict[str, tuple[str, str, str]]) -> None:
|
||
"""aliases: config_key -> (goods_id, goods_code, base_unit)"""
|
||
self.alias_map = dict(aliases)
|
||
|
||
# -- 单行对码 --
|
||
def match_one(self, order_item_id: int, common_name: str, approval_no: str,
|
||
spec: str, manufacturer: str) -> MatchResult:
|
||
key = build_match_key(common_name, approval_no, spec, manufacturer)
|
||
|
||
# 1) 累积别名优先(ALIAS = 历史人工确认沉淀的别名命中,不是当次人工介入)
|
||
if key in self.alias_map:
|
||
gid, gcode, unit = self.alias_map[key]
|
||
return MatchResult(order_item_id, key, "ALIAS", gid, gcode, unit)
|
||
|
||
# 2) 四要素精确
|
||
hits = self.goods_index.get(key, [])
|
||
if len(hits) == 1:
|
||
r = hits[0]
|
||
return MatchResult(order_item_id, key, "EXACT", r.goods_id, r.goods_code, r.base_unit)
|
||
if len(hits) > 1:
|
||
return MatchResult(order_item_id, key, "MULTI",
|
||
candidates=[self._cand(r) for r in hits])
|
||
|
||
# 3) 同商品名多规格 → 出候选给人工选
|
||
name_hits = self.name_index.get(_norm_text(common_name), [])
|
||
if name_hits:
|
||
return MatchResult(order_item_id, key, "MULTI",
|
||
candidates=[self._cand(r) for r in name_hits])
|
||
|
||
# 4) 无命中
|
||
return MatchResult(order_item_id, key, "NONE")
|
||
|
||
@staticmethod
|
||
def _cand(r: GoodsRecord) -> dict:
|
||
return {
|
||
"goods_id": r.goods_id,
|
||
"goods_code": r.goods_code,
|
||
"common_name": r.common_name,
|
||
"spec": r.spec,
|
||
"base_unit": r.base_unit,
|
||
"manufacturer": r.manufacturer,
|
||
"approval_no": r.approval_no,
|
||
}
|
||
|
||
|
||
# ---------- Oracle 取数(拿到真实连接串后启用) ----------
|
||
|
||
def load_goods_from_oracle(settings) -> list[GoodsRecord]:
|
||
"""
|
||
从 Oracle 只读拉全量货品主数据。
|
||
需要: pip install oracledb
|
||
真实表名/列名在 config.py 里改。
|
||
"""
|
||
import oracledb # 延迟导入,没配 Oracle 时不影响其他功能
|
||
|
||
sql = f"""
|
||
SELECT {settings.ORACLE_COL_COMMON_NAME},
|
||
{settings.ORACLE_COL_APPROVAL_NO},
|
||
{settings.ORACLE_COL_SPEC},
|
||
{settings.ORACLE_COL_UNIT},
|
||
{settings.ORACLE_COL_MANUFACTURER},
|
||
{settings.ORACLE_COL_GOODS_ID},
|
||
{settings.ORACLE_COL_GOODS_CODE}
|
||
FROM {settings.ORACLE_GOODS_TABLE}
|
||
"""
|
||
conn = oracledb.connect(
|
||
user=settings.ORACLE_USER,
|
||
password=settings.ORACLE_PASSWORD,
|
||
dsn=settings.ORACLE_DSN,
|
||
)
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(sql)
|
||
rows = cur.fetchall()
|
||
finally:
|
||
conn.close()
|
||
|
||
out = []
|
||
for row in rows:
|
||
out.append(GoodsRecord(
|
||
common_name=str(row[0] or ""),
|
||
approval_no=str(row[1] or ""),
|
||
spec=str(row[2] or ""),
|
||
base_unit=str(row[3] or ""),
|
||
manufacturer=str(row[4] or ""),
|
||
goods_id=str(row[5] or ""),
|
||
goods_code=str(row[6] or ""),
|
||
))
|
||
return out
|