6a9e41f53c
后端：GET /audit/modules 与 /logs 的 modules 改为下发最终选项 [{value,label}]
· 历史命名（入库管理/采购入库/成品入库/库存管理）折叠进「入库」一项且
**不再单独列出**，下拉里看不到历史包袱。展开仍只由后端 expand_modules
负责 —— 前端若自己再拼一份成员表，将来后端加成员会静默失效（本项目
已经踩过两次这种漂移）。
· 英文历史 module（image_embeddings 2719 条 / purchase_request 75 条 /
sys_element 6 条）在这里翻成中文 label，value 保留英文原值否则筛选
匹配不上。映射表 MODULE_LABELS 从**前端** AuditLog.vue 的 moduleMap
收进 audit_labels.py —— 那份硬编副本历史上已经漂移过一次。
· /audit/labels 新增 moduleDisplay {原始值 → 展示名}，同时覆盖聚合成员与
英文历史值：只覆盖后者会出现「下拉显示『入库』、表格显示『库存管理』」，
用户会以为筛选没生效。
后端：新增「触发来源」
· 摘要追加 （来源：/outbound/request） 形式的 URL 尾段（去掉 /api/vN 前缀）。
解决一个具体误解：业务方看到「入库/库存管理」里有大量非库管人员的
UPDATE，以为越权改库存，实际是出库/借还/盘点等单据流转触发的自动扣减。
实测该类 UPDATE 的来源：outbound 1582、outbound/request 424、
borrow/dispatch 102、stocktake/update-quantity 95 —— 光看"谁改的"永远
解释不清，必须能看出"哪个流程触发的"。
· 来源拼在**截断之后**，保证这条关键信息永远不会被截掉。
· 列表与导出共用 changes_summary，故 Excel 台账同样带来源。
后端：其余
· 新增 target_keyword（ilike 同时匹配 target_id / target_name），
放进共用的 _build_audit_query，导出自动支持。
· action 在响应里归一化为 CREATE/UPDATE/DELETE（复用 audit_labels
.canon_action，覆盖批量删除/分配/归还等全部历史别名）。库里还有 1573 条
中文 action，归一后前端不必再为每种历史写法兜底。
· 新增 summary 字段。不算在模型 @property 上：摘要要把 user_id/base_id
翻成人名/物料名，需要查库，模型属性里发查询就是 N+1；改为整页一次
load_ref_maps 批量解析。
· 过滤对象 repr 脏值（<MaterialBase 3030>）。快照收集早期漏跳关系属性，
库里留了 1529 条这种值 —— 不是业务数据，真正的值在对应 _id 字段里。
判据收窄到「<类名 空格 内容>」，避免误伤备注里的「<急件>」。
前端 AuditLog.vue
· 模块下拉改平铺 [{value,label}]，删除硬编的 moduleMap
· 新增「操作对象」模糊搜索
· 合并「操作人」「姓名」两列（判据 username==='system' —— 响应里没有
operator_type 字段，那是请求参数名，照搬会永远不显示系统标签）
· 新增「变更摘要」列；操作对象显示为「名称 #id」
· 操作时间改为恒显示北京时间，不再按浏览器时区换算，与数据库/日报/导出一致
· 详情弹窗改 el-drawer，按 action 分支渲染：UPDATE 出对比表，
CREATE/DELETE 出快照属性表，并跳过对象 repr
· 抽屉顶部高亮展示触发来源（METHOD + URL 告警条）
验证：后端 18 + 22 + 10 项断言全过（模块选项无历史名且「入库」=四值之和、
英文 value 仍可筛选、url_source 边界、来源不被截断、导出口径与列表一致、
权限 401/403/200 矩阵）；vue-tsc --noEmit 与 vite build 均 exit=0。
日报回归：三天附件 490.5K/193.5K/10.3K，6 列结构与改动前一致。
761 lines
29 KiB
Python
761 lines
29 KiB
Python
"""
|
||
审计日志 → 可读值 / 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 app.models.base import MaterialBase
|
||
from app.utils.audit_labels import (
|
||
BOOLEAN_FIELDS,
|
||
PERSON_NAME_FIELDS,
|
||
USER_ID_FIELDS,
|
||
action_label,
|
||
bool_label,
|
||
canon_action,
|
||
enum_label,
|
||
field_label,
|
||
id_ref_of,
|
||
person_name_label,
|
||
)
|
||
|
||
# 系统占位账号：菜单/权限初始化一类的自动写入挂在这个名下（口径同审计页）
|
||
SYSTEM_USERNAME = 'system'
|
||
|
||
# 未解析出实体时的占位
|
||
UNKNOWN = '-'
|
||
|
||
# details 里的三种结构各自的键名 —— 集中在此，避免调用方记错
|
||
SNAPSHOT_CREATED = 'created'
|
||
SNAPSHOT_DELETED = 'deleted_snapshot'
|
||
CHANGES_KEY = 'changes'
|
||
|
||
# 变更字段的噪声过滤。
|
||
#
|
||
# 图片/链接/更新时间的变化与业务动作无关，却会把「变更字段数」顶上去 ——
|
||
# 看起来"有 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 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 的识别 —— 形如 '<MaterialBase 3030>'。
|
||
#
|
||
# ★ 来源：快照收集时（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,
|
||
}
|
||
|
||
|
||
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
|
||
if isinstance(val, dict):
|
||
out.append((key, val.get('old'), val.get('new')))
|
||
else:
|
||
# 兼容「非 {old,new} 结构」的历史写法：整值视为新值
|
||
out.append((key, None, val))
|
||
return out
|
||
|
||
|
||
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}
|
||
|
||
|
||
def has_detail_content(details):
|
||
"""details 里是否有可展示的结构（与前端 hasDetailContent 同判据）"""
|
||
if not isinstance(details, dict) or not details:
|
||
return False
|
||
return any(
|
||
details.get(k) for k in (CHANGES_KEY, SNAPSHOT_DELETED, SNAPSHOT_CREATED, 'payload')
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# 取值翻译
|
||
# =============================================================================
|
||
|
||
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)
|
||
|
||
|
||
def changes_summary(row, ref_maps, limit=200, resolved=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 = truncate('；'.join(parts), limit)
|
||
source = url_source(row.url)
|
||
return f"{body}（来源：/{source}）" if source else body
|
||
|
||
for which, verb in ((SNAPSHOT_CREATED, '新增'), (SNAPSHOT_DELETED, '删除前快照')):
|
||
snap = snapshot_of(row, which)
|
||
if snap:
|
||
# ★ 跳过 id：它是主键、每条都有，排在最前面却零信息量，
|
||
# 白白挤掉一个真正有内容的字段（摘要只列前 8 个）。
|
||
keys = [k for k in snap if k != 'id']
|
||
names = [field_label(k) for k in keys[:8]]
|
||
more = '' if len(keys) <= 8 else f" 等 {len(keys)} 个字段"
|
||
return truncate(f"{verb}：{'、'.join(names)}{more}", limit)
|
||
return ''
|
||
|
||
|
||
# =============================================================================
|
||
# 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):
|
||
"""审计页导出的前导列取值（与 AUDIT_HEAD 一一对应）"""
|
||
return [
|
||
_timestamp(row),
|
||
operator_of(row),
|
||
row.module or '',
|
||
action_label(canon_action(row.action)),
|
||
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)
|
||
|
||
# ---------- 审计日志（台账）----------
|
||
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)
|
||
ledger.append(
|
||
audit_head_values(r)
|
||
+ [changes_summary(r, ref_maps, resolved=changes), r.ip_address or '']
|
||
)
|
||
if changes is None:
|
||
continue
|
||
if not changes:
|
||
# ★ 改的全是噪声字段的记录不能凭空消失 —— 它确实发生过。
|
||
detail.append(
|
||
audit_head_values(r)[:6]
|
||
+ ['（仅变更了图片/链接/更新时间等噪声字段）', '', '']
|
||
)
|
||
continue
|
||
for label, old, new in changes:
|
||
detail.append(
|
||
audit_head_values(r)[:6]
|
||
+ [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()
|