# -*- coding:utf-8 -*-
"""商机产线需求挖掘能力（P1）：众包需求 → 语义聚类 → TopX/指定类型。

分层边界（pipeline-extension-patterns）：
  · embedding 走 pipeline_service.rag_client.rag_embed_texts（凭据单点在 rag）。
  · 向量近邻全部压给 VDB(Milvus)（upapp.rag-vdb），产线侧零向量计算。
  · 聚类 = kNN + τ阈值 + 纯 Python 并查集（无原生依赖，Nuitka 友好，确定性可复现）。
  · LLM 只做簇命名（purpose=utility），失败规则兜底不崩。

机构隔离（用户定夺 2026-09-11）：
  · 基础数据共享——爬虫需求 + embedding 缓存 collection（opp_demand_emb_cache）全机构共用，
    一条需求只嵌一次，跨机构/跨批次复用，成本摊薄。
  · 分析结果隔离——批次/快照/类别全部挂 org_id，所有查询按 ctx.org_id 强制过滤。

VDB 协议（实测，详见 rag-module-operations 技能）：
  createcollection/upsert/query(kNN=vector+pagerows)/expr标量过滤/dropcollection；
  kNN+expr 组合可用；output_fields 可取回 vector（缓存→工作集拷贝）。
"""
import asyncio
import hashlib
import json
import logging

from .opp_common import get_db, new_id, rows_to_dicts, get_param, get_crawler_config

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

DIM = 1024
# 共享 embedding 缓存 collection（基础数据，全机构共用）
CACHE_COL_DEFAULT = "opp_demand_emb_cache"
# 参数默认兜底（params 表优先，禁硬编码为唯一来源）
P = {
    "opp_mine_tau": "0.75",
    "opp_mine_topk": "30",
    "opp_mine_min_cluster": "5",
    "opp_mine_big_split": "150",     # 超过此规模的簇质心二次细分
    "opp_mine_keep_batches": "3",    # 每机构保留最近 N 个批次工作集 collection
    "opp_embed_batch": "10",         # rag /embed 单批上限（实测 dashscope 兼容 10）
}

# 状态机（非法迁移拒绝，对齐 REPORT_TRANSITIONS 范式）
BATCH_TRANSITIONS = {
    "pulling": {"embedding", "failed", "done"},
    "embedding": {"clustering", "failed"},
    "clustering": {"naming", "failed"},
    "naming": {"done", "failed"},
    "done": set(),
    "failed": {"pulling"},   # 失败可重跑
}


def can_batch_transition(cur, nxt):
    return nxt in BATCH_TRANSITIONS.get(cur, set())


# ══════════════════ 参数 / VDB 客户端 ══════════════════

async def get_mining_params(sor):
    cfg = {}
    for k, dv in P.items():
        cfg[k] = await get_param(sor, k, dv)
    cache_col = await get_param(sor, "opp_emb_cache_col", CACHE_COL_DEFAULT)
    cfg["cache_col"] = cache_col
    for ik in ("opp_mine_topk", "opp_mine_min_cluster", "opp_mine_big_split",
               "opp_mine_keep_batches", "opp_embed_batch"):
        try:
            cfg[ik] = int(cfg[ik])
        except (TypeError, ValueError):
            cfg[ik] = int(P[ik])
    try:
        cfg["opp_mine_tau"] = float(cfg["opp_mine_tau"])
    except (TypeError, ValueError):
        cfg["opp_mine_tau"] = float(P["opp_mine_tau"])
    return cfg


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):
    """VDB REST 调用（aiohttp）。返回 (ok, dict|错误串)。失败显式报错，禁静默。"""
    try:
        import aiohttp
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as s:
            async with s.post(base + path, json=payload) as r:
                txt = await r.text()
        try:
            d = json.loads(txt) if txt.strip().startswith("{") else {}
        except Exception:
            d = {}
        if r.status != 200 or (d.get("status") and d.get("status") != "SUCCEEDED"):
            return False, "VDB %s HTTP%d %s" % (path, r.status, (d.get("error") or txt)[:200])
        return True, d
    except Exception as e:
        return False, "VDB %s 调用失败: %s" % (path, str(e)[:160])


def _vec_rows(d):
    """解析 VDB 响应的 rows（兼容嵌套 data.rows）。"""
    data = d.get("data") if isinstance(d, dict) else None
    if isinstance(data, dict):
        return data.get("rows") or []
    if isinstance(data, list):
        return data
    return d.get("rows") or [] if isinstance(d, dict) else []


def demand_vec_id(source, src_id):
    """需求→稳定向量ID（sha1，≤32字符，可复算幂等）。"""
    return hashlib.sha1(("%s:%s" % (source or "", src_id or "")).encode("utf-8")).hexdigest()[:32]


