""" 对码引擎(核心难点)。 数据源: - 货品主数据在 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