KCGL/inventory-backend/app/services/audit_export_service.py

"""
审计日志 → 可读值 / 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,
    module_display,
    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 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 的识别 —— 形如 '<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,
}


# 「快照里带 base_id」= 这条记录就是关于某个物料的库存行
_SNAPSHOT_BASE_ID_SOURCES = (SNAPSHOT_CREATED, SNAPSHOT_DELETED)

# 库存三表 —— 入库数据分别落在采购/半成品/成品库存表里
STOCK_TABLES = ('stock_buy', 'stock_semi', 'stock_product')


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 load_material_context(rows):
    """
    一批审计行 → {row.id: {'sku','name','spec','unit','base_id'}}。

    只为「能无歧义定位到物料」的行给出结果：
      1. 快照里有 base_id（CREATE/DELETE 的库存行）—— 权威，零歧义；
      2. 否则 target_id 唯一命中一张股票表 —— 实测与 (1) 100% 一致。

    定位不到就**不放进去**，由调用方回落到原始 target_name ——
    宁可不补，也不能补错。
    """
    # 1) 先从快照取 base_id / sku
    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 {}

    bases = {
        m.id: m for m in MaterialBase.query.filter(MaterialBase.id.in_(need_base)).all()
    }

    out = {}
    for r in rows:
        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)
        if b is None:
            continue
        out[r.id] = {
            'base_id': base_id,
            'sku': (sku 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, fallback=''):
    """
    物料上下文 → 「SKU - 物料名称 (规格型号)」。取不到就返回空串（调用方回落）。

    ★ 缺哪段就省哪段，不留空括号/空横杠：材料名查不到时显示
      「0000001180 - 」比只显示「0000001180」更让人困惑。
    """
    if not ctx:
        return ''
    head = ctx.get('sku') 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
        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(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)


# 摘要核心属性按**槽位**取：每个槽位只取第一个有值的字段。
#
# ★ 为什么不是简单按字段列表顺序取前 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):
    """快照 → ['入库数量 10 件', '库位 ZZTEST', …]，每个槽位取一个。"""
    parts = []
    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 changes_summary(row, ref_maps, limit=200, resolved=None, material=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

    verb = '新增' if canon_action(row.action) == 'CREATE' else '删除'
    mod = module_display(row.module)
    prefix = f"{verb}（{mod}）：" if mod else f"{verb}："

    for which in (SNAPSHOT_CREATED, SNAPSHOT_DELETED):
        snap = snapshot_of(row, which)
        if not snap:
            continue
        # 优先：核心属性的**值**（多少数量、放哪）—— 这才是用户想知道的
        core = _core_attrs(snap, material)
        if core:
            return truncate(prefix + '、'.join(core), limit)
        # 回落：没有核心属性可提（如 BOM 记录），退化成字段名清单。
        # 跳过 id：主键每条都有，排最前却零信息量，白占位置。
        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"{prefix}{'、'.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, 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)

    # ---------- 审计日志（台账）----------
    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),
               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()