# ══════════════════ 数据层：快照拉取 ══════════════════

async def pull_demands(sor, batch, ctx, cfg, scope, keyword="", category="",
                       days=365, sources="", limit=200):
    """从爬虫平台拉需求 → 写 opp_demand_snap（按 batch+src_id 去重）。返回 (n, err)。"""
    base, token = await get_crawler_config(sor)
    if not token:
        return 0, "params 缺 tender_api_token（爬虫平台接入未配置）"
    org_id = ctx.get("org_id") or ""
    bid = batch["id"]
    offset = 0
    total = 0
    while True:
        params = {"days": days, "limit": limit, "offset": offset, "record_type": "demand"}
        if keyword:
            params["keyword"] = keyword
        if sources:
            params["source"] = sources
        # 复用爬虫只读端点
        from .opp_common import crawler_get
        ok, res = await crawler_get(sor, "/api/demands", params)
        if not ok:
            return total, "拉取需求失败@offset%d: %s" % (offset, res)
        items = (res or {}).get("items") or []
        matched = (res or {}).get("matched", 0)
        for it in items:
            src_id = str(it.get("id") or it.get("source_id") or "")
            title = (it.get("title") or "").strip()
            if not src_id or not title:
                continue
            # 指定类型分析：category 二次过滤（爬虫 item_category 模糊）
            if category and category not in (it.get("item_category") or "") and category not in title:
                continue
            snap_id = new_id()
            vec_id = demand_vec_id(it.get("source"), src_id)
            await sor.C("opp_demand_snap", {
                "id": snap_id, "batch_id": bid, "org_id": org_id,
                "src_id": src_id, "source": it.get("source") or "",
                "title": title[:500], "item_category": (it.get("item_category") or "")[:128],
                "budget_wan": it.get("budget_wan") or 0, "url": (it.get("url") or "")[:500],
                "publish_time": (it.get("publish_time") or "")[:32],
                "embed_status": "new",
            })
            await sor.sqlExe("COMMIT", {})
            # vec_id 暂存到内存映射稍后批量用；这里用 title 去重键
            total += 1
        offset += len(items)
        if not items or offset >= matched:
            break
    return total, ""


# ══════════════════ 数据层：embedding 增量缓存 ══════════════════

async def _ensure_cache_collection(sor, base, col):
    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": DIM},
            {"name": "text", "type": "str", "max_length": 2000}],
        "description": "opp demand embedding cache", "metric": "COSINE"})
    if not ok and "exist" not in str(d).lower():
        return False, d
    return True, d


async def _cache_existing_ids(sor, base, col, ids):
    """批量查缓存 collection 已存在哪些 id（expr in，每批≤200）。返回 set。"""
    found = set()
    for i in range(0, len(ids), 200):
        chunk = ids[i:i + 200]
        expr = "id in [%s]" % ",".join('"%s"' % x for x in chunk)
        ok, d = await vdb_post(sor, base, "/v1/query",
                               {"colname": col, "expr": expr, "output_fields": ["id"]})
        if ok:
            for r in _vec_rows(d):
                if r.get("id"):
                    found.add(r["id"])
    return found


