#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""挖 商机归因 —— 为《集团维度_增减归因看板》补「商机」「风险/机会」两列提供数据。

数据源（只读）：
  · ~/.hermes/profiles/zone-b/cache/documents/doc_b68712feec68_刘子研三季度数据-截止至9.30(1).xls
      sheet「2026年销售数据」「2025年销售数据」 —— 货号# / 商机(飞书)# / 集团客户# / 价税合计
  · 数据/sr_review_kac.json         客诉/要求（货号级）
  · 数据/customer_special_requirements.json  特殊要求（货号级）
  · 数据/mcp_complaints.json        客诉单（客户级，无货号）
  · 数据/activities_feishu.json     飞书活动自由文本（6251 条）

口径：标准销售订单 · 已审核 · 非赠品 · 排除关联方；金额=价税合计；主口径 1–9 月同期。

分解单元 = (集团, 商机, 货号)，非重叠，ΣpairΔ 恒等于集团Δ（守恒门禁）。
显著项 = |Δ| ≥ 5,000 元 或 |Δ| ≥ 5% × |该集团 Δ|。

阶段：
  --pairs      算 pair + 显著 + 证据包 → 数据/_商机_pairs.json
  --extract    GLM 挖活动信号（分片） → 数据/_商机_活动信号_s*.json，--merge N 合一
  --attribute  GLM 逐显著项定归因（分片） → 数据/_商机_归因_s*.json
  --build      合最终可重建数据文件 → 数据/商机归因_20261011.json

