#!/usr/bin/env python3
"""
跨層召回融合

每條召回腿有自己的計分方式，尺度互不相同：wiki 是關鍵字或餘弦、卡片是
gbrain RRF 或餘弦、對話是折扣過的關鍵字比例、反例、可復用資產各一套。
把這些原始數字並排排序，等於拿公分比英吋。

**rank() 用 reciprocal rank fusion（RRF）**：只看每筆在自己那條腿裡的名次，
不看原始分數的絕對值。BM25 的無上限分數跟餘弦的 0~1 從來不需要被弄成可比，
因為只有名次跨過腿的邊界。

    unified_score = Σ（該腿權重 / (K + 腿內名次)）；低於相關度地板的命中再減 1.0，
    穩定墊底（見 rank()）。

權重（wiki > 卡片 > 對話 > 反例 > 動作 > 驗不出來，擠在 0.80~1.0）是平手時的
傾向，不是碾壓；K=60 是 RRF 慣例值。

XKB 的 wiki 腿只有十幾個 topic，常常只回一個弱命中——純 RRF 會讓那個弱命中拿
腿內第 1、壓過強卡片。所以排序分兩層：先所有「在至少一條腿裡過了相關度地板
（RELEVANCE_FLOOR）」的命中，再所有沒過的，各層照 unified_score。名次本身不動，
分層保證弱命中排在所有上地板命中之後、不管別條腿回了幾筆。

relevance() / unified()（舊的 raw/(raw+anchor)×權重）留著只給 relevance 這個
顯示欄位用；排序不走它。anchor 與權重可用環境變數覆寫（XKB_SCORE_ANCHOR_WIKI）。
"""
from __future__ import annotations

import os
from xkb_evidence import identity_key

# 各層實測中位數（2026-07-28，8 個代表性查詢）。改了計分方式就要重新量。
#
# 這裡的鍵是**尺度**，不是資料種類。呼叫端用 score_scale 指定，沒指定才
# 退回 source_type——因為同一種資料可能走不同的計分方式（wiki 有關鍵字
# 版也有語意版，數字差一個量級）。
DEFAULT_ANCHORS = {
    "wiki": 0.93,
    "memory": 0.93,              # 與 wiki 同一支程式、同一種計分
    # 語意召回回傳的是餘弦相似度（門檻 0.65，實測有效區間約 0.65~0.85），
    # 跟關鍵字那套完全不同尺度，所以另立錨點
    "wiki_semantic": 0.72,
    "memory_semantic": 0.72,
    "card": 0.88,
    "card_hybrid": 0.88,         # Same GBrain hybrid score, now with structured provenance.
    "bookmark": 0.88,
    # 卡片被實測成餘弦之後（xkb_relevance 的 rewrite_score）就不再是 RRF，
    # 0.88 那個錨點是照 RRF 量的。餘弦有自己的區間，跟 wiki 語意同一段。
    "card_semantic": 0.72,
    # 關鍵字退路的分數是 _keyword_unit_score 的覆蓋率（0~1），既不是 RRF
    # 也不是餘弦。錨點取這個區間的中點——那是這個尺度的定義，不是量出來的
    # 中位數，所以標明清楚：換了計分方式要重量。
    "card_keyword": 0.5,
    "wiki_keyword": 0.5,
    # 沒辦法跟問題比對過的項目（網址、semantic: id、索引讀不到）。
    # 帶的是後端自己的 RRF，沒有被驗證過。這件事以前是靠在 xkb_relevance
    # 裡改寫分數做到的，那等於捏造一個測量值；尺度的事在尺度表裡解決。
    #
    # 目標是「剛好落在驗證過的卡片之下」，不是「掉到所有東西之下」。
    # 我第一版用 1.6/0.5，典型 RRF 只剩 0.1774，反而低於對話軌跡的
    # 0.4033——於是卡片索引壞掉時，整個封包都是對話軌跡、卡片被截光，
    # 正是「索引壞掉長得像知識庫裡沒東西」。調錨點時要算交叉點，
    # 不能只看自己那一層的數字有沒有變小。
    "unverified": 0.59,
    "contrarian": 0.52,
    "action": 0.31,
    # 對話軌跡：呼叫端（Store.recall）已經乘過 KEYWORD_EVIDENCE_DISCOUNT
    # (0.65) 把它放到與餘弦可比的區間了，所以這裡用餘弦的錨點。
    # 原本這個錨點寫的是「對未折扣比例」的 0.55、權重又再壓一次 0.55，
    # 等於同一個「關鍵字證據較弱」的判斷扣了兩遍：實測滿分命中只得
    # 0.2979，低於每一個剛過門檻的卡片與 wiki，於是知識層一填滿 limit，
    # 對話軌跡就被整段截掉——共享對話記憶等於從 /v1/recall 消失。
    "conversation": 0.72,
}