async def embed_snapshots(sor, batch, project_id, cfg, on_progress=None):
    """增量 embedding：缓存缺的才调 rag_embed_texts，upsert 进共享缓存。
    返回 (vec_id→vector dict, err)。一条需求只嵌一次（全机构复用）。"""
    bid = batch["id"]
    base = await _vdb_base(sor)
    if not base:
        return None, "upapp 缺 rag-vdb（向量库未配置）"
    ok, d = await _ensure_cache_collection(sor, base, cfg["cache_col"])
    if not ok:
        return None, "建缓存 collection 失败: %s" % d

    snaps = rows_to_dicts(await sor.sqlExe(
        "SELECT id, source, src_id, title FROM opp_demand_snap WHERE batch_id=${b}$",
        {"b": bid}), limit=100000)
    await sor.sqlExe("COMMIT", {})
    if not snaps:
        return None, "批次无需求快照"

    # 稳定 vec_id + 文本
    id_map = {}     # snap_id -> vec_id
    text_map = {}   # vec_id -> text
    for s in snaps:
        vid = demand_vec_id(s["source"], s["src_id"])
        id_map[s["id"]] = vid
        text_map[vid] = (s["title"] or "(无标题)")[:1900]
    all_vids = list(set(id_map.values()))

    # 查缓存已有的
    cached = await _cache_existing_ids(sor, base, cfg["cache_col"], all_vids)
    to_embed = [v for v in all_vids if v not in cached]
    logger.debug("embed batch=%s total=%d cached=%d new=%d", bid, len(all_vids), len(cached), len(to_embed))

    # 缺的调 rag_embed_texts
    vecs = {}
    if to_embed:
        from pipeline_service import rag_client as rc
        bs = cfg["opp_embed_batch"]
        texts_to_embed = [text_map[v] for v in to_embed]
        got, err = await rc.rag_embed_texts(project_id, texts_to_embed, batch_size=bs)
        if err:
            return None, "embedding 失败: %s" % err
        if len(got) != len(to_embed):
            return None, "embedding 数量不符(%d/%d)" % (len(got), len(to_embed))
        # upsert 进缓存
        rows = [{"id": to_embed[i], "vector": got[i], "text": texts_to_embed[i]}
                for i in range(len(to_embed))]
        for i in range(0, len(rows), 500):
            ok, d = await vdb_post(sor, base, "/v1/upsert",
                                   {"colname": cfg["cache_col"], "data": rows[i:i + 500]})
            if not ok:
                return None, "缓存 upsert 失败: %s" % d
        for i, v in enumerate(to_embed):
            vecs[v] = got[i]

    # 缓存命中的也要取回向量（构建工作集用）
    if cached:
        cached = list(cached)
        for i in range(0, len(cached), 200):
            chunk = cached[i:i + 200]
            expr = "id in [%s]" % ",".join('"%s"' % x for x in chunk)
            ok, d = await vdb_post(sor, base, "/v1/query",
                                   {"colname": cfg["cache_col"], "expr": expr,
                                    "output_fields": ["id", "vector"]})
            if ok:
                for r in _vec_rows(d):
                    if r.get("id") and isinstance(r.get("vector"), list):
                        vecs[r["id"]] = r["vector"]
    # 更新快照 embed_status
    for sid, vid in id_map.items():
        st = "cached" if vid in vecs else "failed"
        await sor.sqlExe("UPDATE opp_demand_snap SET embed_status=${s}$ WHERE id=${i}$",
                         {"s": st, "i": sid})
    await sor.sqlExe("COMMIT", {})
    return {"id_map": id_map, "text_map": text_map, "vecs": vecs}, ""


# ══════════════════ 数据层：批次工作集 + kNN ══════════════════

async def build_batch_collection(sor, base, batch, cfg, embed_res):
    """从缓存拷向量到批次专属工作集 collection（聚类只在本批内算）。返回 (col, err)。"""
    col = "opp_mine_%s" % (batch["id"][:16].replace("-", ""))
    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": DIM},
            {"name": "text", "type": "str", "max_length": 2000},
            {"name": "snap_id", "type": "str", "max_length": 32}],
        "description": "opp mining batch workset", "metric": "COSINE"})
    if not ok and "exist" not in str(d).lower():
        return None, "建批次 collection 失败: %s" % d
    vecs = embed_res["vecs"]
    text_map = embed_res["text_map"]
    id_map = embed_res["id_map"]   # snap_id -> vec_id
    rows = []
    for sid, vid in id_map.items():
        if vid in vecs:
            rows.append({"id": vid, "vector": vecs[vid], "text": text_map.get(vid, ""), "snap_id": sid})
    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:
            return None, "工作集 upsert 失败: %s" % d
    return col, ""


async def knn_graph(sor, base, col, vecs, topk):
    """每条向量查 topk 邻居（排除自身）。返回 vid -> [(score, vid)]。"""
    nbr = {}
    ids = list(vecs.keys())
    for vid in ids:
        ok, d = await vdb_post(sor, base, "/v1/query", {
            "colname": col, "vector": vecs[vid], "pagerows": topk,
            "output_fields": ["id"]})
        if not ok:
            return None, "kNN 失败@%s: %s" % (vid, d)
        rows = _vec_rows(d)
        nbr[vid] = [(float(r.get("score", 0) or 0), r.get("id"))
                    for r in rows if r.get("id") and r.get("id") != vid]
    return nbr, ""


# ══════════════════ 算法层：并查集聚类 ══════════════════

