# -*- coding:utf-8 -*-
"""P2-③ 类内共性提取引擎：原子化 → 归一化 → 能力域 → 覆盖率。

流程（对单个 cluster 按需执行，用户/agent 触发，不做全批自动跑——成本控制）：
  1. 原子化：类内文档按 20/批 LLM 提取原子需求（并发 Semaphore 3，purpose=utility），
     落 opp_demand_atoms（snap×atom 一行）。
  2. 归一化：原子文本向量化（rag embed 共享通道）→ 簇内 kNN + τ2(默认0.85) 并查集
     得候选组 → **LLM 裁决**组内是否同一标准需求（语义铁律：相似度只当触发器）→
     标准需求命名 → opp_std_demands。
  3. 能力域：LLM 把标准需求分组到能力域（每簇一次调用）→ opp_demand_domains + 回填 domain_id。
  4. 覆盖率：doc_count = 标准需求覆盖的 distinct snap 数；coverage = doc_count/类内文档数；
     tier：core≥0.6 / ext≥0.4 / personal（阈值走 params：opp_cov_core/opp_cov_ext）。

状态：opp_clusters.normalize_status（init/running/done/failed + error 落 naming_evidence 旁路字段
normalize_error——ALTER 补列）。幂等：重跑先清该簇旧 atoms/std/domains 再算。
"""
import asyncio
import json
import logging

from .opp_common import get_db, new_id, rows_to_dicts, get_param
from .opp_mining import _cos, _centroid, demand_vec_id

logger = logging.getLogger("pipeline.opp_normalize")

# 客户端硬超时（秒）：llm_call 的 _timeout 只透传给治理端点，客户端 aiohttp 无超时——
# 上游挂起时客户端会无限等（实测 atomize 卡 6 分钟 atoms=0）。wait_for 快速失败+重试。
LLM_CLIENT_TIMEOUT = 120
LLM_RETRIES = 2


async def _llm(prompt, ctx, timeout=LLM_CLIENT_TIMEOUT):
    """llm_call 客户端超时包装：wait_for + 指数退避重试，全失败抛 RuntimeError。"""
    from pipeline_service.llm_bridge import llm_call
    last = None
    for a in range(LLM_RETRIES):
        try:
            return await asyncio.wait_for(
                llm_call(prompt, purpose="utility", timeout=timeout,
                         org_id=ctx.get("org_id") or "0",
                         user_id=ctx.get("user_id") or "",
                         session_id=ctx.get("session_id") or ""),
                timeout=timeout + 10)
        except Exception as e:
            last = e
            if a < LLM_RETRIES - 1:
                await asyncio.sleep(2 ** a)
    raise RuntimeError("LLM 调用失败(%d次): %s" % (LLM_RETRIES, str(last)[:150]))

P = {
    "opp_norm_atom_batch": "20",     # 原子化每批文档数
    "opp_norm_tau2": "0.85",         # 归一化相似度触发阈值
    "opp_norm_topk": "10",           # 原子 kNN 邻居数
    "opp_cov_core": "0.6",           # 核心共性阈值
    "opp_cov_ext": "0.4",            # 扩展共性阈值
}


async def get_norm_params(sor):
    cfg = {}
    for k, dv in P.items():
        cfg[k] = await get_param(sor, k, dv)
    for ik in ("opp_norm_atom_batch", "opp_norm_topk"):
        try:
            cfg[ik] = int(cfg[ik])
        except (TypeError, ValueError):
            cfg[ik] = int(P[ik])
    for fk in ("opp_norm_tau2", "opp_cov_core", "opp_cov_ext"):
        try:
            cfg[fk] = float(cfg[fk])
        except (TypeError, ValueError):
            cfg[fk] = float(P[fk])
    return cfg


async def _set_norm_status(sor, cluster_id, status, err=""):
    await sor.sqlExe(
        "UPDATE opp_clusters SET normalize_status=${s}$, normalize_error=${e}$ WHERE id=${c}$",
        {"s": status, "e": err[:500], "c": cluster_id})
    await sor.sqlExe("COMMIT", {})


async def _load_cluster(sor, cluster_id):
    recs = await sor.sqlExe(
        "SELECT c.*, b.org_id AS b_org FROM opp_clusters c "
        "LEFT JOIN opp_mining_batches b ON b.id=c.batch_id WHERE c.id=${c}$",
        {"c": cluster_id})
    await sor.sqlExe("COMMIT", {})
    if not recs:
        return None
    return rows_to_dicts(recs, limit=1)[0]