# 來源權威度：消化過的結論 > 原始素材 > 補充提醒。
# RRF 之下這是 RRF 貢獻的乘數，不是正規化分數的乘數——某一層改了計分方式
# 不需要重調這裡，因為那一層只貢獻一個名次。壓縮在 0.5~1.0：權威度是
# 平手時的傾向，不該讓 wiki 第 5 名壓過卡片第 1 名。
DEFAULT_WEIGHTS = {
    # 權威度是**平手時的傾向**，不是跨好幾個名次的碾壓。K=60 之下，相鄰名次
    # 的 RRF 貢獻很接近，所以權重必須擠在一個窄帶裡：wiki 第 2 名壓過卡片
    # 第 1 名可以，wiki 第 7 名壓過卡片第 1 名不行。實測 0.96 對 1.0 的差
    # 大約值「一個名次」。
    "wiki": 1.0,
    "memory": 1.0,
    "wiki_semantic": 1.0,
    "memory_semantic": 1.0,
    "wiki_keyword": 1.0,
    "card": 0.96,
    "card_hybrid": 0.96,
    "bookmark": 0.96,
    "card_semantic": 0.96,
    "card_keyword": 0.96,
    "card_bm25": 0.96,
    "wiki_bm25": 1.0,
    "conversation": 0.93,
    "contrarian": 0.90,
    "action": 0.87,
    # 驗不出來（網址、semantic: id、索引讀不到）：不是零（不代表不相關），
    # 但穩定墊在所有驗證過的東西之下。
    "unverified": 0.80,
}

# RRF 常數。60 是慣例值——讓相鄰名次的貢獻接近，融合才穩。
K_RRF = 60


# 腿內相關度的地板。relevance() 設計成「該腿中位相關」落在 0.5；低於這個地板
# 算明顯偏弱。偏弱的命中名次往後推，排在所有上地板命中之後——這樣「wiki 腿
# 只回一個弱命中」不會因為腿內唯一就搶到第 1 名壓過強卡片（審查 finding 1）。
RELEVANCE_FLOOR = 0.35

FALLBACK_ANCHOR = 0.5
FALLBACK_WEIGHT = 0.5


def _env_override(prefix: str, source_type: str, default: float) -> float:
    raw = os.getenv(f"{prefix}_{source_type.upper()}")
    if not raw:
        return default
    try:
        return float(raw)
    except ValueError:
        return default


def anchor_for(source_type: str) -> float:
    return _env_override("XKB_SCORE_ANCHOR", source_type,
                         DEFAULT_ANCHORS.get(source_type, FALLBACK_ANCHOR))


def weight_for(source_type: str) -> float:
    return _env_override("XKB_SCORE_WEIGHT", source_type,
                         DEFAULT_WEIGHTS.get(source_type, FALLBACK_WEIGHT))


def relevance(raw_score: float, source_type: str) -> float:
    """把單層的原始分數壓成 0~1。該層的中等相關會落在 0.5。"""
    if raw_score <= 0:
        return 0.0
    anchor = anchor_for(source_type)
    if anchor <= 0:
        return 1.0
    return raw_score / (raw_score + anchor)


def unified(raw_score: float, source_type: str) -> float:
    """舊的對齊分數（raw/(raw+anchor)×權重）。rank() 已改用 RRF，不再呼叫它；
    留著給還在讀這個欄位的舊呼叫端。"""
    return round(relevance(raw_score, source_type) * weight_for(source_type), 4)


def _leg_of(item: dict) -> str:
    """這筆結果是哪一條召回腿產生的。

    score_scale 是「分數怎麼算的」，由算出它的地方標上；同一個 score_scale
    的兩筆一定來自同一條腿，可以照各自的原始分數排出腿內名次。沒標的退回
    source_type。
    """
    return str(item.get("score_scale") or item.get("source_type", "") or "?")


