""" 审计日志 → 可读值 / Excel 的共用层。 为什么单独抽这一层 ------------------ 「把 audit_logs 的一行翻成人话」有两个消费者: · 每日日报的 Excel 附件 —— `daily_report_service.render_excel()` · 审计页的按条件导出 —— `api/v1/audit.py` 的 `GET /audit/logs/export` 两边各写一份的代价不是"多写点代码",而是**结果会不一致**:同一条 「实际审批人ID: 7」,一边翻了人名一边没翻;同一个 status=3,出库模块是 「已完成(已出库)」、报废模块是「已执行(已报废)」,某一边漏了按模块查表就会译错。 **译错的报表比不翻译的更难发现** —— 裸数字至少一看就知道没处理过。 所以这里的边界是:凡是「审计日志 → 人看得懂的东西」都放这里; 上层的两个消费者只决定「取哪些行、列怎么排」。 历史坑(本模块已处理,勿在调用方重新发明) -------------------------------------------- · audit_logs.created_at 是 naive 北京时间(由 beijing_time() 写入); · target_id 在不同模块之间会撞号,定位一条记录必须同时看 module + url; · details 有三种互不相同的结构,键名还不统一: CREATE → {'created': {...}} UPDATE → {'changes': {字段: {old, new}}} DELETE → {'deleted_snapshot': {...}} 记错键名不会报错,只会得到一张空表。 """ import json import re from collections import Counter, defaultdict from sqlalchemy import text from app.extensions import db from app.models.base import MaterialBase from app.utils.audit_labels import ( BOOLEAN_FIELDS, CHANGES_KEY, PERSON_NAME_FIELDS, SNAPSHOT_CREATED, SNAPSHOT_DELETED, USER_ID_FIELDS, action_label, bool_label, canon_action, enum_label, field_label, id_ref_of, is_hidden_credential_field, module_display, person_name_label, ) # 系统占位账号:菜单/权限初始化一类的自动写入挂在这个名下(口径同审计页) SYSTEM_USERNAME = 'system' # 未解析出实体时的占位 UNKNOWN = '-' # details 的键名统一在 audit_labels.py 定义(那里还要下发给前端), # 此处只做引用 —— 两个模块各存一份字符串就是在等着它们漂移。 # 变更字段的噪声过滤。 # # 图片/链接/更新时间的变化与业务动作无关,却会把「变更字段数」顶上去 —— # 看起来"有 3 个字段变了",实际业务上什么都没发生。故在**计数之前**就滤掉。 # 快照(新增/删除)里同样滤掉:那几个字段是超长字符串,进 Excel 会把列宽 # 撑爆且无业务价值。 IGNORED_CHANGE_FIELDS = frozenset({ 'updated_at', 'product_image', 'product_image_remark', 'manual_link', 'manual_link_remark', 'purchase_link', }) # 单元格字符上限。Excel 硬上限是 32767,取 2000 是因为更长的值 # (items_json 整包明细、超长报错文本)在单元格里已无法阅读。 # **截断处补省略号**,不做静默截断。 EXCEL_CELL_LIMIT = 2000 # 快照展开成列的上限。快照是整表全字段,不同表并集理论上可达数百列; # 超过此值按「出现频次」从低到高丢弃,并由调用方显式写明丢了几列。 EXCEL_SNAPSHOT_COL_LIMIT = 80 # 明细表的固定前导列。 # ★ URL 单独成列而不是拼进"描述"里:审计日志的 target_id 在不同模块会撞号, # 定位一条记录必须同时看 module + url。 AUDIT_HEAD = ['时间', '操作人', '模块', '操作类型', '操作对象', '目标ID', 'URL'] # 日报附件的表头。 # # ★ 与 AUDIT_HEAD 有两处**故意的**不同,都是为了不改变既有附件的形态: # 1. 不含「操作类型」—— 日报按操作类型分了表(新增/修改/删除明细各一个 # sheet),每行再重复一遍同一个值是纯冗余:多一列宽度、零信息量。 # 审计页导出是一张混合台账,则必须有这一列。 # 2. 这里叫「目标名称」且排在「目标ID」之前 —— 日报附件一直以来的列序。 # 审计页的 UI 管它叫「操作对象」。**不改它**:既有使用者可能按列位置 # 取数,换序会让数据整体错位一列(Excel 不报错,只是全错)。 DAILY_REPORT_HEAD = ['时间', '操作人', '模块', '目标ID', '目标名称', 'URL'] # 屏幕/表格里的变更值截断长度(纯文本正文用;Excel 用 EXCEL_CELL_LIMIT) TEXT_VALUE_LIMIT = 40 # ============================================================================= # 基础类型转换 # ============================================================================= def is_blank(val): """ 值是否「空」—— 用于过滤无意义的变更与快照字段。 ★ 判据必须**严格**:None / '' / [] / {} / '-' 才算空。 **false 和 0 不算空** —— `is_returned: false`、`quantity: 0` 是明确的 业务事实,用 Python 的真值判断(`if not val`)会把它们一起吃掉, 那是在悄悄篡改业务数据。 """ if val is None: return True if isinstance(val, str): return val.strip() in ('', '-') if isinstance(val, (list, dict)): return len(val) == 0 return False def is_noop_change(old, new): """ 该变更是否「等于没改」。 ★ 典型来源:字段从 NULL 被写成空串(或反之)。库里实测有 98 条记录 带这种变更 —— 写进摘要就是「备注:空→空」,纯噪音,还会把真正有意义 的变更挤出截断长度之外。 """ if is_blank(old) and is_blank(new): return True # 空串与 NULL 在业务上是同一件事,两者之间的"变化"也不算改 return text_of(old) == text_of(new) def as_int(value): """尽力转 int;转不了返回 None(bool 不算数 —— 它是 int 的子类)""" if isinstance(value, bool): return None if isinstance(value, int): return value if isinstance(value, float) and value.is_integer(): return int(value) if isinstance(value, str): s = value.strip() if s.lstrip('-').isdigit(): return int(s) return None # 对象 repr 的识别 —— 形如 ''。 # # ★ 来源:快照收集时(audit_listener._collect_snapshot)本应跳过关系属性, # 但早期版本漏过去了,于是一批历史快照里留下了这种「被 str() 的对象」。 # 它**不是业务数据**:真正有用的值在对应的 `xxx_id` 字段里(parent_id=3030), # 原样显示只会让人以为数据坏了。 # # ★ 判据要**窄**:只认「<类名 空格 内容>」这一种形态,不能见 '<'/'>' 就滤, # 否则会把正常的业务值也误伤(备注里写「<急件>」是很正常的)。 _OBJECT_REPR = re.compile(r'^<[A-Za-z_][\w.]*(\s+[^<>]*)?>$') def is_object_repr(s): """该字符串是否是「被 str() 的 ORM 对象」而非业务数据""" return isinstance(s, str) and bool(_OBJECT_REPR.match(s.strip())) # 对象 repr 的替代显示。明确指向 _id 字段,而不是留空 —— 留空会让人以为 # "这个字段本来就没值",而实际上值在别处。 OBJECT_REPR_TEXT = '(对象引用,见对应 _id 字段)' def text_of(val): """ 任意值 → 文本。 · 空值 → 空串(调用方决定显示成「空」还是留空单元格); · dict / list → 紧凑 JSON —— 直接 str() 会得到 Python repr (单引号 + True/None),那是给开发者看的,不是给人看的。 """ if val is None or val == '': return '' if isinstance(val, (dict, list)): try: return json.dumps(val, ensure_ascii=False, separators=(',', ':')) except (TypeError, ValueError): return str(val) if is_object_repr(val): return OBJECT_REPR_TEXT return str(val) def truncate(text, limit): """截断并补省略号。limit 为 None 或 <=0 时不截断。""" if not limit or limit <= 0 or len(text) <= limit: return text return text[:limit] + '…' def fmt_value(val, limit=TEXT_VALUE_LIMIT): """纯文本正文里的变更值显示:空值统一显示「空」(与审计页详情弹窗一致)。""" s = text_of(val) if not s: return '空' return truncate(s, limit) def cell(val, module=None, key=None, ref_maps=None): """ Excel 单元格取值。 ★ 空值给**空串**而不是「空」:附表是拿来筛选/排序/透视的,混入「空」 这个字面量会让"按旧值排序"把空值排到中间,也让 COUNTIF 失真。 正文那边要「空」是因为纯文本里空着根本看不出有这一项,两处需求相反。 ★ 传 module/key 时会先做 ID→实体的翻译(翻不动则原样显示); 值已由 resolved_changes() 翻过时不要传,避免重复解析。 ★ dict/list 转紧凑 JSON,超长按 EXCEL_CELL_LIMIT 截断并补省略号。 """ if module is not None: resolved = resolve_value(module, key, val, ref_maps or {}) if resolved is not None: val = resolved return truncate(text_of(val), EXCEL_CELL_LIMIT) # ============================================================================= # 操作人 # ============================================================================= def operator_of(row): """ 操作人显示名:显示名(账号),与旧日报的「平板(pingban)」同形。 ★ display_name 库里存法**不统一**,实测有四种: '高雪(gaoxue)' / '高闯/gaochuang' / '杜邢宸(duxingchen)' / ''(空) 直接 f"{display_name}({username})" 会拼出 '杜邢宸(duxingchen)(duxingchen)' 这种账号重复的怪名字。 故先把 display_name 里已内嵌的账号部分剥掉,再统一格式化。 """ dn = (getattr(row, 'display_name', '') or '').strip() un = (getattr(row, 'username', '') or '').strip() # 剥掉内嵌账号:'高雪(gaoxue)'→'高雪','高闯/gaochuang'→'高闯' for sep in ('(', '/'): if sep in dn: dn = dn.split(sep)[0].strip() if dn and un and dn != un: return f"{dn}({un})" return dn or un or UNKNOWN def fmt_name(raw): """ 统一库里五花八门的操作人写法。 audit_logs 侧有 display_name/username 两个字段可以拼,但 trans_outbound.operator_name / trans_borrow.dispatch_operator 这类 **流水表**只存了一个字符串,实测有 '高闯/gaochuang' 与 '杜邢宸(duxingchen)' 两种风格。同一封邮件里两种风格并存会显得很随意, 故统一成「名(账号)」。 """ s = (raw or '').strip() if not s: return UNKNOWN if '/' in s: name, _, acct = s.partition('/') name, acct = name.strip(), acct.strip() if name and acct: return f"{name}({acct})" return name or acct or s return s # ============================================================================= # 引用 ID → 实体的批量解析 # # 逐条查库是 N+1,故先把涉及的 ID 按类型收集起来,每类一次查完。 # ============================================================================= def load_user_map(user_ids): """{user_id: '名(账号)'}""" if not user_ids: return {} from app.models.system import SysUser rows = SysUser.query.filter(SysUser.id.in_(user_ids)).all() return {u.id: fmt_name(u.username) for u in rows} def load_material_name_map(base_ids): """{base_id: 物料名}""" if not base_ids: return {} rows = MaterialBase.query.filter(MaterialBase.id.in_(base_ids)).all() return {m.id: (m.name or '').strip() for m in rows} def load_menu_map(menu_ids): """ {menu_id: 菜单名}。 ★ 0 是 sys_menu.parent_id 的默认值,含义是「没有上级」= 顶级菜单 —— 不能当成"查不到的 ID"而显示成 0。 """ mapping = {0: '顶级菜单'} ids = {i for i in menu_ids if i} if not ids: return mapping from app.models.system import SysMenu for m in SysMenu.query.filter(SysMenu.id.in_(ids)).all(): mapping[m.id] = (m.name or '').strip() or f'菜单#{m.id}' return mapping def load_return_ledger_map(return_ids): """ {trans_return.id: 'SKU 的退回流水(不良品 1.0)'}。 退回流水表本身没有可读单号,能标识它这段记录的就是「哪个物料、什么类型、 多少数量」—— 所以直接把这三样拼出来,而不是显示一个光秃秃的行号。 """ if not return_ids: return {} from app.models.transaction import TransReturn out = {} for r in TransReturn.query.filter(TransReturn.id.in_(return_ids)).all(): qty = float(r.return_qty or 0) rtype = (r.return_type or '').strip() out[r.id] = f"{r.sku or '-'} 的退回流水({rtype} {qty})".replace('( ', '(') return out REF_LOADERS = { 'user': load_user_map, 'material': load_material_name_map, 'menu': load_menu_map, 'return_ledger': load_return_ledger_map, } # 「快照里带 base_id」= 这条记录就是关于某个物料的库存行 _SNAPSHOT_BASE_ID_SOURCES = (SNAPSHOT_CREATED, SNAPSHOT_DELETED) # 库存三表 —— 入库数据分别落在采购/半成品/成品库存表里 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): """ {target_id: (base_id, sku)} —— **仅在 target_id 恰好命中一张股票表**时给出。 ★ 为什么撞号必须放弃(这是本函数存在的全部理由): 实测在库存类记录上,target_id「唯一命中」股票表时,解出的 base_id 与 快照里的 base_id **1230/1230 完全一致**;而一旦命中 2 张表就 24/27 错、 命中 3 张表就 44/44 全错。 补一个**错的**物料名比不补更糟 —— 那是"看起来完全可信的错误答案", 业务方会照着它去找一个根本不相干的物料。 ★ 为什么按 id 去查三张表而不是按 module 定位: 「入库管理」这个 module 名下混着三张表的记录,module 本身分辨不出是哪张; 只能靠 id 命中情况反推,命中多张就是无法判定,直接放弃。 """ ids = set() for t in target_ids: iv = as_int(t) if iv is not None: ids.add(iv) if not ids: return {} from app.models.inbound.buy import StockBuy from app.models.inbound.product import StockProduct from app.models.inbound.semi import StockSemi hits = defaultdict(dict) # stock_id -> {表名: (base_id, sku)} for name, model in (('buy', StockBuy), ('semi', StockSemi), ('product', StockProduct)): for r in model.query.filter(model.id.in_(ids)).all(): hits[r.id][name] = (getattr(r, 'base_id', None), getattr(r, 'sku', '') or '') return {sid: next(iter(tbls.values())) 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) # BOM 是唯一一张「父件—子件」关系表:一行 = 父件用 N 个某子件。 # # ★ 为什么必须按 module 判定,不能靠「target_id 命中 bom_table」这个特征: # 后者实测**大量误命中** —— 入库管理 3655 条、系统管理的 /permissions/assign # 1415 条、image_embeddings 1187 条、出库管理 726 条,它们的 target_id 都会 # 撞上某条 bom_table 记录。照着补子件名就是把毫不相干的零件安到别的记录上。 # # 而 module 是由**表名**推得的(audit_listener._get_module_name: # bom_table → 'BOM管理'),按 module 判定等价于按表判定,是可靠的。 _BOM_MODULE = 'BOM管理' def _child_id_from_snapshot(row): """CREATE/DELETE 快照里的 child_id;UPDATE 没有快照,返回 None""" for which in (SNAPSHOT_CREATED, SNAPSHOT_DELETED): cid = as_int(snapshot_of(row, which).get('child_id')) if cid: return cid return None def load_child_lookup(rows): """ 一批审计行 → {row.id: {'id','name','dosage'}} —— 父子关系里的**子件**。 父子关系记录(BOM)批量插入时,同一秒会有几十条 target_name 完全相同的 日志,光看父件分不出哪条对应哪个子件。这里把子件解出来供摘要差异化。 两条路径: 1. 快照里的 child_id —— CREATE/DELETE(实测 BOM CREATE 覆盖率 100%) 2. 否则若 module 是 BOM,用 target_id 反查 bom_table —— UPDATE (实测 584/634 可解) ★ 子件名称**批量**解析(一次 MaterialBase 查询),不逐行查库: BOM 一次插入几十条,逐行反查就是 N+1。 ★ 用量只在快照里有;UPDATE 行拿不到就不显示 —— 宁可不显示,也不去猜。 """ if not rows: return {} picked = {} # row_pk -> (child_id, dosage) need_lookup = [] # 无快照 child_id 的 BOM 行,留给 bom_table 兜底 for r in rows: if (r.module or '') != _BOM_MODULE: continue cid = _child_id_from_snapshot(r) if cid: dosage = None for which in (SNAPSHOT_CREATED, SNAPSHOT_DELETED): snap = snapshot_of(r, which) if snap.get('dosage') is not None: dosage = snap.get('dosage') break picked[r.id] = (cid, dosage) else: tid = as_int(getattr(r, 'target_id', None)) if tid is not None: need_lookup.append((r.id, tid)) if need_lookup: from app.models.bom import BomTable ids = {tid for _, tid in need_lookup} found = { b.id: b.child_id for b in BomTable.query.filter(BomTable.id.in_(ids)).all() } for rid, tid in need_lookup: cid = as_int(found.get(tid)) if cid: picked[rid] = (cid, None) if not picked: return {} child_ids = {cid for cid, _ in picked.values()} names = load_material_name_map(child_ids) return { rid: {'id': cid, 'name': names.get(cid, ''), 'dosage': dosage} for rid, (cid, dosage) in picked.items() } def load_material_context(rows): """ 一批审计行 → {row.id: {'base_id','code','name','spec','unit'}}。 解析线索见 _material_ref_of。**先扫一遍收集线索、再批量查**, 最后逐行套用 —— 全程 5 次查询(SKU 索引 1 次 + 股票表 3 次 + material_base 1 次),与页大小无关。 解析不到的行**不放进去**,由调用方回落到原始 target_name —— 宁可不补,也不能补错。 """ if not rows: 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 = { m.id: m for m in MaterialBase.query.filter(MaterialBase.id.in_(need_base)).all() } out = {} for rid, (base_id, code) in refs.items(): b = bases.get(base_id) if b is None: continue out[rid] = { 'base_id': base_id, 'code': (code or '').strip(), 'name': (b.name or '').strip(), 'spec': (b.spec_model or '').strip(), 'unit': (b.unit or '').strip(), } return out def format_target_display(ctx): """ 物料上下文 → 「编码 - 物料名称 (规格型号)」。取不到就返回空串(调用方回落)。 ★ 编码这段逐行不同:物料行走 SKU,BOM 行走 bom_no(见 _material_ref_of)。 ★ 缺哪段就省哪段,不留空括号/空横杠:材料名查不到时显示 「0000001180 - 」比只显示「0000001180」更让人困惑。 """ if not ctx: return '' head = ctx.get('code') or '' name, spec = ctx.get('name') or '', ctx.get('spec') or '' if not (head or name): return '' tail = f"{name} ({spec})" if name and spec else (name or '') if head and tail: return f"{head} - {tail}" return head or tail def _collect_ref(ref_ids, user_ids, module, key, value): """把一个候选值按类型塞进对应的收集集合(只收集,不查库)""" iv = as_int(value) if iv is None: return if key in USER_ID_FIELDS: user_ids.add(iv) return ref = id_ref_of(module, key) if ref: ref_ids[ref].add(iv) def load_ref_maps(rows): """ 一批 audit_logs 行 → {ref 类型: {id: 显示名}}。 ★ 覆盖三种结构的**全部**引用列,不只是 UPDATE 的变更值: 新增/删除的整字段快照里同样有 base_id / parent_id 这类引用列 (BOM 的整表快照尤其多)。只收集 UPDATE 的话,附表里就会冒出 「父级ID: 1234」这种裸数字。 """ user_ids = set() ref_ids = defaultdict(set) for r in rows: module = getattr(r, 'module', None) for key, old, new in changes_of(r): _collect_ref(ref_ids, user_ids, module, key, old) _collect_ref(ref_ids, user_ids, module, key, new) for which in (SNAPSHOT_CREATED, SNAPSHOT_DELETED): for key, value in snapshot_of(r, which).items(): _collect_ref(ref_ids, user_ids, module, key, value) ref_maps = {'user': load_user_map(user_ids)} for ref, ids in ref_ids.items(): loader = REF_LOADERS.get(ref) if loader: ref_maps[ref] = loader(ids) return ref_maps # ============================================================================= # details 解析 # ============================================================================= def changes_of(row): """ UPDATE 记录的字段变更 → [(字段, 旧值, 新值)];结构异常时返回空列表。 噪声字段在这里就滤掉,理由见 IGNORED_CHANGE_FIELDS 的说明。 """ details = row.details or {} if not isinstance(details, dict): return [] changes = details.get(CHANGES_KEY) if not isinstance(changes, dict): return [] out = [] for key, val in changes.items(): if key in IGNORED_CHANGE_FIELDS: continue # ★ 凭据类字段一律不出现在任何出口(详情、摘要、Excel 导出)。 # 在这里滤是源头做法 —— 逐个出口去补必然漏掉某一个, # 而漏掉的那个就是泄漏点。 if is_hidden_credential_field(key): continue if isinstance(val, dict): old, new = val.get('old'), val.get('new') else: # 兼容「非 {old,new} 结构」的历史写法:整值视为新值 old, new = None, val # ★ 等于没改的(NULL ↔ 空串)在这里就丢掉:它会污染摘要(「备注:空→空」) # 并把真正有意义的变更挤出截断长度;在 Excel 里也白占一行。 if is_noop_change(old, new): continue out.append((key, old, new)) return out def snapshot_of_any(row): """ 该行的业务快照 + 它的展示标题,按 SNAPSHOT_VIEWS 的优先级取第一个可用的。 返回 (title, data) 或 (None, None)。data 可能是 dict(常规)也可能是 list(`payload` 实测有整包就是数组的),由调用方按类型处置。 """ from app.utils.audit_labels import SNAPSHOT_VIEWS details = getattr(row, 'details', None) if not isinstance(details, dict): return None, None for key, title in SNAPSHOT_VIEWS: data = details.get(key) if data: return title, data return None, None def snapshot_of(row, which): """ CREATE / DELETE 记录的整字段快照 → {字段: 值};结构异常时返回 {}。 which 取 SNAPSHOT_CREATED / SNAPSHOT_DELETED —— 两张表用了不同的键名, 写错不会报错,只会得到一张空表,故集中在此处而非散落到调用点。 快照里同样过一遍 IGNORED_CHANGE_FIELDS(理由同 changes_of)。 """ details = row.details or {} if not isinstance(details, dict): return {} snap = details.get(which) if not isinstance(snap, dict): return {} return { k: v for k, v in snap.items() if k not in IGNORED_CHANGE_FIELDS # 凭据类同样在源头滤掉(理由见 changes_of) and not is_hidden_credential_field(k) } def sanitize_details(details): """ details(含嵌套快照与数组)→ 剔除凭据类字段后的**新**结构。 ★★ 必须在这层做,不能只靠前端隐藏:前端隐藏只是"不显示",原始响应里 照样有明文 —— 任何人打开 devtools 就读得到。 **凡是要挡的数据,都要在接口出口剥掉。** 前端那份只是为了不把 DOM 也渲染出来(双保险),不是防线本身。 ★ 实测来源:`用户管理/新增` 的 payload 快照里存着明文密码(27 条), 写入路径把请求体整包记进了审计 —— 请求体里是哈希前的原始密码。 ★ 递归进 dict 与 list:payload 里嵌着 items 数组,数组元素对象上同样 可能有凭据字段,只看第一层会漏。 ★ 返回新对象,**不原地改** ORM 对象持有的 dict —— 那会污染 session 里的实体,同一个请求后续若还要用到原始数据就会拿到被改过的版本。 """ if isinstance(details, dict): return { k: sanitize_details(v) for k, v in details.items() if not is_hidden_credential_field(k) } if isinstance(details, list): return [sanitize_details(v) for v in details] return details def has_detail_content(details): """details 里是否有可展示的结构(与前端 hasDetailContent 同判据)""" if not isinstance(details, dict) or not details: return False # 键名取自 SNAPSHOT_VIEWS + changes —— 与前端渲染用的是同一份, # 不在这里另列一遍(漏一个就会出现"有详情按钮但点开是空的") from app.utils.audit_labels import SNAPSHOT_VIEWS keys = [k for k, _ in SNAPSHOT_VIEWS] + [CHANGES_KEY] return any(details.get(k) for k in keys) # ============================================================================= # 取值翻译 # ============================================================================= def resolve_value(module, field, value, ref_maps): """ 把单个变更值翻成人类可读形式;翻不动返回 None(调用方回落原值)。 四类映射,优先级即从上到下: · 布尔列 → 是 / 否 · 人名串 → '杜邢宸/duxingchen' → '杜邢宸(duxingchen)' · 引用 ID 的列 → 查实体(用户 / 物料 / 菜单 / 退回流水…, 指向哪张表由 audit_labels.id_ref_of 按模块判定) · 枚举列 → 按模块查状态表(见 audit_labels.enum_label) ★ 翻不动就**原样显示**,绝不硬编一个可能错的中文 —— 报表里的错译 比裸数字更难被发现,也更危险。 """ if field in BOOLEAN_FIELDS: text = bool_label(value) if text is not None: return text # 人名串('杜邢宸/duxingchen')→ '杜邢宸(duxingchen)'。 # 按字段名限定,理由见 audit_labels.PERSON_NAME_FIELDS 的说明。 if field in PERSON_NAME_FIELDS: text = person_name_label(value) if text is not None: return text iv = as_int(value) if iv is not None: if field in USER_ID_FIELDS: return (ref_maps.get('user') or {}).get(iv) ref = id_ref_of(module, field) if ref: got = (ref_maps.get(ref) or {}).get(iv) if got is not None: return got return enum_label(module, field, value) def resolved_changes(module, changes, ref_maps): """ 一组字段变更 → [(中文标签, 旧值, 新值)]。 ★ 只做「翻译」(ID→名称),**不做「格式化」**(截断、空值占位)—— 正文与 Excel 对格式化的要求正好相反:正文要短(截断 40 字、空值显示 「空」),附表要全(不截断、空值留空便于筛选排序)。 翻译逻辑才是必须共用的部分,各写一份迟早漂移成 "邮件里是人名、附表里是裸 ID"。 ★ 「实际审批人ID: 空 → 7」解析成人名后,标签里的「ID」后缀就不贴切了, 故解析成功时去掉后缀 → 「实际审批人: 空 → 杜邢宸(duxingchen)」。 """ out = [] for key, old, new in changes: old_t = resolve_value(module, key, old, ref_maps) new_t = resolve_value(module, key, new, ref_maps) label = field_label(key) is_ref = key in USER_ID_FIELDS or id_ref_of(module, key) is not None if is_ref and (old_t is not None or new_t is not None) and label.endswith('ID'): label = label[:-2] out.append(( label, old if old_t is None else old_t, new if new_t is None else new_t, )) return out # 请求 URL 的 /api 或 /api/vN 前缀 —— 对用户零信息量,还占摘要的字符预算 _API_VERSION_RE = re.compile(r'^/?api(?:/v\d+)?/?', re.IGNORECASE) def url_source(url): """ 请求 URL → 简短的「触发来源」。去掉 /api/vN 前缀,保留剩余路径。 ★ 这个信号解决的是一个**具体的误解**:业务方看到「入库/库存管理」里有 大量非库管人员的 UPDATE,以为是越权改库存 —— 实际是出库、借还、盘点 等单据流转触发的自动扣减。实测该类 UPDATE 的来源分布: 1582 /api/v1/outbound 出库触发扣减 424 /api/v1/outbound/request 108 /api/v1/inbound/stock/draft/add 102 /api/v1/transactions/borrow/dispatch 借出 95 /api/v1/inbound/stock/stocktake/update-quantity 光看"谁改的"永远解释不清,必须能看出"哪个流程触发的"。 ★ 去掉版本前缀是因为它零信息量;剩下的路径段正是要的东西。 长度封顶 40 字,避免个别超长 URL 把摘要挤没。 """ s = (url or '').strip() if not s: return '' return truncate(_API_VERSION_RE.sub('', s).strip('/'), 40) # 摘要核心属性按**槽位**取:每个槽位只取第一个有值的字段。 # # ★ 为什么不是简单按字段列表顺序取前 3 个:库存快照里 in_quantity / # stock_quantity / available_quantity 三个数量字段的值往往完全相同, # 取前 3 个会得到「入库数量 10 件、总库存 10 件、可用库存 10 件」—— # 三个数字一样,纯冗余,还把库位这类更该看到的信息挤掉了。 # 按槽位(数量 / 位置 / 物料)各取一个,信息不重复。 # # ★ 为什么不再罗列字段名:实测一条入库快照有 29 个字段,罗列成 # 「新增:SKU、base、状态、…等 29 个字段」信息量为零 —— 用户看完还是不知道 # "到底新增了什么"。业务方真正想知道的是"多少数量、放在哪"。 # ★ 只有两个槽位:数量与位置。**不含物料** —— 物料由「操作对象」列承担 # (那一列已经是「SKU - 名称 (规格)」),摘要里再来一遍「SKU 0000002270」 # 是同一屏内的重复,白白吃掉截断长度。用户要的也正是「10 件 (仓库: ZZTEST)」 # 这种只讲数量和位置的形式。 _SUMMARY_SLOTS = ( ('数量', ('in_quantity', 'quantity', 'stock_quantity', 'available_quantity', 'out_quantity', 'return_qty', 'dosage')), ('位置', ('warehouse_location', 'warehouse_loc', 'location', 'return_location')), ) # 数量类字段 —— 摘要里带上单位(单位来自物料主数据,查不到就不带) _QUANTITY_FIELDS = frozenset(_SUMMARY_SLOTS[0][1]) def _fmt_number(val): """数字去掉无意义的 .0(10.0 → 10);非数字原样返回。""" if isinstance(val, bool): return str(val) if isinstance(val, float) and val.is_integer(): return str(int(val)) return text_of(val) def _core_attrs(snap, material=None): """ 快照 + 已解析的物料 → 摘要里的核心属性片段,每个槽位取一个。 ★ 物料名取自**解析结果**(material['name']),不是快照里的某个字段: 快照里只有 sku('0000001180'),名称在 material_base 上。这类单据 (报废/退回)的操作对象虽然也显示名称,但摘要要能独立读懂 —— 导出成 Excel、或被复制粘贴出来时,旁边那一列不一定在。 """ parts = [] name = (material or {}).get('name') or '' if name: parts.append(truncate(name, 24)) for _slot, keys in _SUMMARY_SLOTS: for key in keys: if key not in snap or is_blank(snap[key]): continue text = truncate(_fmt_number(snap[key]), 30) # ★ 单位只在物料主数据里确实有 unit 时才带 —— 不猜。 # 但**纯数字的单位要丢掉**:库里有一批 unit 落成了 '1', # 显示成「入库数量 48 1」比不带单位还难读。 unit = (material or {}).get('unit') or '' if key in _QUANTITY_FIELDS and unit and not unit.isdigit(): text = f"{text} {unit}" parts.append(f"{field_label(key)} {text}") break # 该槽位已取到,不再看后面的候选字段 return parts def _child_text(child): """子件的展示名:优先物资名,退化成 #ID;两者都没有返回空串""" if not child: return '' return (child.get('name') or '').strip() or ( f"#{child['id']}" if child.get('id') else '') def changes_summary(row, ref_maps, limit=200, resolved=None, material=None, child=None): """ 一行日志的变更摘要(一句话),供台账 sheet 用。 台账里只给一条能扫读的摘要,完整逐字段对比在「变更明细」sheet —— 把人话摘要和可筛选的数据分开,两者各司其职。 resolved: 已算好的 resolved_changes 结果。导出时同一行的翻译要同时喂给 台账摘要和明细长表,调用方预计算一次传进来即可 —— 否则每行 要跑两遍 resolve_value(4 万行的导出上是可感知的浪费)。 """ if canon_action(row.action) == 'UPDATE': if resolved is None: resolved = resolved_changes(row.module, changes_of(row), ref_maps) if not resolved: return '(仅变更了图片/链接/更新时间等噪声字段)' # ★ 分隔符统一用全角(:和;)—— 摘要是一句中文,半角冒号混在全角分号 # 里很难看,也和日报正文的排版不一致。 parts = [f"{label}:{fmt_value(old)}→{fmt_value(new)}" for label, old, new in resolved] body = ';'.join(parts) # ★ BOM 的 UPDATE 行必须带出子件:一次批量归档会同时改几十行, # 光看"是否启用 是→否"完全分不出改的是哪个子件(实测一种摘要 # 重复 135 次)。子件名放在最前面,扫一眼就能区分。 ct = _child_text(child) if ct: body = f"子件 {ct}:{body}" # ★ 来源**拼在截断之后**:先截变更内容、再追加来源,保证来源这条关键 # 信息永远不会被截掉。反过来(整个串一起截)会把来源切没, # 而"哪触发的"正是这条摘要存在的理由。 body = truncate(body, limit) source = url_source(row.url) return f"{body}(来源:/{source})" if source else body is_create = canon_action(row.action) == 'CREATE' verb = '新增' if is_create else '删除' mod = module_display(row.module) prefix = f"{verb}({mod}):" if mod else f"{verb}:" # ★ 用 snapshot_of_any:`payload` 是与 `created` 平级的第三种快照存法 # (借库/出库/采购入库走这条),只看 created/deleted_snapshot 会让 # 这些记录的新增摘要整列为空。 _title, snap = snapshot_of_any(row) if not isinstance(snap, dict): return '' # ★ 父子关系记录(BOM)优先走子件分支:一次新建会插入几十条, # 子件不同但父件相同,不把子件写出来就是几十行一模一样的日志。 # 实测同一秒同一 bom_no 最多 58 条。 ct = _child_text(child) if ct: body = f"{'添加' if is_create else '移除'}子件 {ct}" if (child or {}).get('dosage') is not None: body += f"(用量 {_fmt_number(child['dosage'])})" return truncate(prefix + body, limit) # 优先:核心属性的**值**(多少数量、放哪)—— 这才是用户想知道的 core = _core_attrs(snap, material) if core: return truncate(prefix + '、'.join(core), limit) # 回落:没有核心属性可提(如 BOM 记录、payload 型的借库单), # 退化成字段名清单。跳过 id(主键每条都有、零信息量)与凭据类字段 # (摘要里出现「password」这种字样本身就是多余的暴露)。 keys = [k for k in snap if k != 'id' and not is_hidden_credential_field(k)] names = [] for k in keys[:8]: v = snap[k] # 数组字段带上条数 —— 只写「物料明细」看不出有几行, # 「物料明细 1 项」才是有用的信息 names.append(f"{field_label(k)} {len(v)} 项" if isinstance(v, list) else field_label(k)) more = '' if len(keys) <= 8 else f" 等 {len(keys)} 个字段" return truncate(f"{prefix}{'、'.join(names)}{more}", limit) # ============================================================================= # Excel 排版 # ============================================================================= def display_width(text): """ 估算字符串的显示宽度:CJK 字符按 2 列、其余按 1 列。 ★ 直接拿 len() 当列宽,中文列会明显偏窄(一个汉字占两个字符位), 「物料名称」这类列会被截成「物料名…」。 """ return sum(2 if ord(ch) > 0x2E80 else 1 for ch in str(text)) def snapshot_headers(cols, reserved=()): """ 快照字段 → 表头。**重名的补原字段名**。 ★ 重名有两个来源,必须一起处理: 1. FIELD_LABELS 里多个字段映射到同一中文名 —— is_active / is_enabled → 「是否启用」 warehouse_location / warehouse_loc → 「库位」 operator_name / operator → 「操作人」 2. 快照字段与**固定前导列**撞名 —— BOM 快照里就有 operator 字段, 中文名同样是「操作人」,与 AUDIT_HEAD 的第 2 列撞上。 (实测 2026-07-16 的 59 列新增明细正是被这一条卡住。) Excel 不会因为表头重名报错,但人分不清哪列是哪列, 且按名字取列(VLOOKUP / 脚本)会静默取到第一个 —— 故重名时 退化成「操作人(operator)」。 """ labels = [field_label(c) for c in cols] counted = Counter(labels) + Counter(reserved) return [ f"{lab}({col})" if counted[lab] > 1 else lab for lab, col in zip(labels, cols) ] def fill_sheet(ws, headers, rows, wrap_cols=()): """ 写入表头 + 数据行,套用统一样式。 样式与 export_service/excel_task.py 的库存导出保持一致(深蓝表头、 隔行浅蓝、细边框),这样公司内部几个导出的观感统一。 wrap_cols: 需自动换行的列序号(从 1 起)。长文本列(旧值/新值/URL) 不换行会被右侧有值的单元格**盖住**,看起来像"没有这一列"。 """ from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter header_fill = PatternFill("solid", fgColor="1F4E79") header_font = Font(bold=True, color="FFFFFF", size=11) header_align = Alignment(horizontal='center', vertical='center', wrap_text=True) thin = Side(style='thin', color='BFBFBF') border = Border(left=thin, right=thin, top=thin, bottom=thin) data_font = Font(size=10) even_fill = PatternFill("solid", fgColor="DEEAF1") ws.append(list(headers)) for idx in range(1, len(headers) + 1): c = ws.cell(row=1, column=idx) c.fill = header_fill c.font = header_font c.alignment = header_align c.border = border # 表头行加高,否则带换行的中文表头会被压成一条缝 ws.row_dimensions[1].height = 28 for i, row in enumerate(rows): ws.append(list(row)) r = i + 2 for ci in range(1, len(headers) + 1): c = ws.cell(row=r, column=ci) c.font = data_font c.border = border c.alignment = Alignment(horizontal='left', vertical='top', wrap_text=(ci in wrap_cols)) if i % 2 == 1: c.fill = even_fill # 冻结首行:几千行明细里往下滚,没有表头就不知道在看哪一列 ws.freeze_panes = 'A2' # 列宽按内容估算,封顶 60 —— 不封顶的话一个超长 URL 会把该列拉到屏幕外, # 其他列全被挤没。只采样前 200 行,避免大数据量下空转。 for ci, head in enumerate(headers, start=1): width = display_width(head) for row in rows[:200]: width = max(width, display_width(row[ci - 1])) ws.column_dimensions[get_column_letter(ci)].width = min(width + 2, 60) def _timestamp(row): return row.created_at.strftime('%Y-%m-%d %H:%M:%S') if row.created_at else '' def audit_head_values(row, display=None): """ 审计页导出的前导列取值(与 AUDIT_HEAD 一一对应)。 display: 补全后的操作对象(「SKU - 名称 (规格)」)。不传则用库里的 target_name —— 后者常常只是 SKU 或 'stock_buy ID:1667'。 """ return [ _timestamp(row), operator_of(row), row.module or '', action_label(canon_action(row.action)), display or row.target_name or '', row.target_id or '', row.url or '', ] def daily_report_head_values(row): """日报附件的前导列取值(与 DAILY_REPORT_HEAD 一一对应,列序不同,见其说明)""" return [ _timestamp(row), operator_of(row), row.module or '', row.target_id or '', row.target_name or '', row.url or '', ] # 台账 sheet 的列 = 固定前导列 + 变更摘要 + IP LEDGER_EXTRA = ['变更摘要', 'IP地址'] def build_audit_workbook(rows, ref_maps=None, summary_rows=None, note=None): """ 一批 audit_logs 行 → .xlsx 字节流(内存中构建,不落盘)。 工作表: 汇总 —— 调用方给出的筛选条件 + 本次导出的口径说明 审计日志 —— 一行一条记录(台账) 变更明细 —— 一行一个变更字段(仅 UPDATE),便于按字段筛选透视 ★ 为什么用长表放变更:宽表要取所有记录的变更字段并集,而不同模块改的 字段完全不同,并集轻松上百列且绝大多数格子是空的。长表列固定 8 个, 想筛"今天所有改过状态的记录"只需对「字段」列做一次筛选。 ★ 全程 BytesIO,不写临时文件 —— 导出是随时可能发生的按需操作, 落盘就意味着要么写清理逻辑,要么让磁盘被日复一日地吃掉。 note: 截断等需要显式告知的说明(None 表示无)。**不静默截断**是本项目 报表的一贯要求。 """ import io from openpyxl import Workbook if ref_maps is None: ref_maps = load_ref_maps(rows) materials = load_material_context(rows) children = load_child_lookup(rows) # ---------- 审计日志(台账)---------- ledger = [] detail = [] for r in rows: # UPDATE 行的翻译只算一次,同时喂给台账摘要与明细长表(见 changes_summary) changes = (resolved_changes(r.module, changes_of(r), ref_maps) if canon_action(r.action) == 'UPDATE' else None) mat = materials.get(r.id) display = format_target_display(mat) or None ledger.append( audit_head_values(r, display=display) + [changes_summary(r, ref_maps, resolved=changes, material=mat, child=children.get(r.id)), r.ip_address or ''] ) if changes is None: continue head6 = audit_head_values(r, display=display)[:6] if not changes: # ★ 改的全是噪声字段的记录不能凭空消失 —— 它确实发生过。 detail.append( head6 + ['(仅变更了图片/链接/更新时间等噪声字段)', '', ''] ) continue for label, old, new in changes: detail.append(head6 + [label, cell(old), cell(new)]) # ---------- 汇总 ---------- summary = list(summary_rows or []) summary.append(('本次导出', '记录数', len(rows))) summary.append(('本次导出', '变更明细行数', len(detail))) if note: summary.append(('本次导出', '注意', note)) wb = Workbook() ws = wb.active ws.title = '汇总' fill_sheet(ws, ['分类', '项目', '数值/说明'], summary, wrap_cols=(3,)) fill_sheet( wb.create_sheet('审计日志'), AUDIT_HEAD + LEDGER_EXTRA, ledger, wrap_cols=(7, 8), # URL 与变更摘要 ) fill_sheet( wb.create_sheet('变更明细'), AUDIT_HEAD[:6] + ['字段', '旧值', '新值'], detail, wrap_cols=(7, 8, 9), ) buf = io.BytesIO() wb.save(buf) return buf.getvalue()