async def _cluster_snaps(sor, cluster_id):
    recs = await sor.sqlExe(
        "SELECT id, title, source, src_id FROM opp_demand_snap WHERE cluster_id=${c}$",
        {"c": cluster_id})
    await sor.sqlExe("COMMIT", {})
    return rows_to_dicts(recs, limit=5000)


# ══════════════════ 1. 原子化 ══════════════════

async def _atomize(sor, snaps, cfg, ctx):
    """LLM 批量提取原子需求。返回 [(snap_id, atom_text)]。"""
    from pipeline_service.llm_bridge import llm_call
    sem = asyncio.Semaphore(3)
    bs = cfg["opp_norm_atom_batch"]
    results = []
    lock = asyncio.Lock()

    async def _one(chunk):
        titles = "\n".join("%d. %s" % (i + 1, s["title"]) for i, s in enumerate(chunk))
        prompt = (
            "以下是同一类软件众包需求的标题（编号列表）。从每条标题提取它表达的**原子软件需求**"
            "（如「微信小程序开发」「考勤记录」「合同到期提醒」），一条标题可提取1-3个原子需求，"
            "无法判断的跳过。\n" + titles +
            "\n\n只输出 JSON 数组，元素为 {\"no\": 编号, \"atom\": \"原子需求文本\"}，"
            "不要解释、不要 markdown 代码块。")
        async with sem:
            try:
                raw = await _llm(prompt, ctx, timeout=90)
                raw = (raw or "").strip()
                if raw.startswith("```"):
                    raw = raw.strip("`")
                    if raw.lower().startswith("json"):
                        raw = raw[4:]
                arr = json.loads(raw)
                items = []
                for it in arr if isinstance(arr, list) else []:
                    if not isinstance(it, dict):
                        continue
                    no = it.get("no")
                    atom = str(it.get("atom") or "").strip()[:255]
                    # 容错：LLM 可能输出 "1"/1.0/True，统一 int() 转换后范围校验
                    try:
                        no_i = int(no)
                    except (TypeError, ValueError):
                        continue
                    if 1 <= no_i <= len(chunk) and atom:
                        items.append((chunk[no_i - 1]["id"], atom))
                async with lock:
                    results.extend(items)
            except Exception as e:
                logger.debug("atomize chunk failed: %s", e)
                async with lock:
                    results.extend([(s["id"], s["title"][:255]) for s in chunk])

    chunks = [snaps[i:i + bs] for i in range(0, len(snaps), bs)]
    await asyncio.gather(*(_one(ch) for ch in chunks))
    # 产出率兜底：原子数 < 文档数 50% 时，缺原子的文档用标题作原子
    # （众包标题本身即浓缩需求语义；保证覆盖率统计可算，2026-09-12 实测 46 文档仅 6 原子）
    covered = set(sid for sid, _ in results)
    if len(covered) < len(snaps) * 0.5:
        for s in snaps:
            if s["id"] not in covered:
                results.append((s["id"], s["title"][:255]))
    return results


# ══════════════════ 2. 归一化（embedding + LLM 裁决）══════════════════

