feat(audit): 操作对象补全改为按特征全局解析,摘要联动物料名

此前补全只覆盖「入库/采购」,报废、退回、BOM 仍只显示光秃秃的 SKU 或 bom_no。
根因是解析走的是「快照 base_id / target_id 命中股票表」两条硬路径,
而这几类记录的可用线索完全不同(报废快照有 sku 无 base_id、BOM 只有
parent_id、UPDATE 行两样都没有)。

改为**按特征五级解析,不看 module 名单**(同一 module 内不同记录的线索
本来就不一样,按 module 判断必然覆盖不全):

  1. 快照 base_id            —— 库存行的权威线索
  2. 快照 sku → SKU 索引     —— 报废/退回/借还(单据表有 sku、无 base_id)
  3. target_name 是 SKU      —— 库存/退回/报废 的 UPDATE 行(无快照)
  4. 快照 parent_id          —— BOM 关系行
  5. target_id 唯一命中股票表 —— target_name 为 'stock_buy ID:N' 的入库行

★ 「按 SKU 匹配」不能直接查 material_base —— 那张表**没有 sku 列**,
  SKU 在 7 张流水/库存表上(stock_buy/semi/product、stock_adjustment、
  stock_service、trans_defective_goods、trans_repair)。故建 SKU→base_id
  索引:把这些表 UNION 后要求一个 SKU 只对应一个物料。
  实测并集 2178 个 SKU、0 歧义;审计里 432/438 个 SKU 型 target_name 可解。
  表名从 information_schema 动态枚举并缓存,**不硬编** —— 硬编清单在加表时
  会静默漏掉(本项目反复踩过的漂移)。

★ SKU 形状护栏 ^\d{10}$:实测库里 2178 个 SKU **全部**是 10 位纯数字,
  而字母开头的 target_name(bom_no/单号/角色码)与 SKU 集合命中数为 0 ——
  护栏既不误伤也不漏,挡掉了"拿单号当 SKU 查"的误配风险。

★ 第 4 条取 parent_id 而非 child_id:BOM 行的 bom_no(操作对象)属于父件,
  实测同一 bom_no 的所有子行 parent_id 一致('IH-L2V1J' → parent 44)。
  用子件会显示成只在 BOM 里出现一次的小零件名。

★ 编码段优先用库里的 target_name:单号(BOR-…)、bom_no(IH-L2V1J)本身
  就是业务标识,比 SKU 更能定位记录 —— 把「BOR-…」换成 SKU 是**换掉**了
  有用信息。只有 target_name 是内部兜底形态('stock_buy ID:1667')时才用 SKU。

★ 批量查询:先扫一遍收集线索再批量查,全程恒为 5 次查询(SKU 索引 1 +
  股票表 3 + material_base 1),与页大小无关。实测 20 条与 200 条都是 5 次。

效果(真实数据):
  OUT-20260915-0909-0001 - SIF金属箱门包边 (061工)
  BOR-20260922-0001 - 防水余弦接收器 (FOV0008/FOV0008)
  IH-L2V1J - 穹顶光源V1J (IH-L2V1J/类A)
  0000001180 - 华为路由AX3 wifi6路由器3000M (Det0001/Det0001)
覆盖率 972/4000 → 2985/4000,与快照 base_id **零冲突**。

摘要联动:CREATE/DELETE 摘要的「物料」段取自解析结果(快照里只有 sku,
名称在 material_base 上),形如
  「新增(报废管理):华为路由AX3 wifi6路由器3000M、数量 1 件」。
物料名排在数量/位置之前;短名如「防水余弦接收器」不加截断。

验证:24 + 16 项断言全过(五级路径逐条覆盖、错误率为 0、业务单号保留为
编码段、查询次数恒定、导出口径与列表一致、权限 401/403/200、字典零遗漏);
vue-tsc --noEmit 与 vite build 均 exit=0。
日报回归:三天附件 490.5K / 193.4K / 10.3K,6 列结构未变。
This commit is contained in:
yueli
2026-09-28 10:25:55 +08:00
parent 46aec0b954
commit 91672214c8

View File

