#!/usr/bin/env python3
"""
xbrain_recall.py — XKB hybrid search bridge (XBrain layer)

Provides semantic + keyword hybrid search over the XKB knowledge base
using an embedded vector store (pgvector + RRF + Gemini embedding).

Usage:
    python3 scripts/xbrain_recall.py "agent memory workflow"
    python3 scripts/xbrain_recall.py "SEO 策略" --limit 5 --json
    python3 scripts/xbrain_recall.py "query" --no-semantic   # keyword-only

Environment / config (in priority order):
    GBRAIN_DIR          path to vector store runtime (overrides all)
    GEMINI_API_KEY      required for semantic (vector) search
    --env-file          explicit dotenv file for runtime credential injection
    XKB_ENV_FILE        fallback dotenv file when --env-file is omitted

Process environment variables take precedence over either dotenv file. No
host-specific config or private credential fallback is consulted.
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))

import xkb_failures
from runtime_config import runtime_env

import xkb_provenance

# ── Config ─────────────────────────────────────────────────────────────────────

def _safe_runtime_env() -> dict[str, str]:
    """讀執行期設定；讀不到就當作空的。

    這一行在 import 時就會執行，而 load_env_file 對不存在的路徑會拋
    FileNotFoundError。呼叫端只接 ImportError，所以一個過期的 XKB_ENV_FILE
    會讓「模組載不進來」變成從 search() 裡拋出的 FileNotFoundError，
    而不是安靜地退回關鍵字召回。設定不見是要說的，但不是用讓模組無法載入
    的方式說。
    """
    try:
        return runtime_env()
    except Exception as err:  # noqa: BLE001
        print(f"[xkb] xbrain 讀不到執行期設定，改用行程環境：{type(err).__name__}: {err}",
              file=sys.stderr, flush=True)
        return {}


_RUNTIME_ENV = _safe_runtime_env()


def _resolve_gbrain_dir(settings: dict[str, str] | None = None) -> Path | None:
    """Find the gbrain runtime directory, in priority order."""
    settings = _RUNTIME_ENV if settings is None else settings
    candidates = [
        os.getenv("GBRAIN_DIR"),
        settings.get("GBRAIN_DIR"),
        str(Path.home() / "Desktop" / "gbrain"),
        str(Path.home() / "gbrain"),
        "/opt/gbrain",
    ]
    for c in candidates:
        if c and Path(c).joinpath("src", "cli.ts").exists():
            return Path(c)
    return None


GBRAIN_DIR = _resolve_gbrain_dir()
GBRAIN_AVAILABLE = GBRAIN_DIR is not None


def _resolve_bun() -> str:
    """Find bun executable — checks PATH first, then common install locations."""
    import shutil
    if shutil.which("bun"):
        return "bun"
    for candidate in [
        str(Path.home() / ".bun" / "bin" / "bun"),
        "/usr/local/bin/bun",
        "/opt/homebrew/bin/bun",
    ]:
        if Path(candidate).exists():
            return candidate
    return "bun"  # let subprocess give the real error


BUN = _resolve_bun()

GEMINI_API_KEY = _RUNTIME_ENV.get("GEMINI_API_KEY", "")


def _make_subprocess_env(
    semantic: bool,
    settings: dict[str, str] | None = None,
) -> dict[str, str]:
    """Build subprocess env from the portable runtime contract."""
    settings = _RUNTIME_ENV if settings is None else settings
    env = {**settings}
    gemini_api_key = settings.get("GEMINI_API_KEY", "")
    if semantic and gemini_api_key:
        env["GEMINI_API_KEY"] = gemini_api_key
        # gbrain v0.42+ gateway 的 google recipe 讀這個變數名
        env["GOOGLE_GENERATIVE_AI_API_KEY"] = gemini_api_key
    elif not semantic:
        env.pop("GEMINI_API_KEY", None)
        env.pop("GOOGLE_GENERATIVE_AI_API_KEY", None)
        env.pop("OPENAI_API_KEY", None)
    return env


# ── Core query ─────────────────────────────────────────────────────────────────

def xbrain_query(
    query: str,
    *,
    limit: int = 10,
    no_expand: bool = True,
    semantic: bool = True,
    env_file: str | Path | None = None,
    diagnostics: dict | None = None,
) -> list[dict[str, Any]]:
    """
    Run hybrid search and return structured results.
    Returns [] if XBrain is unavailable or query fails (never raises).

    Result keys: slug, title, type, chunk_text, score, source_url, stale
    """
    state = diagnostics if diagnostics is not None else {}
    state.update(available=False, attempted=False, used=False, status="unavailable")
    settings = runtime_env(env_file) if env_file is not None else _RUNTIME_ENV
    gbrain_dir = _resolve_gbrain_dir(settings)
    if gbrain_dir is None:
        return []
    state.update(available=True, attempted=True)

    cmd = [BUN, "run", str(gbrain_dir / "src" / "cli.ts")]
    # The human CLI ignores --json and truncates evidence to 100 characters.
    # The operation surface returns the same search as lossless JSON.
    cmd += ["call", "query", json.dumps({"query": query, "limit": limit,
                                        "expand": not no_expand})]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            env=_make_subprocess_env(semantic, settings),
            cwd=str(gbrain_dir),
            # 召回是互動路徑；連不上後端時，這個秒數就是使用者要等的時間。
            timeout=int(os.getenv("XKB_XBRAIN_TIMEOUT", "30")),
        )
    except subprocess.TimeoutExpired as err:
        state["status"] = "timeout"
        # 逾時、找不到 bun、非零離開碼——三種都回空陣列，而空陣列的意思是
        # 「這個主題我們沒有知識」。它們其實是「問不到」。這正是下面那段
        # 註解在講的同一件事，只是原本只有最後一種會留下記錄。
        xkb_failures.note("semantic recall (xbrain) 逾時", err)
        return []
    except FileNotFoundError as err:
        state["status"] = "unavailable"
        xkb_failures.note("semantic recall (xbrain)：PATH 上沒有 bun", err)
        return []
    except Exception as err:
        state["status"] = "error"
        # 語意搜尋的後端掛掉，跟「這個主題我們沒有知識」是兩件事。
        # 2026-05-04 的十二週靜默故障，就是因為這兩者長得一模一樣。
        xkb_failures.note("semantic recall (xbrain)", err)
        return []

    if result.returncode != 0:
        state["status"] = "error"
        # 只留離開碼，不要把 stderr 貼進來：後端的錯誤訊息會回音請求內容，
        # 而請求裡有憑證。tests/test_xbrain_recall_runtime.py 的
        # test_failed_backend_does_not_echo_credential 就是釘這件事的。
        xkb_failures.note("semantic recall (xbrain)", RuntimeError(f"exit {result.returncode}"))
        return []

    raw = result.stdout.strip()
    state["status"] = "empty"
    if not raw or raw == "No results.":
        return []

    truncated_display = False
    try:
        items = json.loads(raw)
    except json.JSONDecodeError:
        # gbrain v0.42+ 移除了 query --json，輸出為「[分數] slug -- 內文」行格式
        items = _parse_line_format(raw)
        truncated_display = True
        if not items:
            state["status"] = "invalid_response"
            return []

    if not isinstance(items, list) or not all(isinstance(item, dict) for item in items):
        state["status"] = "invalid_response"
        return []

    results = []
    for item in items:
        chunk = item.get("chunk_text", "")
        slug = item.get("slug", "")
        source_url = _source_url(item)
        results.append({
            "slug": slug,
            "title": item.get("title", ""),
            "type": item.get("type", ""),
            "chunk_text": chunk,
            "score": item.get("score", 0.0),
            "score_scale": "card_hybrid",
            "source_url": source_url,
            "stale": item.get("stale", False),
            **({"excerpt_truncated": True} if truncated_display else {}),
        })
    state.update(used=bool(results), status="used" if results else "empty")
    return results


def _parse_line_format(raw: str) -> list[dict[str, Any]]:
    """解析 gbrain v0.42+ 的行格式輸出：`[0.9102] slug -- chunk text`。
    非結果行（gateway 警告等）自動略過。"""
    import re

    # gbrain 在最後一筆結果之後還會印東西（警告、計時、棄用提示）。原本的
    # 續行判斷是「不是結果、而且已經看過至少一筆」，所以那些行全部被接到
    # 最後一張卡的內文裡，然後當成知識顯示給使用者看。
    RESULT = re.compile(r"^\[(\d+\.\d+)\]\s+(\S+)\s+--\s?(.*)$")
    NOISE = re.compile(
        r"^\s*(?:\[|warn|warning|error|deprecat|note:|info:|took\b|elapsed\b|\d+\s*ms\b)",
        re.I,
    )
    MAX_CONTINUATION_CHARS = 4000

    items: list[dict[str, Any]] = []
    for line in raw.splitlines():
        stripped = line.strip()
        match = RESULT.match(stripped)
        if match:
            items.append({
                "slug": match.group(2),
                "title": "",
                "type": "",
                "chunk_text": match.group(3),
                "score": float(match.group(1)),
                "stale": False,
            })
            continue
        if not items or not stripped or NOISE.match(stripped):
            continue
        # 續行。原本用未 strip 的行判斷開頭，所以縮排過的結果會被當成內文。
        last = items[-1]
        if len(last["chunk_text"]) < MAX_CONTINUATION_CHARS:
            last["chunk_text"] += "\n" + stripped
    return items


def _source_url(item: dict) -> str:
    """Prefer exact metadata, then canonical frontmatter; never guess GitHub."""
    for metadata in (item, item.get("frontmatter") or {}):
        if metadata.get("source_url"):
            return str(metadata["source_url"])
    slug = item.get("slug", "")
    if slug.startswith("xkb-case-"):
        try:
            slug = bytes.fromhex(slug.removeprefix("xkb-case-")).decode("utf-8")
        except (ValueError, UnicodeError):
            pass
    import xkb_paths
    import xkb_frontmatter
    roots = (xkb_paths.CARDS_DIR, xkb_paths.BOOKMARKS_DIR / "youtube")
    for root in roots:
        leaf = slug.removeprefix("youtube-") if root.name == "youtube" else slug
        candidate = root / (leaf + ".md")
        try:
            if candidate.is_file() and candidate.resolve().is_relative_to(root.resolve()):
                value = xkb_frontmatter.parse(candidate.read_text(encoding="utf-8")).get("source_url")
                if value:
                    return str(value).strip("\"'")
        except (OSError, UnicodeError):
            # Local provenance is optional; a damaged local copy must not erase
            # otherwise valid remote search hits.
            continue
    return _extract_source_url(item.get("chunk_text", "")) or _url_from_slug(slug)


def _extract_source_url(text: str) -> str:
    """Extract source URL from card content (multiple patterns)."""
    import re
    patterns = [
        r"Source URL:\s*(https?://\S+)",
        r"來源:\s*(https?://\S+)",
        r"source_url:\s*(https?://\S+)",
        r"- (?:Tweet|原始來源|Source):\s*(https?://\S+)",
    ]
    for pat in patterns:
        m = re.search(pat, text)
        if m:
            return m.group(1).rstrip(".,)")
    return ""


def _url_from_slug(slug: str) -> str:
    """Reconstruct source URL from slug naming conventions."""
    import re
    # Tweet IDs: 15-20 digit slugs → x.com URL
    if re.fullmatch(r"\d{15,20}", slug):
        return f"https://x.com/i/status/{slug}"
    # legacy-* slugs derived from tweet files: extract numeric suffix
    m = re.search(r"(\d{15,20})$", slug)
    if m:
        return f"https://x.com/i/status/{m.group(1)}"
    # YouTube slugs: youtube-{video_id}
    m = re.match(r"youtube-([A-Za-z0-9_\-]{11})$", slug)
    if m:
        return f"https://www.youtube.com/watch?v={m.group(1)}"
    # GitHub owner/repo both allow hyphens: this encoding is irreversible.
    # Never manufacture a provenance URL from it.
    return ""


# ── Formatted output ───────────────────────────────────────────────────────────

def format_for_conversation(results: list[dict], query: str) -> str:
    """Format recall results for injecting into conversation context."""
    if not results:
        return ""

    lines = [f"◆ 相關既有知識（query: {query!r}）\n"]
    for i, r in enumerate(results[:5], 1):
        title = xkb_provenance.strip_markers(r["title"] or r["slug"])
        url = r["source_url"]
        snippet = xkb_provenance.strip_markers(
            r["chunk_text"][:200].replace("\n", " ").strip())
        score_pct = f"{r['score'] * 100:.1f}%"

        lines.append(f"{i}. **{title}** ({score_pct})")
        if snippet:
            lines.append(f"   {snippet}…")
        if url:
            lines.append(f"   {url}")
        lines.append("")

    return "\n".join(lines)


# ── Backward compatibility alias ──────────────────────────────────────────────
# Old scripts that import gbrain_recall can import from here instead.
gbrain_query = xbrain_query


# ── CLI ────────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(description="XBrain hybrid search")
    parser.add_argument("query", help="Search query")
    parser.add_argument("--limit", type=int, default=10)
    parser.add_argument("--json", action="store_true", dest="output_json")
    parser.add_argument("--no-semantic", action="store_true")
    parser.add_argument(
        "--env-file",
        help="Explicit XKB dotenv file (process environment takes precedence)",
    )
    args = parser.parse_args()

    try:
        settings = runtime_env(args.env_file) if args.env_file else _RUNTIME_ENV
    except (FileNotFoundError, OSError, ValueError) as exc:
        print(f"Unable to load XKB runtime configuration: {exc}", file=sys.stderr)
        sys.exit(1)

    if _resolve_gbrain_dir(settings) is None:
        print("XBrain not available. Set GBRAIN_DIR or clone to ~/Desktop/gbrain.",
              file=sys.stderr)
        sys.exit(1)

    if not args.no_semantic and not settings.get("GEMINI_API_KEY"):
        print(
            "Semantic recall requires GEMINI_API_KEY via the process environment "
            "or the selected XKB env file (process environment takes precedence).",
            file=sys.stderr,
        )
        sys.exit(1)

    results = xbrain_query(
        args.query,
        limit=args.limit,
        semantic=not args.no_semantic,
        env_file=args.env_file,
    )

    if args.output_json:
        print(json.dumps(results, ensure_ascii=False, indent=2))
    else:
        print(format_for_conversation(results, args.query))


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

if __name__ == "__main__":
    xkb_usage.record(__file__)
    main()