async def _normalize(sor, atoms, project_id, cfg, ctx):
    """atoms=[(snap_id, atom_text)] → 标准需求组 [(name, [snap_ids], [atom_texts])]。"""
    if not atoms:
        return []
    texts = [a[1] for a in atoms]
    from pipeline_service import rag_client as rc
    vecs, err = await rc.rag_embed_texts(project_id, texts, batch_size=10)
    if err or not vecs or len(vecs) != len(texts):
        raise RuntimeError("原子向量化失败: %s" % (err or "数量不符"))

    # 簇内 kNN（VDB 临时 collection）
    base = await _vdb_base(sor)
    col = "opp_norm_%s" % new_id()[:12]
    ok, d = await _vdb_post(sor, base, "/v1/createcollection", {
        "colname": col, "fields": [
            {"name": "id", "type": "str", "is_primary": True, "max_length": 64},
            {"name": "vector", "type": "fvector", "dim": 1024},
            {"name": "text", "type": "str", "max_length": 500}],
        "description": "opp normalize workset", "metric": "COSINE"})
    if not ok and "exist" not in str(d).lower():
        raise RuntimeError("建归一化工作集失败: %s" % d)
    rows = [{"id": "a%d" % i, "vector": vecs[i], "text": texts[i][:490]}
            for i in range(len(texts))]
    for i in range(0, len(rows), 500):
        ok, d = await _vdb_post(sor, base, "/v1/upsert",
                                {"colname": col, "data": rows[i:i + 500]})
        if not ok:
            await _vdb_post(sor, base, "/v1/dropcollection", {"colname": col})
            raise RuntimeError("归一化 upsert 失败: %s" % d)
    nbr = {}
    for i in range(len(texts)):
        ok, d = await _vdb_post(sor, base, "/v1/query", {
            "colname": col, "vector": vecs[i], "pagerows": cfg["opp_norm_topk"],
            "output_fields": ["id"]})
        if not ok:
            await _vdb_post(sor, base, "/v1/dropcollection", {"colname": col})
            raise RuntimeError("归一化 kNN 失败: %s" % d)
        rr = (d.get("data") or {}).get("rows") or []
        nbr["a%d" % i] = [(float(r.get("score", 0) or 0), r.get("id"))
                          for r in rr if r.get("id") and r.get("id") != "a%d" % i]
    await _vdb_post(sor, base, "/v1/dropcollection", {"colname": col})

    # τ2 并查集得候选组
    ids = ["a%d" % i for i in range(len(texts))]
    groups = _union_find_groups(ids, nbr, cfg["opp_norm_tau2"])

    # LLM 裁决 + 命名（每组一次，并发 3）
    from pipeline_service.llm_bridge import llm_call
    sem = asyncio.Semaphore(3)
    out = []
    lock = asyncio.Lock()

    async def _judge(g):
        members = [int(x[1:]) for x in g]
        uniq = []
        seen = set()
        for m in members:
            if texts[m] not in seen:
                seen.add(texts[m])
                uniq.append(texts[m])
        if len(uniq) == 1:
            name, keep = uniq[0], members
        else:
            prompt = (
                "以下是候选的同类原子软件需求（去重后）。判断它们是否表达**同一个标准软件需求**"
                "（语义相同/包含关系算同一；明显不同功能不算）。\n" +
                "\n".join("- " + u for u in uniq[:12]) +
                "\n\n若同一：输出 JSON {\"same\": true, \"name\": \"标准需求名(≤12字)\"}；"
                "若不同：输出 {\"same\": false, \"split\": [[编号...], [编号...]]} 给出应拆开的分组"
                "（编号从1起，对应上面列表顺序）。只输出 JSON。")
            async with sem:
                try:
                    raw = await _llm(prompt, ctx, timeout=90)
                    raw = (raw or "").strip().strip("`")
                    if raw.lower().startswith("json"):
                        raw = raw[4:]
                    d2 = json.loads(raw)
                except Exception as e:
                    logger.debug("judge failed, keep group: %s", e)
                    d2 = {"same": True, "name": uniq[0][:40]}
            if d2.get("same"):
                name = str(d2.get("name") or uniq[0]).strip()[:128]
                keep = members
            else:
                # 拆分：每个子组单独成标准需求，命名用子组首条
                async with lock:
                    for sub in (d2.get("split") or []):
                        idxs = [members[i - 1] for i in sub
                                if isinstance(i, int) and 1 <= i <= len(members)]
                        if idxs:
                            out.append((uniq[0][:128] if len(idxs) == 1 else
                                        texts[idxs[0]][:128], idxs,
                                        [texts[i] for i in idxs]))
                return
        async with lock:
            out.append((name, keep, [texts[m] for m in keep]))

    await asyncio.gather(*(_judge(g) for g in groups))
    return out


def _union_find_groups(ids, nbr, tau):
    parent = {v: v for v in ids}

    def find(x):
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    for v in ids:
        for sc, nid in nbr.get(v, []):
            if sc >= tau and nid in parent:
                ra, rb = find(v), find(nid)
                if ra != rb:
                    parent[rb] = ra
    comps = {}
    for v in ids:
        comps.setdefault(find(v), []).append(v)
    return sorted(comps.values(), key=len, reverse=True)