@ -30,6 +30,9 @@ import json
import re import re
from collections import Counter, defaultdict from collections import Counter, defaultdict
from sqlalchemy import text
from app.extensions import db
from app.models.base import MaterialBase from app.models.base import MaterialBase
from app.utils.audit_labels import ( from app.utils.audit_labels import (
BOOLEAN_FIELDS, BOOLEAN_FIELDS,
@ -351,6 +354,96 @@ _SNAPSHOT_BASE_ID_SOURCES = (SNAPSHOT_CREATED, SNAPSHOT_DELETED)
# 库存三表 —— 入库数据分别落在采购/半成品/成品库存表里 # 库存三表 —— 入库数据分别落在采购/半成品/成品库存表里
STOCK_TABLES = ('stock_buy', 'stock_semi', 'stock_product') STOCK_TABLES = ('stock_buy', 'stock_semi', 'stock_product')
# SKU 形状:实测库里 2178 个 SKU **全部**是 10 位纯数字。
#
# ★ 这个护栏不是为了好看,是为了防误配:target_name 里混着 bom_no
# ('IH-L2V1J')、单号('BOR-2026…')、角色码('PURCHASER')等等,
# 若不加形状限制就拿去当 SKU 查,理论上可能撞上某个 SKU 而给出**错的**物料名。
# 实测字母开头的 target_name 与 SKU 集合命中数为 0,即护栏不误伤也不漏。
_SKU_RE = re.compile(r'^\d{10}$')
def is_sku_like(text):
"""该字符串是否长得像物料编码(10 位纯数字)"""
return bool(isinstance(text, str) and _SKU_RE.match(text.strip()))
# target_name 的**内部兜底形态**:audit_listener._get_target_name 在找不到
# 可读名称时会写成 'stock_buy ID:1667' —— 表名 + 主键,对业务人员零价值。
_TARGET_NAME_FALLBACK_RE = re.compile(r'^[\w.]+ ID:\d+$')
def _display_code(row, sku):
"""
操作对象的「编码段」该用谁。
★ 优先用库里的 target_name:单号('BOR-20260721-0001')、BOM 编号
('IH-L2V1J')本身就是业务标识,比 SKU 更能定位一条记录 ——
把借还记录的「BOR-…」换成 SKU 是把有用的信息**换掉**了,不是补充。
实测借还/报废这类单据表的 target_name 就是单号,而它们的快照里带着 sku。
只有 target_name 是内部兜底形态('stock_buy ID:1667')或为空时,
才退而用 SKU —— 那种串比 SKU 还不可读。
"""
tname = (getattr(row, 'target_name', '') or '').strip()
if tname and not _TARGET_NAME_FALLBACK_RE.match(tname):
return tname
return (sku or '').strip()
_sku_tables_cache = None
def _sku_source_tables():
"""
同时含 `sku` 与 `base_id` 的物理表名(进程内查一次并缓存)。
★ 不硬编表名列表:实测有 7 张(stock_buy / stock_semi / stock_product /
stock_adjustment / stock_service / trans_defective_goods / trans_repair),
将来加业务表就会变。硬编的清单在加表时会**静默漏掉**新表 ——
这正是本项目反复踩过的漂移坑(前端 moduleMap、双胞胎页面都是这么坏的)。
改为从 information_schema 枚举,自维护。
"""
global _sku_tables_cache
if _sku_tables_cache is None:
rows = db.session.execute(text("""
SELECT a.table_name
FROM information_schema.columns a
JOIN information_schema.columns b
ON a.table_name = b.table_name AND b.column_name = 'base_id'
WHERE a.column_name = 'sku' AND a.table_schema = current_schema()
""")).fetchall()
_sku_tables_cache = tuple(r[0] for r in rows)
return _sku_tables_cache
def load_sku_base_map(skus):
"""
{sku: base_id} —— SKU → 物料。**对应多个物料的 SKU 不返回**。
★ SKU 不在 material_base 上(那张表没有 sku 列),而在流水/库存表上。
故把这些表 UNION 起来取 (sku, base_id),并要求一个 SKU 只对应一个物料。
实测并集 2178 个 SKU、0 个歧义;但「歧义即丢弃」的判断仍然保留 ——
数据是会变的,不能靠今天的一次实测兜底。
"""
wanted = {s.strip() for s in skus if isinstance(s, str) and s.strip()}
if not wanted:
return {}
tables = _sku_source_tables()
if not tables:
return {}
union = " UNION ALL ".join(
f"SELECT sku, base_id FROM {t} WHERE sku IS NOT NULL AND sku <> ''"
for t in tables
)
rows = db.session.execute(text(f"""
SELECT sku, count(DISTINCT base_id) AS n, min(base_id) AS base_id
FROM ({union}) u WHERE sku = ANY(:skus)
GROUP BY sku
"""), {'skus': sorted(wanted)}).fetchall()
return {r[0]: r[2] for r in rows if r[1] == 1}
def _stock_row_lookup(target_ids): def _stock_row_lookup(target_ids):
""" """
@ -388,60 +481,131 @@ def _stock_row_lookup(target_ids):
for sid, tbls in hits.items() if len(tbls) == 1} for sid, tbls in hits.items() if len(tbls) == 1}
def _material_ref_of(row, sku_map, stock_map):
"""
单行审计记录 → (base_id, code) 或 None。
**按特征解析,不按 module 名单** —— 同一个 module(如「报废管理」)里
不同记录的可用线索并不一样(有的快照带 base_id、有的只带 sku、
UPDATE 行两样都没有),按 module 判断必然覆盖不全。
五级,按可靠性从高到低:
1. 快照 base_id —— 库存行的权威线索
2. 快照 sku → SKU 索引 —— 报废/退回 的快照里带 sku
3. target_name 是 SKU —— 库存/退回/报废 的 UPDATE 行(无快照)
4. 快照 parent_id —— BOM 关系行(见下)
5. target_id 唯一命中股票表 —— target_name 是 'stock_buy ID:N' 的入库行
★ 第 4 条为什么取 parent_id 而不是 child_id:BOM 行是「父件用 N 个 子件」
的边,而这条记录的 bom_no(操作对象)属于**父件**。实测同一 bom_no 的
所有子行 parent_id 一致('IH-L2V1J' → parent 44),故用父件才与
操作对象对得上;用子件会显示成一个只在 BOM 里出现一次的小零件名。
★ 两条查找结果(sku_map / stock_map)由调用方**批量**查好传进来:
逐行查库在 50 行的页面上就是上百次查询。
定位不到返回 None —— 宁可不补,也不能补错。
"""
snaps = [s for s in (snapshot_of(row, w) for w in _SNAPSHOT_BASE_ID_SOURCES) if s]
tname = (row.target_name or '').strip()
ref = None # (base_id, 用于编码段的 sku 或 None)
# 1) 快照 base_id
for snap in snaps:
bid = as_int(snap.get('base_id'))
if bid:
ref = (bid, (snap.get('sku') or '').strip() or None)
break
# 2) 快照 sku(报废/退回/借还这类单据表:有 sku、没有 base_id)
if ref is None:
for snap in snaps:
s = str(snap.get('sku') or '').strip()
if s and sku_map.get(s):
ref = (sku_map[s], s)
break
# 3) target_name 本身就是 SKU
if ref is None and is_sku_like(tname) and sku_map.get(tname):
ref = (sku_map[tname], tname)
# 4) BOM 关系行:parent_id 与 child_id 同时在,说明这是一条父子关系边
if ref is None:
for snap in snaps:
if 'parent_id' in snap and 'child_id' in snap:
pid = as_int(snap.get('parent_id'))
if pid:
ref = (pid, None)
break
# 5) target_id 唯一命中一张股票表
if ref is None:
tid = as_int(getattr(row, 'target_id', None))
hit = stock_map.get(tid) if tid is not None else None
if hit and hit[0]:
ref = (hit[0], (hit[1] or '').strip() or None)
if ref is None:
return None
base_id, sku = ref
return base_id, _display_code(row, sku)
def load_material_context(rows): def load_material_context(rows):
""" """
一批审计行 → {row.id: {'sku','name','spec','unit','base_id'}}。 一批审计行 → {row.id: {'base_id','code','name','spec','unit'}}。
只为「能无歧义定位到物料」的行给出结果: 解析线索见 _material_ref_of。**先扫一遍收集线索、再批量查**,
1. 快照里有 base_id(CREATE/DELETE 的库存行)—— 权威,零歧义; 最后逐行套用 —— 全程 5 次查询(SKU 索引 1 次 + 股票表 3 次 +
2. 否则 target_id 唯一命中一张股票表 —— 实测与 (1) 100% 一致。 material_base 1 次),与页大小无关。
定位不到就**不放进去**,由调用方回落到原始 target_name —— 解析不到的行**不放进去**,由调用方回落到原始 target_name ——
宁可不补,也不能补错。 宁可不补,也不能补错。
""" """
# 1) 先从快照取 base_id / sku if not rows:
snap_ctx = {} # row_pk -> (base_id, sku)
target_ids = []
for r in rows:
base_id, sku = None, ''
for which in _SNAPSHOT_BASE_ID_SOURCES:
snap = snapshot_of(r, which)
if not snap:
continue
base_id = as_int(snap.get('base_id')) or base_id
sku = sku or (snap.get('sku') or '')
if base_id:
snap_ctx[r.id] = (base_id, sku)
elif as_int(getattr(r, 'target_id', None)) is not None:
# 2) 没有快照 base_id 的(典型是 UPDATE),留给股票表兜底
target_ids.append(r.target_id)
stock_ctx = _stock_row_lookup(target_ids)
# 汇总所有要查的 base_id,一次查完
need_base = {b for b, _ in snap_ctx.values()}
need_base |= {b for b, _ in stock_ctx.values() if b}
if not need_base:
return {} return {}
# ---------- 第 1 遍:收集线索 ----------
sku_candidates = set()
target_ids = []
for r in rows:
for w in _SNAPSHOT_BASE_ID_SOURCES:
s = snapshot_of(r, w).get('sku')
if s:
sku_candidates.add(str(s).strip())
tname = (r.target_name or '').strip()
if is_sku_like(tname):
sku_candidates.add(tname)
tid = as_int(getattr(r, 'target_id', None))
if tid is not None:
target_ids.append(tid)
# ---------- 批量查 ----------
sku_map = load_sku_base_map(sku_candidates)
stock_map = _stock_row_lookup(target_ids)
# ---------- 第 2 遍:解析 ----------
refs = {}
for r in rows:
ref = _material_ref_of(r, sku_map, stock_map)
if ref:
refs[r.id] = ref
if not refs:
return {}
need_base = {bid for bid, _ in refs.values()}
bases = { bases = {
m.id: m for m in MaterialBase.query.filter(MaterialBase.id.in_(need_base)).all() m.id: m for m in MaterialBase.query.filter(MaterialBase.id.in_(need_base)).all()
} }
out = {} out = {}
for r in rows: for rid, (base_id, code) in refs.items():
ctx = snap_ctx.get(r.id)
if ctx is None:
ctx = stock_ctx.get(as_int(getattr(r, 'target_id', None)))
if not ctx:
continue
base_id, sku = ctx
b = bases.get(base_id) b = bases.get(base_id)
if b is None: if b is None:
continue continue
out[r.id] = { out[rid] = {
'base_id': base_id, 'base_id': base_id,
'sku': (sku or '').strip(), 'code': (code or '').strip(),
'name': (b.name or '').strip(), 'name': (b.name or '').strip(),
'spec': (b.spec_model or '').strip(), 'spec': (b.spec_model or '').strip(),
'unit': (b.unit or '').strip(), 'unit': (b.unit or '').strip(),
@ -449,16 +613,17 @@ def load_material_context(rows):
return out return out
def format_target_display(ctx, fallback=''): def format_target_display(ctx):
""" """
物料上下文 → 「SKU - 物料名称 (规格型号)」。取不到就返回空串(调用方回落)。 物料上下文 → 「编码 - 物料名称 (规格型号)」。取不到就返回空串(调用方回落)。
★ 编码这段逐行不同:物料行走 SKU,BOM 行走 bom_no(见 _material_ref_of)。
★ 缺哪段就省哪段,不留空括号/空横杠:材料名查不到时显示 ★ 缺哪段就省哪段,不留空括号/空横杠:材料名查不到时显示
「0000001180 - 」比只显示「0000001180」更让人困惑。 「0000001180 - 」比只显示「0000001180」更让人困惑。
""" """
if not ctx: if not ctx:
return '' return ''
head = ctx.get('sku') or '' head = ctx.get('code') or ''
name, spec = ctx.get('name') or '', ctx.get('spec') or '' name, spec = ctx.get('name') or '', ctx.get('spec') or ''
if not (head or name): if not (head or name):
return '' return ''
@ -708,8 +873,18 @@ def _fmt_number(val):
def _core_attrs(snap, material=None): def _core_attrs(snap, material=None):
"""快照 → ['入库数量 10 件', '库位 ZZTEST', …],每个槽位取一个。""" """
快照 + 已解析的物料 → 摘要里的核心属性片段,每个槽位取一个。
★ 物料名取自**解析结果**(material['name']),不是快照里的某个字段:
快照里只有 sku('0000001180'),名称在 material_base 上。这类单据
(报废/退回)的操作对象虽然也显示名称,但摘要要能独立读懂 ——
导出成 Excel、或被复制粘贴出来时,旁边那一列不一定在。
"""
parts = [] parts = []
name = (material or {}).get('name') or ''
if name:
parts.append(truncate(name, 24))
for _slot, keys in _SUMMARY_SLOTS: for _slot, keys in _SUMMARY_SLOTS:
for key in keys: for key in keys:
if key not in snap or is_blank(snap[key]): if key not in snap or is_blank(snap[key]):