def _union_find(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

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


def _cos(a, b):
    return sum(x * y for x, y in zip(a, b))


def _centroid(vecs, members):
    n = len(members)
    return [sum(vecs[m][k] for m in members) / n for k in range(len(vecs[members[0]]))]


def cluster_vectors(vecs, nbr, tau, min_cluster, big_split):
    """kNN 图 → τ 并查集 → 大簇二次细分 → 小组质心并入。返回 (final_clusters, other_ids)。
    final_clusters: list[list[vid]]，other_ids: 未入簇的 vid。"""
    ids = list(vecs.keys())
    comps = _union_find(ids, nbr, tau)
    # 超大簇质心二次细分（众包标题模板化重复会过度聚合）
    pre = []
    for c in comps:
        if len(c) > big_split:
            members = set(c)
            sub_nbr = {v: [(sc, n) for sc, n in nbr.get(v, []) if n in members] for v in c}
            pre.extend(_union_find(c, sub_nbr, min(tau + 0.10, 0.95)))
        else:
            pre.append(c)
    big = [c for c in pre if len(c) >= min_cluster]
    small = [c for c in pre if len(c) < min_cluster]
    # 小组并入最近大簇（质心相似度 ≥ τ-0.10），否则归其他
    other = []
    if big and small:
        cents = [(c, _centroid(vecs, c)) for c in big]
        for sc_ in small:
            cc = _centroid(vecs, sc_)
            best, bs = None, -1
            for bc, bcent in cents:
                sim = _cos(cc, bcent)
                if sim > bs:
                    best, bs = bc, sim
            if best is not None and bs >= tau - 0.10:
                best.extend(sc_)
            else:
                other.extend(sc_)
    else:
        for c in small:
            other.extend(c)
    big.sort(key=len, reverse=True)
    return big, other


# ══════════════════ 算法层：LLM 簇命名（规则兜底）══════════════════

async def name_clusters(sor, batch, clusters, vecs, text_map, cfg, ctx):
    """为每个簇命名：LLM(utility) 并发调用 + 失败/异常规则兜底。
    返回 list[{name, members, centroid_snap, samples, evidence}]。

    性能实测教训（2026-09-11）：42 簇串行 LLM 命名 540s 占批次全程 551s——
    改并发（Semaphore 限 5，防上游限流）后命名阶段应降至 ~1/5。
    """
    total = len(vecs)
    # 阶段1（纯CPU，串行）：算质心/样例/排名
    prepared = []
    for ci, members in enumerate(clusters):
        cent = _centroid(vecs, members)
        sims = sorted(((_cos(vecs[m], cent), m) for m in members), reverse=True)
        samples = []
        seen = set()
        for _, m in sims:
            t = text_map.get(m, "")
            if t and t not in seen:
                seen.add(t)
                samples.append(t)
            if len(samples) >= 8:
                break
        prepared.append({
            "rank": ci + 1, "members": members, "size": len(members),
            "share": round(len(members) / total, 4) if total else 0,
            "centroid_vid": sims[0][1] if sims else "",
            "samples": samples[:8], "name": "", "evidence": "rule",
        })

    # 阶段2：LLM 命名（并发限5；单簇失败不影响整体，走规则兜底）
    from pipeline_service.llm_bridge import llm_call
    sem = asyncio.Semaphore(5)

    async def _name_one(item):
        async with sem:
            try:
                prompt = (
                    "以下是同一类软件众包需求的标题样例（已去重）：\n" +
                    "\n".join("- " + s for s in item["samples"][:8]) +
                    "\n\n请用不超过12个汉字给这一类需求起一个简洁准确的类别名（如「微信小程序开发」"
                    "「企业网站定制」「AI短视频制作」），只输出类别名本身，不要解释、不要标点。")
                raw = await llm_call(prompt, purpose="utility", timeout=60,
                                     org_id=ctx.get("org_id") or "0",
                                     user_id=ctx.get("user_id") or "",
                                     session_id=ctx.get("session_id") or "")
                name = (raw or "").strip().splitlines()[0][:40] if raw else ""
                # LLM 可能输出带引号/序号，清洗
                name = name.strip("「」\"'。. 　")
                if name:
                    item["name"] = name
                    item["evidence"] = "llm"
            except Exception as e:
                logger.debug("LLM 命名失败,走规则兜底: %s", e)

    await asyncio.gather(*(_name_one(it) for it in prepared))
    # 阶段3：规则兜底
    for it in prepared:
        if not it["name"]:
            it["name"] = _rule_name(it["samples"])
    return prepared


def _rule_name(samples):
    """规则兜底：取样例标题的最长公共前缀片段或首条去模板词。"""
    if not samples:
        return "未命名类别"
    # 去常见模板前缀
    strip = ["我需要", "需要", "其他", "服务需求", "服务采购", "服务", "需求", "采购", "合作"]
    def clean(t):
        for s in strip:
            t = t.replace(s, "")
        return t.strip()
    cleaned = [clean(s) for s in samples if clean(s)]
    if not cleaned:
        return samples[0][:20]
    # 取最高频的 2-gram 词
    from collections import Counter
    cnt = Counter()
    for t in cleaned:
        for n in (4, 3, 2):
            for i in range(len(t) - n + 1):
                cnt[t[i:i + n]] += 1
    if cnt:
        top = cnt.most_common(1)[0][0]
        return top[:20]
    return cleaned[0][:20]