# -*- coding:utf-8 -*-
"""P1 聚类算法验证：真实众包需求 → embedding → VDB kNN → 并查集聚类 → 簇质量报告。

在 pipeline-app 测试机执行：./py3/bin/python /tmp/p1_cluster_validate.py
链路（与生产工具同协议，全部实测过的端点）：
  1. 爬虫平台 /api/demands 分页拉全量需求（params 表读 base/token）
  2. rag /rag/api/embed.dspy 批量向量化（HTTP 会话，admin 登录拿 cookie）
  3. VDB(Milvus) 测试 collection：upsert + /v1/query kNN(pagerows=30)
  4. 阈值 τ → 并查集连通分量 → 簇；小组并入最近大簇或归"其他"
  5. 输出簇统计 + 每簇 top10 样例标题（人工审聚类质量）
结果写 /tmp/p1_cluster_result.json。任何一步失败如实报错退出，不静默。
"""
import asyncio
import json
import os
import sys
import time

WORKDIR = "/d/pipeline/pipeline-app"
os.chdir(WORKDIR)
sys.path.insert(0, WORKDIR)

from appPublic.folderUtils import ProgramPath          # noqa: E402
from appPublic.jsonConfig import getConfig             # noqa: E402
from appPublic.event_dispatcher import EventDispatcher # noqa: E402
from sqlor.dbpools import DBPools                      # noqa: E402
from ahserver.serverenv import ServerEnv               # noqa: E402

p = ProgramPath()
config = getConfig(WORKDIR, NS={'workdir': WORKDIR, 'ProgramPath': p})
DBPools(config.databases)
se = ServerEnv()
se.event_dispatcher = EventDispatcher()
se.get_module_dbname = lambda m: 'pipeline'

TAU = float(os.environ.get("P1_TAU", "0.75"))       # 相似度阈值（可调参对比）
TOPK = 30                                            # kNN 邻居数
MIN_CLUSTER = 5                                      # 最小簇规模
TEST_COL = "opp_p1_validate_%d" % int(time.time())
DIM = 1024
APP_PORT = 9090


async def _sql(sql, args=None):
    async with DBPools().sqlorContext("pipeline") as sor:
        recs = await sor.sqlExe(sql, args or {})
        await sor.sqlExe("COMMIT", {})
    return recs or []


class Progress:
    def __init__(self, total, label):
        self.total, self.label, self.n, self.t0 = total, label, 0, time.time()

    def tick(self, k=1):
        self.n += k
        if self.n % 50 == 0 or self.n >= self.total:
            el = time.time() - self.t0
            print("  %s %d/%d (%.0fs)" % (self.label, self.n, self.total, el), flush=True)