def rank(results: list[dict]) -> list[dict]:
    """用 reciprocal rank fusion 融合多條召回腿，就地補上 unified_score 並排序。

    舊版把每層的原始分數用手調錨點壓到 0~1 再乘權重——那個設計會因為調一個
    錨點就讓某層靜默壓過另一層（這個 session 為此繞了六輪）。RRF 融合的是
    **名次**，所以 BM25 的無上限分數跟餘弦的 0~1 從來不需要被弄成可比。

    unified_score = Σ（該腿權重 / (K + 腿內名次)），對它出現過的每一條腿加總
    （目前每筆只在一條腿；加 BM25 腿後、同一張卡被 BM25 和向量都撈到會有兩項）。

    排序分兩層：先所有「在至少一條腿裡過了相關度地板」的命中（照 unified_score），
    再所有沒過的。XKB 的 wiki 腿只有十幾個 topic、常常只回一個弱命中——分層排序
    保證那個弱命中排在**所有**上地板命中之後，不管別條腿回了幾筆。

    idempotent：matched_by / leg_rank / relevance / unified_score 每次呼叫都重建。
    原始分數 <= 0 或缺分的項目不拿名次，歸到下層、墊底。
    """
    # 按路徑去重：一張卡被 BM25 和向量都撈到（最強訊號）只留一筆，但兩條腿
    # 的貢獻都要算進去。search() 原本按路徑丟掉 BM25 的重複，於是雙腿加分
    # 永遠不會發生；light 路徑上同一頁 wiki 被兩條腿撈到會重複出現。
    uniq: dict = {}
    for item in results:
        k = identity_key(item) or ("anonymous", id(item))
        if k not in uniq:
            item["matched_by"] = []
            item["_rrf"] = 0.0
            item["_above_floor"] = False
            item["_legs"] = []
            item.pop("leg_rank", None)
            item.pop("relevance", None)
            uniq[k] = item
        uniq[k]["_legs"].append((_leg_of(item), float(item.get("score") or 0.0)))
    survivors = list(uniq.values())

    legs: dict[str, list[tuple[float, dict]]] = {}
    for surv in survivors:
        # 同一筆在同一條腿裡只算一次（取最高分）。合併後 _legs 可能對同一條腿
        # 有多筆（同一頁 wiki 的兩段、search() 不再去重後的重複列），不去重就
        # 會把 w/(K+i) 加兩次、matched_by 也重複。
        best: dict[str, float] = {}
        for leg, score in surv["_legs"]:
            if leg not in best or score > best[leg]:
                best[leg] = score
        surv["_legs"] = list(best.items())
        surv["_has_semantic"] = any(str(l).endswith("_semantic") for l in best)
        for leg, score in best.items():
            legs.setdefault(leg, []).append((score, surv))

    for leg, pairs in legs.items():
        # 有分數的排名次；<=0 或缺分的沉到這條腿最後。
        pairs.sort(key=lambda ps: (ps[0] > 0.0, ps[0]), reverse=True)
        w = weight_for(leg)
        # 相關度地板只管語意腿。餘弦有校準過的絕對門檻（~0.55），小到某個值
        # 就是「其實沒關係」。BM25／關鍵字／memory／conversation 命中要嘛是
        # 字面命中、要嘛已過各自上游門檻，沒有這種問題——對它們套地板等於
        # 拿沒發生過的比較把整條腿打到 -1.0，broken index 會因此長得像空知識庫
        # （本專案記錄在案）。thin wiki leg 那個問題出在 wiki_semantic，仍受地板。
        is_semantic = leg.endswith("_semantic")
        for i, (score, surv) in enumerate(pairs, 1):
            if score <= 0.0:
                surv.setdefault("relevance", 0.0)
                continue
            surv["matched_by"].append(leg)
            surv["leg_rank"] = min(surv.get("leg_rank", i), i)
            surv["_rrf"] += w / (K_RRF + i)
            if is_semantic:
                rel = relevance(score, leg)
                if rel >= RELEVANCE_FLOOR:
                    surv["_above_floor"] = True
                r = round(rel, 4)
            else:
                surv["_above_floor"] = True
                # 名次代理只在「這一筆完全沒有語意腿」時才拿來當顯示 relevance——
                # 不要用 rank-1 => 1.0 蓋掉同一張卡真的餘弦強度。
                r = round(1.0 / i, 4) if not surv.get("_has_semantic") else -1.0
            if r > surv.get("relevance", -1.0):
                surv["relevance"] = r

    for surv in survivors:
        surv.pop("_legs", None)
        # 下地板的減 1.0——_rrf 都在 0.01~0.02，減完必為負，穩定墊在所有上地板
        # 之下。這樣「照 unified_score 由大到小排」就等於回傳順序。
        above_floor = surv.pop("_above_floor", False)
        # Demotion is decided at the service merge boundary from explicit
        # relevance verdicts. Ranking only applies the flag; retrieval scores
        # cannot override a judgement measured on another scale.
        demoted = bool(surv.get("demoted"))
        surv["unified_score"] = round(surv.pop("_rrf", 0.0)
                                      - (0.0 if above_floor else 1.0)
                                      - (1.0 if demoted else 0.0), 6)
    return sorted(survivors, key=lambda r: r["unified_score"], reverse=True)


import xkb_usage  # noqa: E402  — 量測誰在跑，見 scripts/xkb_usage.py

if __name__ == "__main__":
    xkb_usage.record(__file__)
    # 示範 RRF 融合：三條腿合在一起，看排序
    demo = [
        {"title": "wiki 強命中", "score": 0.85, "score_scale": "wiki_semantic"},
        {"title": "wiki 弱命中（腿內唯一）", "score": 0.31, "score_scale": "wiki_semantic"},
        {"title": "卡片 強命中", "score": 0.88, "score_scale": "card_semantic"},
        {"title": "卡片 中命中", "score": 0.66, "score_scale": "card_semantic"},
        {"title": "對話 滿分", "score": 0.65, "score_scale": "conversation"},
        {"title": "驗不出來的網址", "score": 0.9, "score_scale": "unverified"},
        {"title": "動作提示（無分）", "score": 0.0, "source_type": "action"},
    ]
    for r in rank(demo):
        print(f"  {r['unified_score']:.6f}  rel={r['relevance']:.3f}  "
              f"leg#{r.get('leg_rank', '-')}  {r['title']}")