async def _vdb_base(sor):
    recs = await sor.sqlExe("SELECT baseurl FROM upapp WHERE id='rag-vdb'", {})
    await sor.sqlExe("COMMIT", {})
    return (recs[0].baseurl or "").rstrip("/") if recs else ""


async def _vdb_post(sor, base, path, payload, timeout=40):
    from .opp_mining import vdb_post
    return await vdb_post(sor, base, path, payload, timeout=timeout)


# ══════════════════ 3. 能力域 ══════════════════

async def _domains(sor, cluster_id, org_id, std_rows, ctx):
    """LLM 把标准需求分组到能力域。std_rows=[{id,name}]。返回 [(domain_name, [std_ids])]。"""
    if not std_rows:
        return []
    from pipeline_service.llm_bridge import llm_call
    names = "\n".join("%d. %s" % (i + 1, s["name"]) for i, s in enumerate(std_rows))
    prompt = (
        "以下是某类软件的标准需求清单。把它们归入 2-6 个**能力域**"
        "（如「人员管理」「工时与考勤」「合同管理」「报表统计」）。\n" + names +
        "\n\n只输出 JSON 数组 [{\"domain\": \"域名\", \"nos\": [编号...]}]，覆盖全部编号，不解释。")
    try:
        raw = await _llm(prompt, ctx, timeout=90)
        raw = (raw or "").strip().strip("`")
        if raw.lower().startswith("json"):
            raw = raw[4:]
        arr = json.loads(raw)
    except Exception as e:
        logger.debug("domains failed: %s", e)
        return [("通用能力", [s["id"] for s in std_rows])]
    out = []
    for g in arr if isinstance(arr, list) else []:
        nos = g.get("nos") or []
        ids = [std_rows[i - 1]["id"] for i in nos
               if isinstance(i, int) and 1 <= i <= len(std_rows)]
        if ids:
            out.append((str(g.get("domain") or "未命名域")[:128], ids))
    covered = set(i for _, ids in out for i in ids)
    left = [s["id"] for s in std_rows if s["id"] not in covered]
    if left:
        out.append(("其他能力", left))
    return out


# ══════════════════ 主流程 ══════════════════