async def main():
    import aiohttp
    timeout = aiohttp.ClientTimeout(total=60)

    # ── 1. 拉全量需求 ──
    recs = await _sql("SELECT params_name, params_value FROM params WHERE params_name IN ('tender_api_base','tender_api_token')")
    pm = {r.params_name: r.params_value for r in recs}
    base = (pm.get("tender_api_base") or "http://192.168.16.2:9085").rstrip("/")
    token = pm.get("tender_api_token") or ""
    if not token:
        print("FATAL: params 无 tender_api_token")
        sys.exit(1)
    hdr = {"X-API-Token": token}

    demands = []
    offset = 0
    async with aiohttp.ClientSession(timeout=timeout) as s:
        while True:
            async with s.get(base + "/api/demands", headers=hdr,
                             params={"days": 3650, "limit": 200, "offset": offset}) as r:
                if r.status != 200:
                    print("FATAL: /api/demands HTTP", r.status, await r.text())
                    sys.exit(1)
                d = await r.json()
            items = d.get("items") or []
            demands.extend(items)
            offset += len(items)
            if not items or offset >= d.get("matched", 0):
                break
    print("1. 需求快照拉取: %d 条" % len(demands))
    if not demands:
        print("FATAL: 无需求数据")
        sys.exit(1)

    # ── 2. embedding（生产同款通道：rag_client Bearer → /rag/api/embed.dspy）──
    # 坑实录：独立脚本走 cookie 会话必 401——登录 cookie 带 Secure 标志，aiohttp
    # 客户端拒绝经明文 http:// 发送；生产路径本来就是 Bearer token，不受此限。
    # 独立脚本需 initEnv() 注入 password_decode 等宿主全局函数（dapi 发 key 依赖）。
    from ahserver.globalEnv import initEnv
    initEnv()
    from pipeline_service import rag_client as rc
    _pid = os.environ.get("P1_PROJECT_ID", "")
    if not _pid:
        print("FATAL: 缺环境变量 P1_PROJECT_ID（真实商机项目id，用于 owner→Bearer key）")
        sys.exit(1)
    texts = [(it.get("title") or "").strip() for it in demands]
    texts = [t if t else "(无标题)" for t in texts]
    vectors = [None] * len(texts)
    pr = Progress(len(texts), "embedding")
    GROUP = 100
    for i in range(0, len(texts), GROUP):
        chunk = texts[i:i + GROUP]
        vs = None
        for attempt in range(3):
            vs, err = await rc.rag_embed_texts(_pid, chunk, batch_size=10)
            if not err and vs:
                break
            vs = None
            if attempt == 2:
                print("FATAL: embed 失败@%d-%d: %s" % (i, i + len(chunk) - 1, err))
                sys.exit(1)
            await asyncio.sleep(2 ** attempt)
        vectors[i:i + len(chunk)] = vs
        pr.tick(len(chunk))
    assert all(v is not None for v in vectors), "存在未嵌入条目"
    print("2. embedding 完成: %d 条 × %d 维" % (len(vectors), len(vectors[0])))

    # ── 3. VDB 常驻缓存 collection：upsert + kNN ──
    async with aiohttp.ClientSession(timeout=timeout) as s:
        async def vpost(path, payload):
            r = await s.post(vdb_base + path, json=payload)
            return r.status, await r.text()
        recs = await _sql("SELECT baseurl FROM upapp WHERE id='rag-vdb'")
        vdb_base = (recs[0].baseurl or "").rstrip("/")
        if not vdb_base:
            print("FATAL: upapp 无 rag-vdb")
            sys.exit(1)

        st, body = await vpost("/v1/createcollection", {"colname": TEST_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": "batch_id", "type": "str", "max_length": 64}],
            "description": "P1 validate", "metric": "COSINE"})
        if "SUCCEEDED" not in body:
            print("FATAL: createcollection", body[:200])
            sys.exit(1)

        # upsert 分批 500
        ids = ["d%d" % i for i in range(len(vectors))]
        for i in range(0, len(vectors), 500):
            rows = [{"id": ids[j], "vector": vectors[j], "text": texts[j][:1900], "batch_id": "p1v"}
                    for j in range(i, min(i + 500, len(vectors)))]
            st, body = await vpost("/v1/upsert", {"colname": TEST_COL, "data": rows})
            if "SUCCEEDED" not in body:
                print("FATAL: upsert@%d" % i, body[:200])
                sys.exit(1)
        print("3a. VDB upsert 完成")

        # kNN：每条查 TOPK 邻居
        nbr = {}   # id -> [(score, nid)]
        pr = Progress(len(ids), "kNN")
        for i, vid in enumerate(ids):
            st, body = await vpost("/v1/query", {
                "colname": TEST_COL, "vector": vectors[i], "pagerows": TOPK,
                "output_fields": ["id"]})
            d = json.loads(body)
            rows = (d.get("data") or {}).get("rows") or []
            nbr[vid] = [(float(r.get("score", 0)), r.get("id")) for r in rows if r.get("id") != vid]
            pr.tick()
        print("3b. kNN 完成")

        await vpost("/v1/dropcollection", {"colname": TEST_COL})
        print("3c. 测试 collection 已清理")

    # ── 4. 并查集 ──
    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

    edges = 0
    for vid in ids:
        for sc, nid in nbr[vid]:
            if sc >= TAU:
                union(vid, nid)
                edges += 1
    comps = {}
    for vid in ids:
        comps.setdefault(find(vid), []).append(vid)
    clusters = sorted(comps.values(), key=len, reverse=True)
    print("4. 聚类: τ=%.2f 边=%d 分量=%d (≥%d条的簇=%d)" % (
        TAU, edges, len(clusters), MIN_CLUSTER, sum(1 for c in clusters if len(c) >= MIN_CLUSTER)))

    # 小组并入最近大簇（质心相似度）；仍不够 → 其他
    big = [c for c in clusters if len(c) >= MIN_CLUSTER]
    small = [c for c in clusters if len(c) < MIN_CLUSTER]
    vec_by_id = {ids[i]: vectors[i] for i in range(len(ids))}
    text_by_id = {ids[i]: texts[i] for i in range(len(ids))}

    def centroid(members):
        n = len(members)
        return [sum(vec_by_id[m][k] for m in members) / n for k in range(DIM)]

    merged_into_other = []
    if big and small:
        cents = [(c, centroid(c)) for c in big]
        for sc_ in small:
            sc_cent = centroid(sc_)
            best, best_sim = None, -1
            for bc, bc_cent in cents:
                sim = sum(a * b for a, b in zip(sc_cent, bc_cent))
                if sim > best_sim:
                    best, best_sim = bc, sim
            if best is not None and best_sim >= TAU - 0.10:
                best.extend(sc_)
            else:
                merged_into_other.extend(sc_)
    elif small and not big:
        for c in small:
            merged_into_other.extend(c)
    final = [c for c in big if len(c) >= MIN_CLUSTER]
    final.sort(key=len, reverse=True)

    # ── 5. 质量报告 ──
    total = len(ids)
    covered = sum(len(c) for c in final)
    report = {
        "tau": TAU, "topk": TOPK, "min_cluster": MIN_CLUSTER,
        "total_demands": total, "n_clusters": len(final),
        "covered": covered, "other_bucket": len(merged_into_other),
        "clusters": [],
    }
    for ci, c in enumerate(final):
        # 簇内离质心最近的 10 条做样例
        cent = centroid(c)
        sims = sorted(((sum(a * b for a, b in zip(vec_by_id[m], cent)), m) for m in c), reverse=True)
        samples = [text_by_id[m] for _, m in sims[:10]]
        report["clusters"].append({
            "rank": ci + 1, "size": len(c), "share": round(len(c) / total, 4),
            "samples": samples,
        })
    with open("/tmp/p1_cluster_result.json", "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=1)
    print("5. 报告已写 /tmp/p1_cluster_result.json")
    print()
    print("═══ 簇排名（τ=%.2f）═══" % TAU)
    for cl in report["clusters"][:15]:
        print("#%d size=%d (%.1f%%) 样例: %s" % (
            cl["rank"], cl["size"], cl["share"] * 100,
            " | ".join(cl["samples"][:4])[:160]))
    print("其他桶(未入簇): %d (%.1f%%)" % (len(merged_into_other), len(merged_into_other) / total * 100))


if __name__ == "__main__":
    t0 = time.time()
    asyncio.run(main())
    print("TOTAL %.0fs" % (time.time() - t0))