GLM 凭证直读 ~/.hermes/config.yaml（复用 ai_chat._call_glm）；密钥绝不打印、绝不落盘。
"""
import argparse
import json
import os
import re
import sys
import time
from collections import Counter, defaultdict

import pandas as pd

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(ROOT, '脚本'))

USR = os.path.expanduser(
    '~/.hermes/profiles/zone-b/cache/documents/doc_b68712feec68_刘子研三季度数据-截止至9.30(1).xls')
DA = os.path.join(ROOT, '数据')
OUT_FINAL = os.path.join(DA, '商机归因_20261011.json')
P_PAIRS = os.path.join(DA, '_商机_pairs.json')
P_SIG = os.path.join(DA, '_商机_活动信号.json')
P_ATTR = os.path.join(DA, '_商机_归因.json')

OPP_EMPTY = '未关联商机'
SKU_EMPTY = '（未填货号）'

OPP_TYPES = ['新项目导入', '项目扩容/升级', '成功替代竞品', '客户下游市场增长', '注册/合规驱动', '复购或长期协议']
RISK_TYPES = ['客诉/质量问题', '竞品性能或价格更优', '客户项目终止/失败', '注册未通过', '客户自研自制',
              '采购策略变更（多源/压价）', '机会性商机未延续']
NO_EVIDENCE = '未找到相关记录'

# 用户已提供（销售口述）—— 天隆三条，不得改写
ORAL = [
    {'sku_match': 'PFM4211', 'grp': '天隆集团', 'opp': '', 'kind': '风险', 'type': '机会性商机未延续',
     'text': 'HPV 海外机会未延续', 'level': '销售口述',
     'evidence': '销售口述（用户提供，2026-10）：PFM4211 HPV 海外机会未延续'},
    {'sku_match': 'M5244-ZJ4', 'grp': '天隆集团', 'opp': '', 'kind': '风险', 'type': '客诉/质量问题',
     'text': '稳定性客诉，已由 MD2104-003006 替换', 'level': '销售口述',
     'evidence': '销售口述（用户提供，2026-10）：M5244-ZJ4 稳定性客诉，已由 MD2104-003006 替换'},
    {'sku_match': 'M4211-3', 'grp': '天隆集团', 'opp': '', 'kind': '机会', 'type': '项目扩容/升级',
     'text': 'HBV 超敏项目势头好', 'level': '销售口述',
     'evidence': '销售口述（用户提供，2026-10）：M4211-3 HBV 超敏项目势头好'},
]


def oral_for(x):
    """命中销售口述（须同集团 + 货号同族），否则 None。"""
    for o in ORAL:
        if x['grp'] == o['grp'] and (x['sku'] == o['sku_match'] or x['sku'].startswith(o['sku_match'] + '-')):
            return o
    return None

ACT_KW = ('丽珠', '复星', '佰奥莱博', '海尔施', '天隆', '新产业', '迪安', '丹威', '基灵',
          '华益美', '安瑞科', '雅为', '军医大学', '杨博')


def norm0(s):
    s = str(s).strip()
    return '' if s in ('nan', 'None', 'NaT', 'null', 'NaN') else s


# ── ① 读 xls → pair ──────────────────────────────────────────
def load_sheet(sheet):
    d = pd.read_excel(USR, sheet_name=sheet, header=0, engine='xlrd')
    d.columns = [str(c).strip() for c in d.columns]
    sk = d['货号#'].fillna('').astype(str).str.strip().replace({'nan': ''})
    d['_sku'] = sk.where(sk != '', SKU_EMPTY)
    d['_d'] = pd.to_datetime(d['合同日期'], errors='coerce')
    d['_amt'] = pd.to_numeric(d['价税合计'], errors='coerce').fillna(0)
    d['_cust'] = d['客户'].astype(str).str.strip()
    g = d['集团客户#'].fillna('').astype(str).str.strip().replace({'nan': '', 'None': '', 'NaN': ''})
    d['_grp'] = g.where(g != '', d['_cust'])
    o = d['商机(飞书)#'].fillna('').astype(str).str.strip().replace({'nan': '', 'None': '', 'NaN': ''})
    d['_opp'] = o.where(o != '', OPP_EMPTY)
    return d


def clean(d):
    k = d['单据类型'].astype(str).str.contains('标准销售订单', na=False)
    k &= d['单据状态'].astype(str).str.contains('已审核', na=False)
    k &= ~d['是否赠品'].astype(str).str.strip().isin(['是', 'Y', 'True', 'true'])
    k &= ~d['_cust'].str.contains('宝锐|宝泰|横琴', na=False, regex=True)
    return d[k].copy()


def period(d, y, m_end=9):
    return d[(d['_d'].dt.year == y) & (d['_d'].dt.month <= m_end)]


def read_json_list(p):
    d = json.load(open(p, encoding='utf-8'))
    if isinstance(d, list):
        return d
    return d.get('items') or next((v for v in d.values() if isinstance(v, list)), [])


def build_pairs():
    d26p = period(clean(load_sheet('2026年销售数据')), 2026)
    d25p = period(clean(load_sheet('2025年销售数据')), 2025)
    g25 = d25p.groupby('_grp')['_amt'].sum()
    g26 = d26p.groupby('_grp')['_amt'].sum()
    ga = pd.concat([g25.rename('a25'), g26.rename('a26')], axis=1).fillna(0.0)
    ga['d'] = ga['a26'] - ga['a25']
    GD = ga['d'].to_dict()

    KEY = ['_grp', '_opp', '_sku']
    p = pd.concat([d25p.groupby(KEY)['_amt'].sum().rename('a25'),
                   d26p.groupby(KEY)['_amt'].sum().rename('a26')], axis=1).fillna(0.0).reset_index()
    p['d'] = p['a26'] - p['a25']
    cu = pd.concat([d25p, d26p]).groupby(KEY)['_cust'].agg(lambda s: '/'.join(sorted(set(s))))
    p = p.merge(cu.rename('cust'), on=KEY, how='left')

    # 守恒门禁
    per = p.groupby('_grp')['d'].sum()
    mx = max(abs(per.get(g, 0.0) - GD.get(g, 0.0)) for g in GD)
    assert mx < 0.05, 'pair 分解不守恒：%.6f' % mx
    assert abs(ga['d'].sum() - (d26p['_amt'].sum() - d25p['_amt'].sum())) < 0.05, '总级不守恒'

    pairs = []
    for _, r in p.iterrows():
        g = r['_grp']
        thr = max(5000.0, 0.05 * abs(GD.get(g, 0.0)))
        pairs.append({
            'grp': g, 'opp': r['_opp'], 'sku': r['_sku'],
            'a25': round(float(r['a25']), 2), 'a26': round(float(r['a26']), 2), 'd': round(float(r['d']), 2),
            'cust': r['cust'], 'thr': round(thr, 2), 'sig': bool(abs(r['d']) >= thr),
        })
    pairs.sort(key=lambda x: -abs(x['d']))

    # ── 证据包 ─────────────────────────────────────────────
    kac = read_json_list(os.path.join(DA, 'sr_review_kac.json'))
    csr = read_json_list(os.path.join(DA, 'customer_special_requirements.json'))
    cmp_ = read_json_list(os.path.join(DA, 'mcp_complaints.json'))
    sku_kac, sku_csr, cust_cmp = defaultdict(list), defaultdict(list), defaultdict(list)
    for it in kac:
        for s in (it.get('skus') or []):
            sku_kac[norm0(s)].append(it)
    for it in csr:
        for s in (it.get('skus') or []):
            sku_csr[norm0(s)].append(it)
    for it in cmp_:
        cust_cmp[norm0(it.get('customer'))].append(it)

    for x in pairs:
        if not x['sig']:
            x['evidence'] = []
            continue
        ev = []
        for it in sku_kac.get(x['sku'], [])[:3]:
            ev.append({'src': '客诉复核(KAC)', 'title': it.get('title'), 'content': (it.get('content') or '')[:120],
                       'date': it.get('activity_date'), 'customer': it.get('customer'),
                       'id': it.get('activity_id'), '货号相关': True})
        for it in sku_csr.get(x['sku'], [])[:3]:
            ev.append({'src': '特殊要求(CSR)', 'title': it.get('title'), 'content': (it.get('content') or '')[:120],
                       'date': (it.get('created_at') or '')[:10], 'customer': it.get('customer'),
                       'id': it.get('id'), '货号相关': True})
        cmps = []
        for c in str(x['cust']).split('/'):
            for it in cust_cmp.get(c.strip(), []):
                nm = str(it.get('name') or '')
                rel = bool(x['sku'] and x['sku'] != SKU_EMPTY and x['sku'] in nm)
                cmps.append({'src': '客诉单(MCP·客户级)', 'title': nm, 'content': str(it.get('type') or ''),
                             'date': it.get('created'), 'customer': it.get('customer'), 'id': it.get('id'),
                             'status': it.get('status'), '货号相关': rel})
        # 货号相关的客诉排前面，最多取 3 条（客户级、货号无关的仅作背景）
        cmps.sort(key=lambda z: (not z['货号相关'],))
        ev.extend(cmps[:3])
        x['evidence'] = ev

    data = {'meta': {'src': os.path.basename(USR), 'period': '2026-01-01~09-30 vs 2025-01-01~09-30',
                     'rule': '标准销售订单·已审核·非赠品·排除关联方；金额=价税合计',
                     'unit': '(集团, 商机, 货号) 非重叠分解', 'thr': '|Δ|≥5000 或 |Δ|≥5%×|集团Δ|'},
            'group_delta': {k: round(v, 2) for k, v in GD.items()},
            'pairs': pairs}
    json.dump(data, open(P_PAIRS, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
    nsig = sum(1 for x in pairs if x['sig'])
    print('pair 总数 %d ｜ 显著 %d ｜ 命名商机显著 %d' % (
        len(pairs), nsig, sum(1 for x in pairs if x['sig'] and x['opp'] != OPP_EMPTY)))
    print('守恒 max|ΣpairΔ-集团Δ| = %.6f ✅' % mx)
    print('→ %s' % P_PAIRS)
    return data


# ── ② 活动相关性筛选 ─────────────────────────────────────────
def load_activities():
    return read_json_list(os.path.join(DA, 'activities_feishu.json'))


def relevant_activities(pairs, custs):
    acts = load_activities()
    out = []
    for a in acts:
        c = norm0(a.get('customer'))
        txt = str(a.get('text') or '') + ' ' + str(a.get('type') or '')
        keep = False
        if c:
            for x in custs:
                if x and (x == c or x in c or c in x):
                    keep = True
                    break
        if not keep and any(k in txt for k in ACT_KW):
            keep = True
        if keep:
            out.append({'id': a.get('id'), 'date': a.get('date'), 'sales': a.get('sales'),
                        'customer': a.get('customer'), 'atype': a.get('atype'), 'type': a.get('type'),
                        'text': re.sub(r'\s+', ' ', str(a.get('text') or ''))[:700]})
    return out


# ── GLM ──────────────────────────────────────────────────────
SYS_SIG = """你是宝锐生物（分子诊断原料供应商）的销售情报提取助手。从飞书活动（拜访/沟通）记录里抽取**可支撑「货号增减归因」的信号**，四类：
① 竞对信号：出现的竞品品牌 + 性能/价格/交付对比；
② 客户需求/项目状态：新项目、扩产、上量、试样、换料、项目暂停/终止；
③ 客诉/质量投诉：质量问题、投诉、退换货、稳定性/批间差等；
④ 项目里程碑：注册/备案、中试、送样、量产导入、协议签署。
只抽取文中**有明确依据**的；寒暄、无信息量、我方单向推销一律不输出。

