guanlan-src/app_backEnd/app_backEnd_guanlan/deid.py

# -*- coding: utf-8 -*-
"""外发脱敏 — 送到第三方模型之前的实体去标识。

与 scripts/windscada_cloud_pack.py 的分工:
  cloud_pack  管**产物**(HTML 源码级, 含文档名正则等 HTML 特有规则), 是上云前的一次性闸;
  本模块      管**外发**(每次调用在线模型时逐条消息脱敏), 是运行期持续生效的闸。
两者共用下面这张实体表 —— 实体只在这里定义一次, 免得两处规则各自漂移。

为什么外发也必须脱: 演示时客户每问一次, 证据包就整包发给第三方模型一次。
产物脱敏管不到这条路径 —— 那是两个不同的出口。
"""
from app_common.app_common_guanlan.api import paths as P
import re

# 实体映射: 长串在前 (中广核如东 → 先替场名会留下 "中广核the site")
ENTITIES = [
    ('中广核', '[owner]'), ('华能', '[owner]'), ('国家能源', '[owner]'),
    ('国电投', '[owner]'), ('大唐', '[owner]'), ('华电', '[owner]'), ('三峡', '[owner]'),
    ('上海电气', 'OEM'), ('西门子', 'OEM'), ('Siemens', 'OEM'),
    ('Envision', 'supplier'), ('远景', 'supplier'), ('ABB', 'supplier'),
    ('如东海上', 'the offshore site'), ('如东', 'the site'),
    ('rudong', 'site'), ('RUDONG', 'SITE'), ('Rudong', 'Site'),
    ('SWT-4.0-130', '4.0 MW class'), ('SWT130-4000', '4.0 MW class'),
]

_RE_PATH = re.compile(r'/(?:Users|Volumes)/[^\s;,)"\']+')
_RE_TARIFF = re.compile(r'[\d.]+\s*(?:~\s*[\d.]+\s*)?元/kWh')
# 显式台数: 2026-08-31 双模审第 5 条逮到 —— 我写了外发脱敏却没拿产物侧那 14 项闸验过它,
# 实测证据包里 "38 台" 原样外发。写了代码 ≠ 验过, 闸必须对**两个出口**都跑。
_RE_FLEET = re.compile(r'(?<![\d.])38\s*(?:台|units|turbines)')


def scrub(t):
    """对一段文本做外发脱敏。非字符串原样返回。"""
    if not isinstance(t, str) or not t:
        return t
    t = _RE_PATH.sub('[local path]', t)
    t = _RE_TARIFF.sub('[reference tariff]', t)
    t = _RE_FLEET.sub('the fleet', t)
    for k, v in ENTITIES:
        t = t.replace(k, v)
    return t


def scrub_messages(msgs):
    """对 chat messages 逐条脱敏; 只动 content, 不碰 role / tool_calls 结构。

    tool_calls 里的 arguments 不脱: 那是**发给本地工具**的检索参数 (台号要原样才查得到),
    并不会离开本机 —— 离开本机的是 content。混着脱会让检索查不到东西。
    """
    out = []
    for m in msgs:
        if isinstance(m, dict) and isinstance(m.get('content'), str):
            m = dict(m, content=scrub(m['content']))
        out.append(m)
    return out


# ── 机组代号 ───────────────────────────────────────────────────────
# 与 scripts/windscada_cloud_pack.py 的 _unit_map 必须**同算法同结果**: 页面上显示 A56,
# 问答答案里也必须是 A56, 否则客户对不上号。故算法写在这里单源, 两边都调。
# md5 定序保证确定性 (跨进程、跨次出包一致); 撞号沿数字轴平移, 绝不把两台并成一个。
def unit_map(ids):
    import hashlib
    m, used = {}, set()
    for i in sorted(set(ids)):
        h = hashlib.md5(('guanlan-unit-' + i).encode()).hexdigest()
        a, n = chr(65 + int(h[:2], 16) % 26), int(h[2:6], 16) % 90 + 10
        while a + str(n) in used:
            n = (n - 9) % 90 + 10
        used.add(a + str(n))
        m[i] = a + str(n)
    if len(set(m.values())) != len(m):
        raise ValueError(f'代号冲突: {len(m)} 台只生成 {len(set(m.values()))} 个')
    return m


_UMAP = None


def unitize_answer(t, ids=None):
    """把答案里的 WTGnn / nn# 换成代号。仅用于**返回给前端**, 不可用于发给模型的消息 ——
    模型要靠原始台号调检索工具, 换了就查不到。"""
    global _UMAP
    if not isinstance(t, str) or not t:
        return t
    if _UMAP is None:
        if ids is None:
            import json as _j, pathlib as _p, re as _r
            p = P.objects_json()
            ids = sorted(set(_r.findall(r'WTG(\d{2})', p.read_text(encoding='utf-8')))) if p.exists() else []
        _UMAP = unit_map(ids) if ids else {}
    import re as _re
    for i, c in _UMAP.items():
        t = t.replace('WTG' + i, c)
        t = _re.sub(r'(?<!\d)' + i + r'#', c + '#', t)
    return t


def ununitize_question(t, ids=None):
    """把公网代号问题反解回 WTGnn, 仅供服务端检索使用; 不回显给前端。"""
    global _UMAP
    if not isinstance(t, str) or not t:
        return t
    if _UMAP is None:
        if ids is None:
            import pathlib as _p, re as _r
            p = P.objects_json()
            ids = sorted(set(_r.findall(r'WTG(\d{2})', p.read_text(encoding='utf-8')))) if p.exists() else []
        _UMAP = unit_map(ids) if ids else {}
    inv = {v: 'WTG' + k for k, v in _UMAP.items()}
    import re as _re
    for alias, raw in sorted(inv.items(), key=lambda kv: -len(kv[0])):
        t = _re.sub(r'(?<![A-Za-z0-9])' + _re.escape(alias) + r'(?![A-Za-z0-9])', raw, t)
    return t