async def normalize_cluster(sor, cluster_id, ctx):
    """对单个类别跑 原子化→归一化→能力域→覆盖率。返回 (ok, msg)。"""
    cl = await _load_cluster(sor, cluster_id)
    if not cl:
        return False, "类别不存在"
    org_id = str(cl.get("org_id") or cl.get("b_org") or "")
    if org_id and org_id != (ctx.get("org_id") or ""):
        return False, "类别不属于当前机构"
    cfg = await get_norm_params(sor)
    snaps = await _cluster_snaps(sor, cluster_id)
    if not snaps:
        return False, "类别内无需求快照"
    project_id = ctx.get("project_id") or ""

    await _set_norm_status(sor, cluster_id, "running")
    try:
        # 幂等：清旧结果
        for tbl in ("opp_demand_atoms", "opp_std_demands", "opp_demand_domains"):
            await sor.sqlExe("DELETE FROM %s WHERE cluster_id=${c}$" % tbl,
                             {"c": cluster_id})
        await sor.sqlExe("COMMIT", {})

        # 1 原子化
        atoms = await _atomize(sor, snaps, cfg, ctx)
        if not atoms:
            await _set_norm_status(sor, cluster_id, "failed", "原子化无结果(LLM全失败)")
            return False, "原子化无结果"
        for sid, atom in atoms:
            await sor.C("opp_demand_atoms", {
                "id": new_id(), "cluster_id": cluster_id, "org_id": org_id,
                "snap_id": sid, "atom_text": atom})
        await sor.sqlExe("COMMIT", {})

        # 2 归一化
        groups = await _normalize(sor, atoms, project_id, cfg, ctx)
        snap_by_id = {s["id"]: s for s in snaps}
        std_rows = []
        for name, members, atom_texts in groups:
            sid_ = new_id()
            snap_ids = sorted({atoms[m][0] for m in members})
            await sor.C("opp_std_demands", {
                "id": sid_, "cluster_id": cluster_id, "org_id": org_id,
                "name": name[:128], "doc_count": len(snap_ids),
                "coverage": 0.0, "tier": "personal",
                "evidence_json": json.dumps(
                    {"atoms": atom_texts[:10], "snaps": snap_ids[:50]}, ensure_ascii=False)})
            std_rows.append({"id": sid_, "name": name, "snap_ids": snap_ids})
            # 回填原子的 std 归属
            for m in members:
                await sor.sqlExe(
                    "UPDATE opp_demand_atoms SET std_demand_id=${s}$ "
                    "WHERE cluster_id=${c}$ AND snap_id=${sn}$ AND atom_text=${a}$",
                    {"s": sid_, "c": cluster_id, "sn": atoms[m][0], "a": atoms[m][1]})
        await sor.sqlExe("COMMIT", {})

        # 4 覆盖率（先算再写 tier）
        n_docs = len(snaps)
        for s in std_rows:
            cov = round(len(s["snap_ids"]) / n_docs, 4) if n_docs else 0.0
            tier = "core" if cov >= cfg["opp_cov_core"] else (
                "ext" if cov >= cfg["opp_cov_ext"] else "personal")
            await sor.sqlExe(
                "UPDATE opp_std_demands SET coverage=${v}$, tier=${t}$ WHERE id=${i}$",
                {"v": cov, "t": tier, "i": s["id"]})
        await sor.sqlExe("COMMIT", {})

        # 3 能力域
        doms = await _domains(sor, cluster_id, org_id,
                              [{"id": s["id"], "name": s["name"]} for s in std_rows], ctx)
        for dname, std_ids in doms:
            did = new_id()
            await sor.C("opp_demand_domains", {
                "id": did, "cluster_id": cluster_id, "org_id": org_id,
                "parent_id": "", "name": dname, "std_count": len(std_ids)})
            for i in range(0, len(std_ids), 200):
                chunk = std_ids[i:i + 200]
                ph = ",".join("${d%d}$" % j for j in range(len(chunk)))
                args = {"did": did}
                for j, sv in enumerate(chunk):
                    args["d%d" % j] = sv
                await sor.sqlExe(
                    "UPDATE opp_std_demands SET domain_id=${did}$ WHERE id IN (%s)" % ph, args)
        await sor.sqlExe("COMMIT", {})

        await _set_norm_status(sor, cluster_id, "done")
        core = sum(1 for s in std_rows
                   if len(s["snap_ids"]) / n_docs >= cfg["opp_cov_core"]) if n_docs else 0
        return True, "归一化完成：原子 %d 条 → 标准需求 %d 个（核心 %d）→ 能力域 %d 个" % (
            len(atoms), len(std_rows), core, len(doms))
    except Exception as e:
        logger.exception("normalize cluster %s failed", cluster_id)
        await _set_norm_status(sor, cluster_id, "failed", str(e))
        return False, "归一化失败: %s" % str(e)[:300]


async def coverage_report(sor, ctx, cluster_id):
    """覆盖率报告（org 隔离）。返回 (ok, dict|err)。"""
    cl = await _load_cluster(sor, cluster_id)
    if not cl:
        return False, "类别不存在"
    org_id = str(cl.get("org_id") or cl.get("b_org") or "")
    if org_id and org_id != (ctx.get("org_id") or ""):
        return False, "类别不属于当前机构"
    recs = await sor.sqlExe(
        "SELECT id, name, doc_count, coverage, tier, domain_id FROM opp_std_demands "
        "WHERE cluster_id=${c}$ ORDER BY coverage DESC", {"c": cluster_id})
    await sor.sqlExe("COMMIT", {})
    stds = rows_to_dicts(recs, limit=200)
    drecs = await sor.sqlExe(
        "SELECT id, name, std_count FROM opp_demand_domains WHERE cluster_id=${c}$",
        {"c": cluster_id})
    await sor.sqlExe("COMMIT", {})
    doms = rows_to_dicts(drecs, limit=50)
    dom_name = {d["id"]: d["name"] for d in doms}
    for s in stds:
        s["domain"] = dom_name.get(s.get("domain_id") or "", "")
    tiers = {}
    for s in stds:
        tiers[s["tier"]] = tiers.get(s["tier"], 0) + 1
    return True, {
        "cluster": {"id": cluster_id, "name": cl.get("name"),
                    "doc_count": cl.get("doc_count"),
                    "normalize_status": cl.get("normalize_status"),
                    "normalize_error": cl.get("normalize_error") or ""},
        "tiers": tiers, "std_demands": stds, "domains": doms,
    }