输出**严格 JSON 数组**（禁 markdown 包裹、禁解释），每项：
{"活动id":"<原样>","日期":"<原样>","客户":"<原样>","类别":"竞对|需求|客诉|里程碑","涉及货号":["..."],"摘要":"≤30字","依据":"<活动原文最关键的一句，≤60字>","置信度":"高|中|低"}
没有可抽取信号可返回 []。"""

SYS_ATTR = """你是宝锐生物经营分析归因助手。给定「集团 × 商机 × 货号」的同比增减（Δ=2026同期−2025同期，元）及其证据（客诉/特殊要求/活动信号），为每一项判定**归因类型 + 一句话原因 + 证据等级**。

归因类型（只能选其一）：
- 机会类（Δ>0 时优先）：新项目导入、项目扩容/升级、成功替代竞品、客户下游市场增长、注册/合规驱动、复购或长期协议
- 风险类（Δ<0 时优先）：客诉/质量问题、竞品性能或价格更优、客户项目终止/失败、注册未通过、客户自研自制、采购策略变更（多源/压价）、机会性商机未延续

证据等级（只能选其一）：
- 系统有据：客诉单/特殊要求/活动原文里明确写着，必须给「出处」（文件+标题/日期/原文片段）
- 推测待核：仅凭间接线索推断，证据不足
**没有任何证据时不得编造原因**：类型填「无」，原因填「未找到相关记录」，等级填「推测待核」。

⚠️ 证据相关性铁律：
- 「客诉单(MCP·客户级)」**不带货号**，字段 `货号相关` 若为 false，表示该客诉是**客户名下其他产品**的，
  **不得**用它给本行货号定因（只能当背景）。只有 `货号相关`=true 或「客诉复核(KAC)」「特殊要求(CSR)」
  （这两类按货号精确命中）才能支持「客诉/质量问题」。
- 特殊要求（CSR）多为**分装/包装/交付/质检**类约定，本身不构成增减原因者，不要强行归因。
- 方向与类型的对应：Δ>0 优先机会类，Δ<0 优先风险类；若证据指向相反方向，以「无 + 未找到相关记录」处理。

