# -*- coding:utf-8 -*-
"""P1 聚类质量分析（多τ对比 + 超大簇二次细分 + 向量/kNN缓存）。

在 pipeline-app 测试机：P1_PROJECT_ID=<真实商机项目> ./py3/bin/python /tmp/p1_cluster_analyze.py
缓存：/tmp/p1_vec_cache.json（向量+文本）、/tmp/p1_knn_cache.json（kNN图）——存在则跳过 embedding/kNN（省费用）。
输出 /tmp/p1_analyze_result.json + 控制台对比。
"""
import asyncio, json, os, sys, time
WORKDIR="/d/pipeline/pipeline-app"; os.chdir(WORKDIR); sys.path.insert(0,WORKDIR)
from appPublic.folderUtils import ProgramPath
from appPublic.jsonConfig import getConfig
from appPublic.event_dispatcher import EventDispatcher
from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv
from ahserver.globalEnv import initEnv
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'
initEnv()

TAUS=[0.70,0.75,0.82,0.88]
TOPK=30; MIN_CLUSTER=5; DIM=1024; BIG_SPLIT=150   # 超过此规模的簇二次细分
VEC_CACHE="/tmp/p1_vec_cache.json"; KNN_CACHE="/tmp/p1_knn_cache.json"
TEST_COL="opp_p1_an_%d"%int(time.time())


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


def centroid(vecs):
    n=len(vecs)
    return [sum(v[k] for v in vecs)/n for k in range(DIM)]


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


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[v]:
            if sc>=tau: union(v,nid)
    comps={}
    for v in ids: comps.setdefault(find(v),[]).append(v)
    return sorted(comps.values(),key=len,reverse=True)


async def main():
    import aiohttp
    timeout=aiohttp.ClientTimeout(total=60)
    # ── 拉需求 ──
    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 ""
    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:
                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
    texts=[(it.get("title") or "").strip() or "(无标题)" for it in demands]
    print("需求:",len(texts))

    # ── 向量（缓存优先）──
    if os.path.exists(VEC_CACHE):
        c=json.load(open(VEC_CACHE)); vectors=c["vectors"]; assert len(vectors)==len(texts),"缓存与需求数不符,删缓存重跑"
        print("向量: 命中缓存")
    else:
        from pipeline_service import rag_client as rc
        pid=os.environ.get("P1_PROJECT_ID","")
        if not pid: print("FATAL: 缺 P1_PROJECT_ID"); sys.exit(1)
        vectors=[None]*len(texts)
        for i in range(0,len(texts),100):
            vs,err=None,None
            for a in range(3):
                vs,err=await rc.rag_embed_texts(pid,texts[i:i+100],batch_size=10)
                if not err and vs: break
                vs=None
                if a==2: print("FATAL embed:",err); sys.exit(1)
                await asyncio.sleep(2**a)
            vectors[i:i+len(vs)]=vs
            if i%500==0: print("  embed",i,flush=True)
        json.dump({"vectors":vectors},open(VEC_CACHE,"w"))
        print("向量: 计算完成并缓存")

    ids=["d%d"%i for i in range(len(vectors))]
    # ── kNN（缓存优先）──
    if os.path.exists(KNN_CACHE):
        nbr=json.load(open(KNN_CACHE)); nbr={k:[tuple(x) for x in v] for k,v in nbr.items()}
        print("kNN: 命中缓存")
    else:
        recs=await _sql("SELECT baseurl FROM upapp WHERE id='rag-vdb'")
        vdb=(recs[0].baseurl or "").rstrip("/")
        async with aiohttp.ClientSession(timeout=timeout) as s:
            async def vp(path,pl):
                r=await s.post(vdb+path,json=pl); return await r.text()
            await vp("/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":"P1an","metric":"COSINE"})
            for i in range(0,len(vectors),500):
                rows=[{"id":ids[j],"vector":vectors[j],"text":texts[j][:1900],"batch_id":"p1an"} for j in range(i,min(i+500,len(vectors)))]
                await vp("/v1/upsert",{"colname":TEST_COL,"data":rows})
            nbr={}
            for i,vid in enumerate(ids):
                d=json.loads(await vp("/v1/query",{"colname":TEST_COL,"vector":vectors[i],"pagerows":TOPK,"output_fields":["id"]}))
                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]
                if i%500==0: print("  kNN",i,flush=True)
            await vp("/v1/dropcollection",{"colname":TEST_COL})
        json.dump(nbr,open(KNN_CACHE,"w"))
        print("kNN: 完成并缓存")

    # ── 多τ + 大簇细分 ──
    vec_by={ids[i]:vectors[i] for i in range(len(ids))}
    txt_by={ids[i]:texts[i] for i in range(len(ids))}
    result={"total":len(ids),"by_tau":[]}
    for tau in TAUS:
        comps=union_find(ids,nbr,tau)
        # 大簇二次细分：簇内子图用 tau+0.10 重聚
        final=[]
        for c in comps:
            if len(c)>BIG_SPLIT:
                sub_nbr={v:[(sc,n) for sc,n in nbr[v] if n in set(c)] for v in c}
                subs=union_find(c,sub_nbr,min(tau+0.10,0.95))
                final.extend(subs)
            else:
                final.append(c)
        big=[c for c in final if len(c)>=MIN_CLUSTER]
        # 小组并入最近大簇
        small=[c for c in final if len(c)<MIN_CLUSTER]
        if big and small:
            cents=[(c,centroid([vec_by[m] for m in c])) for c in big]
            other=[]
            for sc_ in small:
                cc=centroid([vec_by[m] for m in sc_]); best,bs=None,-1
                for bc,bcent in cents:
                    sm=cos(cc,bcent)
                    if sm>bs: best,bs=bc,sm
                if best is not None and bs>=tau-0.10: best.extend(sc_)
                else: other.extend(sc_)
        else:
            other=[m for c in small for m in c]
        big.sort(key=len,reverse=True)
        covered=sum(len(c) for c in big)
        entry={"tau":tau,"n_clusters":len(big),"covered":covered,"other":len(other),
               "cover_rate":round(covered/len(ids),3),"top":[]}
        for ci,c in enumerate(big[:12]):
            cent=centroid([vec_by[m] for m in c])
            sims=sorted(((cos(vec_by[m],cent),m) for m in c),reverse=True)
            # 去重样例（众包标题重复多）
            seen=[]; 
            for _,m in sims:
                if txt_by[m] not in seen: seen.append(txt_by[m])
                if len(seen)>=6: break
            entry["top"].append({"rank":ci+1,"size":len(c),"share":round(len(c)/len(ids),3),"samples":seen})
        result["by_tau"].append(entry)
        print("\n═══ τ=%.2f 簇=%d 覆盖=%d(%.0f%%) 其他=%d ═══"%(tau,len(big),covered,covered/len(ids)*100,len(other)))
        for cl in entry["top"][:10]:
            print(" #%d %d条(%.1f%%): %s"%(cl["rank"],cl["size"],cl["share"]*100," | ".join(cl["samples"][:4])[:130]))
    json.dump(result,open("/tmp/p1_analyze_result.json","w"),ensure_ascii=False,indent=1)
    print("\n报告: /tmp/p1_analyze_result.json")


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