输出**严格 JSON 数组**（禁 markdown），每项：
{"键":"<原样回填>","方向":"机会|风险|无","类型":"<上述之一，无则'无'>","一句话":"≤34字","证据等级":"系统有据|销售口述|推测待核","出处":"<文件名+标题/日期/原文片段，无则空>"}
逐项输出，不得合并、不得漏项。"""


MINIMAX_KEY_FILE = os.path.expanduser('~/.config/minimax/music_api_key')
MINIMAX_URL = 'https://api.minimaxi.com/v1/text/chatcompletion_v2'
MINIMAX_MODEL = 'MiniMax-Text-01'


def _call_minimax(system, user):
    """MiniMax 文本模型（api.minimaxi.com 域，key 与图像/音乐同域同 key）。
    key 从既有文件读取，绝不打印、绝不落盘。"""
    import urllib.request
    try:
        key = open(MINIMAX_KEY_FILE, encoding='utf-8').read().strip()
    except Exception:
        key = ''
    if not key:
        raise RuntimeError('MiniMax key 未找到')
    body = {'model': MINIMAX_MODEL,
            'messages': [{'role': 'system', 'content': system},
                         {'role': 'user', 'content': user}],
            'max_tokens': 16000, 'temperature': 0.3}
    req = urllib.request.Request(
        MINIMAX_URL, data=json.dumps(body).encode('utf-8'),
        headers={'Content-Type': 'application/json',
                 'Authorization': 'Bearer %s' % key}, method='POST')
    with urllib.request.urlopen(req, timeout=180) as r:
        d = json.loads(r.read())
    msg = (d.get('choices') or [{}])[0].get('message') or {}
    return (msg.get('content') or '').strip()


DEEPSEEK_URL = 'https://api.deepseek.com'
DEEPSEEK_MODEL_DEFAULT = 'deepseek-flash'


def _deepseek_cfg():
    """读 ~/.hermes/config.yaml providers.deepseek。只取 key/model，绝不打印、绝不落盘。"""
    import yaml
    try:
        cfg = yaml.safe_load(open(os.path.expanduser('~/.hermes/config.yaml'), encoding='utf-8'))
        ds = (cfg.get('providers') or {}).get('deepseek') or {}
        return {'key': (ds.get('api_key') or ''),
                'base': (ds.get('base_url') or DEEPSEEK_URL).rstrip('/'),
                'model': ds.get('default_model') or DEEPSEEK_MODEL_DEFAULT}
    except Exception:
        return {'key': '', 'base': DEEPSEEK_URL, 'model': DEEPSEEK_MODEL_DEFAULT}


def _call_deepseek(system, user):
    """DeepSeek（OpenAI 兼容）。thinking 必须 disabled，否则答案落在 reasoning_content。"""
    import urllib.request
    cfg = _deepseek_cfg()
    if not cfg['key']:
        raise RuntimeError('DeepSeek key 未找到')
    body = {'model': cfg['model'],
            'messages': [{'role': 'system', 'content': system},
                         {'role': 'user', 'content': user}],
            'max_tokens': 8000, 'temperature': 0.3,
            'thinking': {'type': 'disabled'}}
    req = urllib.request.Request(
        cfg['base'] + '/chat/completions', data=json.dumps(body).encode('utf-8'),
        headers={'Content-Type': 'application/json',
                 'Authorization': 'Bearer %s' % cfg['key']}, method='POST')
    with urllib.request.urlopen(req, timeout=300) as r:
        d = json.loads(r.read())
    msg = (d.get('choices') or [{}])[0].get('message') or {}
    return (msg.get('content') or msg.get('reasoning_content') or '').strip()


# 引擎降级链（默认按用户口径：GLM 主 → MiniMax 兜底 → DeepSeek 末位保底）
# 任一引擎连续失败 2 次即熔断，本进程后续不再空等它；可用 ATTR_ENGINES 覆盖。
ENGINE_CHAIN = [x.strip() for x in os.environ.get('ATTR_ENGINES', 'glm,minimax,deepseek').split(',') if x.strip()]
_FAIL = {}          # 引擎连续失败计数：同一引擎连挂 2 次即熔断，本进程后续不再空等它


def llm(system, user, eng=None):
    """按 ENGINE_CHAIN 逐个尝试，返回 (content, engine)。空返/异常都算失败。"""
    for name in ENGINE_CHAIN:
        if _FAIL.get(name, 0) >= 2:
            continue
        try:
            if name == 'glm':
                import ai_chat
                c, e = ai_chat._call_glm(system, [{'role': 'user', 'content': user}])
                if not c or e:
                    raise RuntimeError('GLM 空返/错误')
            elif name == 'deepseek':
                c = _call_deepseek(system, user)
            elif name == 'minimax':
                c = _call_minimax(system, user)
            else:
                continue
            if not c:
                raise RuntimeError('%s 空返' % name)
            if eng is not None:
                eng[name] = eng.get(name, 0) + 1
            return c, name
        except Exception as ex:
            _FAIL[name] = _FAIL.get(name, 0) + 1
            print('   ⚠️ %s 失败(%d/2)：%s' % (name, _FAIL[name], str(ex)[:80]))
    if eng is not None:
        eng['fail'] = eng.get('fail', 0) + 1
    return None, None


def parse_arr(content):
    if not content:
        return []
    m = re.search(r'\[.*\]', content, re.S)
    if not m:
        return []
    for cand in (m.group(0), m.group(0).replace('\n', ' ')):
        try:
            v = json.loads(cand)
            return v if isinstance(v, list) else []
        except Exception:
            continue
    return []


def do_extract(shard=0, shards=1, batch=12, limit=0):
    D = json.load(open(P_PAIRS, encoding='utf-8'))
    custs = sorted({c.strip() for x in D['pairs'] if x['sig'] for c in str(x['cust']).split('/') if c.strip()})
    acts = relevant_activities(D['pairs'], custs)
    json.dump(acts, open(os.path.join(DA, '_商机_相关活动.json'), 'w', encoding='utf-8'),
              ensure_ascii=False, indent=1)
    if shards > 1:
        acts = [a for i, a in enumerate(acts) if i % shards == shard]
    outp = P_SIG if shards == 1 else P_SIG.replace('.json', '_s%d.json' % shard)
    done, items, eng = {}, [], {}
    if os.path.exists(outp):
        prev = json.load(open(outp, encoding='utf-8'))
        done = dict(prev.get('by_activity') or {})
        items = list(prev.get('items') or [])
        eng = dict(prev.get('engines') or {})
    todo = [a for a in acts if a['id'] not in done]
    if limit:
        todo = todo[:limit]
    print('分片 %d/%d 相关活动 %d ｜ 已处理 %d ｜ 本次 %d（每批 %d）' % (
        shard + 1, shards, len(acts), len(done), len(todo), batch))
    ok = err = 0
    for i in range(0, len(todo), batch):
        grp = todo[i:i + batch]
        c, engine = llm(SYS_SIG, '从以下 %d 条活动记录中抽取信号：\n%s' % (
            len(grp), json.dumps([{'活动id': a['id'], '日期': a['date'], '客户': a['customer'],
                                   '类型': a['atype'] or a['type'], '内容': a['text']} for a in grp],
                                 ensure_ascii=False)), eng)
        its = parse_arr(c)
        if c is None:
            err += 1
        else:
            ok += 1
        for it in its:
            if isinstance(it, dict) and it.get('摘要'):
                it['_引擎'] = engine
                items.append(it)
        for a in grp:
            done[a['id']] = 1
        json.dump({'updated': time.strftime('%Y-%m-%d %H:%M:%S'), 'engines': eng,
                   'by_activity': done, 'items': items},
                  open(outp, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
        print('  批 %d/%d ✓ [%s] 本批 %d 条（累计 %d）' % (
            i // batch + 1, (len(todo) + batch - 1) // batch, engine, len(its), len(items)))
    print('分片 %d 完成：成功 %d 批 ｜ 失败 %d 批 ｜ 累计信号 %d ｜ 引擎 %s' % (
        shard + 1, ok, err, len(items), eng))
    return items


def merge_extract(shards):
    items, done, eng = [], {}, {}
    for i in range(shards):
        p = P_SIG.replace('.json', '_s%d.json' % i)
        if os.path.exists(p):
            d = json.load(open(p, encoding='utf-8'))
            items += d.get('items') or []
            done.update(d.get('by_activity') or {})
            for k, v in (d.get('engines') or {}).items():
                eng[k] = eng.get(k, 0) + v
    json.dump({'updated': time.strftime('%Y-%m-%d %H:%M:%S'), 'engines': eng,
               'by_activity': done, 'items': items},
              open(P_SIG, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
    print('合并 %d 片 → 信号 %d 条 ｜ 覆盖活动 %d ｜ 引擎 %s' % (shards, len(items), len(done), eng))


# ── ③ 逐显著项定归因 ─────────────────────────────────────────
def do_attribute(shard=0, shards=1, batch=6, limit=0):
    D = json.load(open(P_PAIRS, encoding='utf-8'))
    sigs = [x for x in D['pairs'] if x['sig']]
    sigs.sort(key=lambda x: -abs(x['d']))
    if shards > 1:
        sigs = [x for i, x in enumerate(sigs) if i % shards == shard]
    if limit:
        sigs = sigs[:limit]

    sig_items = []
    if os.path.exists(P_SIG):
        sig_items = json.load(open(P_SIG, encoding='utf-8')).get('items') or []
    # 预置：销售口述（天隆三条）
    pre = {}
    for x in D['pairs']:
        o = oral_for(x)
        if not o:
            continue
        pre[key_of(x)] = {'方向': o['kind'], '类型': o['type'], '一句话': o['text'],
                          '证据等级': o['level'], '出处': o['evidence'],
                          '键': key_of(x), 'sku': x['sku'], 'preset': True}

    outp = P_ATTR if shards == 1 else P_ATTR.replace('.json', '_s%d.json' % shard)
    got, eng = {}, {}
    if os.path.exists(outp):
        prev = json.load(open(outp, encoding='utf-8'))
        got = {k: v for k, v in (prev.get('by_key') or {}).items()}
        eng = dict(prev.get('engines') or {})

    def pack(x):
        rel = [s for s in sig_items
               if (norm0(s.get('客户')) and any(c.strip() and c.strip() in norm0(s.get('客户'))
                                                 for c in str(x['cust']).split('/')))
               or any(norm0(k) and norm0(k) in x['sku'] for k in (s.get('涉及货号') or []))
               or any(t and t in str(s.get('依据') or '') + str(s.get('摘要') or '')
                      for t in re.findall(r'[A-Za-z]{2,}\d+[\w-]*', x['sku']))]
        return {
            '键': key_of(x), '集团': x['grp'], '商机': x['opp'], '货号': x['sku'],
            '25同期': x['a25'], '26同期': x['a26'], 'Δ': x['d'], '客户': x['cust'],
            '客诉/要求证据': x['evidence'][:6],
            '活动信号': [{'类别': s.get('类别'), '摘要': s.get('摘要'), '依据': s.get('依据'),
                          '日期': s.get('日期'), '置信度': s.get('置信度')} for s in rel[:6]],
        }

    todo = [x for x in sigs if key_of(x) not in got and key_of(x) not in pre]
    print('分片 %d/%d 显著 %d ｜ 已定 %d ｜ 预置口述 %d ｜ 本次 %d（每批 %d）' % (
        shard + 1, shards, len(sigs), len(got), len(pre), len(todo), batch))
    got.update({k: v for k, v in pre.items()})
    ok = err = 0
    for i in range(0, len(todo), batch):
        grp = todo[i:i + batch]
        c, engine = llm(SYS_ATTR, '逐项归因（%d 项）：\n%s' % (
            len(grp), json.dumps([pack(x) for x in grp], ensure_ascii=False)), eng)
        arr = parse_arr(c)
        if c is None:
            err += 1
        else:
            ok += 1
        byk = {str(it.get('键')): it for it in arr if isinstance(it, dict) and it.get('键')}
        for x in grp:
            k = key_of(x)
            if k in byk:
                it = byk[k]
                lv = it.get('证据等级') or '推测待核'
                if lv not in ('系统有据', '销售口述', '推测待核'):
                    lv = '推测待核'
                ty = it.get('类型') or '无'
                if ty not in OPP_TYPES + RISK_TYPES + ['无']:
                    ty = '无'
                why = it.get('一句话') or NO_EVIDENCE
                if ty == '无' or lv == '推测待核' and not (it.get('出处') or '').strip():
                    if lv == '推测待核' and not (it.get('出处') or '').strip():
                        why = why if why != NO_EVIDENCE else NO_EVIDENCE
                got[k] = {'方向': it.get('方向') or '无', '类型': ty, '一句话': why,
                          '证据等级': lv, '出处': it.get('出处') or '', '键': k, '引擎': engine}
            else:
                got[k] = {'方向': '无', '类型': '无', '一句话': NO_EVIDENCE,
                          '证据等级': '推测待核', '出处': '', '键': k, '引擎': None}
        json.dump({'updated': time.strftime('%Y-%m-%d %H:%M:%S'), 'engines': eng, 'by_key': got},
                  open(outp, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
        print('  批 %d/%d ✓ [%s] 返回 %d 项（累计已定 %d）' % (
            i // batch + 1, (len(todo) + batch - 1) // batch, engine, len(arr), len(got)))
    print('分片 %d 完成：成功 %d 批 ｜ 失败 %d 批 ｜ 已定 %d ｜ 引擎 %s' % (
        shard + 1, ok, err, len(got), eng))
    return got


def merge_attr(shards):
    got, eng = {}, {}
    for i in range(shards):
        p = P_ATTR.replace('.json', '_s%d.json' % i)
        if os.path.exists(p):
            d = json.load(open(p, encoding='utf-8'))
            got.update(d.get('by_key') or {})
            for k, v in (d.get('engines') or {}).items():
                eng[k] = eng.get(k, 0) + v
    json.dump({'updated': time.strftime('%Y-%m-%d %H:%M:%S'), 'engines': eng, 'by_key': got},
              open(P_ATTR, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
    print('合并 %d 片 → 归因 %d 项 ｜ 引擎 %s' % (shards, len(got), eng))


def key_of(x):
    return '%s||%s||%s' % (x['grp'], x['opp'], x['sku'])


# ── ④ 最终数据文件（可重建）──────────────────────────────────
# ── 出处缩写还原（只用已存在的证据原文，绝不生成新内容）────────────
def _evidence_pool():
    """可用来还原省略号的原文池：活动信号依据 + 原始活动全文 + 客诉/CSR/客诉单文本。"""
    pool = []
    sig = {}
    if os.path.exists(P_SIG):
        sig = json.load(open(P_SIG, encoding='utf-8'))
    items = sig.get('items') or []
    acts = {a.get('id'): str(a.get('text') or '') for a in load_activities()}
    for s in items:
        pool.append(str(s.get('依据') or ''))
        pool.append(str(s.get('摘要') or ''))
        t = acts.get(s.get('活动id'))
        if t:
            pool.append(t)
    for f in ('sr_review_kac.json', 'customer_special_requirements.json', 'mcp_complaints.json'):
        for it in read_json_list(os.path.join(DA, f)):
            pool += [str(it.get('content') or ''), str(it.get('title') or ''),
                     str(it.get('name') or ''), str(it.get('description') or '')]
    out = []
    for p in pool:
        p = re.sub(r'\s+', ' ', p).strip()
        if p and '…' not in p and len(p) > 12:
            out.append(p)
    return out


def _expand_one(s, pool):
    """把 s 里的『…』补回原文中间缺失的一段；只在原文池里能定位到才补，否则原样保留（不编造）。

    做法：取『…』左侧 20 字作左锚点 L、右侧 10 字作右锚点 R，在原文池里找同时含
    L…R 的段落，把两锚点之间的原文插回；找不到就停手。补入的是库里原文，非模型生成。
    """
    out = s
    for _ in range(8):
        if '…' not in out:
            break
        i = out.index('…')
        left, right = out[:i], out[i + 1:]
        L = left[-20:] if len(left) >= 8 else left
        R = right[:10]
        best = _match_middle(left, right, pool)
        if best is None:
            break
        out = out[:i] + best + out[i + 1:]
    return out


def _match_middle(left, right, pool):
    """在原文池里定位 left 尾段…right 首段之间的原文；锚点由长到短回退，取最短补全。"""
    for llen in (20, 16, 12, 9, 6):
        L = left[-llen:]
        if len(L) < 6:
            continue
        for rlen in (12, 10, 8, 6, 4):
            R = right[:rlen]
            if len(R) < 4:
                continue
            best = None
            for p in pool:
                a = p.find(L)
                if a < 0:
                    continue
                b = p.find(R, a + len(L))
                if b < 0:
                    continue
                ins = p[a + len(L):b]
                if not ins.strip() or len(ins) > 200:
                    continue
                if best is None or len(ins) < len(best):
                    best = ins
            if best is not None:
                return best
    return None


def expand_excerpts(attr):
    """归因「出处」里模型引用时的缩写（…）→ 还原为证据原文。返回还原条数。"""
    pool = _evidence_pool()
    fixed = 0
    for k, v in (attr or {}).items():
        src = v.get('出处') or ''
        if '…' not in src:
            continue
        new = _expand_one(src, pool)
        if new != src:
            v['出处'] = new
            fixed += 1
    return fixed, pool


def do_build():
    D = json.load(open(P_PAIRS, encoding='utf-8'))
    attr = {}
    eng_sig, eng_attr = {}, {}
    if os.path.exists(P_SIG):
        _s = json.load(open(P_SIG, encoding='utf-8'))
        eng_sig = _s.get('engines') or {}
    if os.path.exists(P_ATTR):
        _a = json.load(open(P_ATTR, encoding='utf-8'))
        attr = _a.get('by_key') or {}
        eng_attr = _a.get('engines') or {}
    # 出处缩写还原（模型引用时写的『…』→ 证据原文；对不上则原样保留）
    n_fix, _pool = expand_excerpts(attr)
    print('出处缩写还原：%d 条（原文池 %d 段）' % (n_fix, len(_pool)))
    # 口述兜底（若 attribute 未跑）
    for x in D['pairs']:
        o = oral_for(x)
        if x['sig'] and o:
            attr.setdefault(key_of(x), {'方向': o['kind'], '类型': o['type'], '一句话': o['text'],
                                        '证据等级': o['level'], '出处': o['evidence'], '键': key_of(x)})

    pairs_out = []
    cnt_lv = Counter()
    cnt_ty = Counter()
    for x in D['pairs']:
        rec = dict(x)
        if x['sig']:
            a = attr.get(key_of(x), {})
            rec['归因'] = {
                '方向': a.get('方向', '无'), '类型': a.get('类型', '无'),
                '一句话': a.get('一句话') or NO_EVIDENCE,
                '证据等级': a.get('证据等级', '推测待核'), '出处': a.get('出处', ''),
            }
            cnt_lv[rec['归因']['证据等级']] += 1
            cnt_ty[rec['归因']['类型'] if rec['归因']['类型'] != '无' else '无（未找到相关记录）'] += 1
        else:
            rec['归因'] = None
        pairs_out.append(rec)
    sigs = [x for x in pairs_out if x['sig']]

    # 看板行查找表：(集团, 货号) → 主导 pair。「商机」「风险/机会」两列描述同一个 pair。
    # 规则：先取 |Δ| 最大的 pair；若有命名商机 pair 且其 |Δ| ≥ 最大 |Δ| 的 30%，
    #       则以该命名 pair 为准（避免「小切片挂名商机」压过真正驱动行的未关联大额）。
    cand = defaultdict(list)
    for x in pairs_out:
        cand['%s||%s' % (x['grp'], x['sku'])].append(x)
    board = {}
    for k, xs in cand.items():
        top = max(xs, key=lambda r: abs(r['d']))
        pick = top
        named = [x for x in xs if x['opp'] != OPP_EMPTY]
        if named:
            bn = max(named, key=lambda r: abs(r['d']))
            if abs(top['d']) == 0 or abs(bn['d']) >= 0.30 * abs(top['d']):
                pick = bn
        x = pick
        a = x.get('归因')
        board[k] = {'opp': x['opp'], 'sig': bool(x['sig']), 'pair_d': x['d'],
                    'dir': (a or {}).get('方向', '无'), 'type': (a or {}).get('类型', '无'),
                    'why': (a or {}).get('一句话', NO_EVIDENCE),
                    'level': (a or {}).get('证据等级', '推测待核'), 'src': (a or {}).get('出处', '')}
    # 命名商机聚合（显著项求和）
    oppagg = defaultdict(lambda: {'d': 0.0, 'n': 0, 'skus': [], 'grps': set(), 'types': Counter(), 'levels': Counter()})
    for x in sigs:
        o = oppagg[x['opp']]
        o['d'] += x['d']; o['n'] += 1; o['skus'].append(x['sku']); o['grps'].add(x['grp'])
        o['types'][x['归因']['类型']] += 1; o['levels'][x['归因']['证据等级']] += 1
    named = []
    for o, v in oppagg.items():
        if o == OPP_EMPTY:
            continue
        named.append({'商机': o, 'Δ': round(v['d'], 2), '显著货号数': v['n'],
                      '货号': sorted(set(v['skus'])), '集团': sorted(v['grps']),
                      '类型分布': dict(v['types']), '证据等级分布': dict(v['levels'])})
    named.sort(key=lambda x: -abs(x['Δ']))
    res = oppagg[OPP_EMPTY]

    out = {
        'meta': {
            '生成日期': time.strftime('%Y-%m-%d %H:%M:%S'),
            '数据源': D['meta']['src'],
            '口径': D['meta']['rule'] + '；主口径 ' + D['meta']['period'],
            '分解单元': D['meta']['unit'],
            '显著项判据': D['meta']['thr'],
            '证据等级定义': ['系统有据（客诉单/特殊要求/活动原文明写，附出处）',
                             '销售口述（用户提供，天隆三条）', '推测待核（模型推断，未复核）'],
            '归因类型': {'机会': OPP_TYPES, '风险': RISK_TYPES},
            '视角声明': '本页归因仅覆盖刘子研名下数据，不得当作全集团口径对外。',
            '商机列说明': '源表「商机(飞书)#」列填值率极低（2026 销售 1.0% / 2025 销售 4.6%），'
                      '未填行统一归「未关联商机」，不丢行。',
        },
        '计数': {'pair总数': len(pairs_out), '显著项数': len(sigs),
                 '命名商机显著项': sum(1 for x in sigs if x['opp'] != OPP_EMPTY),
                 '未关联商机显著项': sum(1 for x in sigs if x['opp'] == OPP_EMPTY),
                 '证据等级分布': dict(cnt_lv), '类型分布': dict(cnt_ty),
                 '有归因文本率': '100%%（含「%s」显式标注）' % NO_EVIDENCE,
                 '挖掘引擎_活动信号': eng_sig, '挖掘引擎_归因': eng_attr,
                 '出处缩写还原条数': n_fix},
        '命名商机聚合': named,
        '未关联商机': {'Δ': round(res['d'], 2), '显著货号数': res['n']},
        '看板行归因': board,          # 键 "集团||货号"
        'pairs': pairs_out,
    }
    json.dump(out, open(OUT_FINAL, 'w', encoding='utf-8'), ensure_ascii=False, indent=1)
    print('显著项 %d（命名商机 %d / 未关联 %d）' % (
        len(sigs), out['计数']['命名商机显著项'], out['计数']['未关联商机显著项']))
    print('证据等级:', dict(cnt_lv))
    print('类型分布:', dict(cnt_ty))
    print('→ %s（%.1f KB）' % (OUT_FINAL, os.path.getsize(OUT_FINAL) / 1024.0))
    return out


if __name__ == '__main__':
    ap = argparse.ArgumentParser()
    ap.add_argument('--pairs', action='store_true')
    ap.add_argument('--extract', action='store_true')
    ap.add_argument('--attribute', action='store_true')
    ap.add_argument('--build', action='store_true')
    ap.add_argument('--merge', type=int, default=0)
    ap.add_argument('--stage', choices=['extract', 'attribute'], default='extract')
    ap.add_argument('--shard', type=int, default=0)
    ap.add_argument('--shards', type=int, default=1)
    ap.add_argument('--batch', type=int, default=0)
    ap.add_argument('--limit', type=int, default=0)
    a = ap.parse_args()
    if a.batch == 0:
        a.batch = 12 if a.stage == 'extract' else 6
    if a.pairs:
        build_pairs()
    elif a.extract:
        do_extract(a.shard, a.shards, a.batch, a.limit)
    elif a.attribute:
        do_attribute(a.shard, a.shards, a.batch, a.limit)
    elif a.merge:
        (merge_extract if a.stage == 'extract' else merge_attr)(a.merge)
    elif a.build:
        do_build()
    else:
        build_pairs()