skills-win/bmem/scripts/bmem_recall.py

#!/usr/bin/env python
"""在本机 bmem 镜像里检索历史：语义条目 / 文档章节 / 事件。

bmem 的 CLI 目前只负责写入与同步，没有检索命令，但它把 Notes 全文、章节拆分、
语义条目和事件都存进了本机 SQLite 镜像。这个脚本把 bmem 当库调用，只读地把它们
查出来——不改任何数据，不连中心库，离线可用。

用法：
    python bmem_recall.py "多径 时延"              # 三类都查
    python bmem_recall.py "CP 长度" --project amem  # 限定项目
    python bmem_recall.py "时间戳" --kind decision  # 只看某类语义条目
    python bmem_recall.py --recent 10               # 不给关键词=看最近发生了什么

中文按 bigram 检索（bmem 写入时做了滑窗补偿），所以两到四个字的词组比整句命中率高。
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
from pathlib import Path


def _bootstrap() -> None:
    """bmem 装在 langgraph conda 环境里，当前解释器没有时换一个再跑。"""
    try:
        import bmem  # noqa: F401
        return
    except ImportError:
        pass

    home = Path(os.path.expanduser("~"))
    candidates = [
        home / "miniconda3" / "envs" / "langgraph" / "python.exe",
        home / ".conda" / "envs" / "langgraph" / "python.exe",
        home / "miniconda3" / "envs" / "langgraph" / "bin" / "python",
        home / ".conda" / "envs" / "langgraph" / "bin" / "python",
    ]
    for python in candidates:
        if python.exists() and Path(sys.executable).resolve() != python.resolve():
            env = dict(os.environ, PYTHONUTF8="1", PYTHONIOENCODING="utf-8")
            raise SystemExit(
                subprocess.run([str(python), __file__, *sys.argv[1:]], env=env).returncode
            )
    raise SystemExit(
        "找不到 bmem 包。它以 editable 方式装在 langgraph conda 环境里；\n"
        "确认 `bmem help` 能跑，或查 ~/skills_env.md 里记录的解释器路径。"
    )


_bootstrap()

from bmem import config as config_mod  # noqa: E402
from bmem.localdb import Database, fts_query  # noqa: E402


def _print_facts(db: Database, query: str, prj_id: str | None, kind: str | None, limit: int) -> int:
    sql = (
        "SELECT f.prj_id, f.rel_path, f.kind, f.status, f.title, f.body, f.event_date, "
        "f.extraction_gen FROM doc_facts f WHERE 1=1"
    )
    params: list[object] = []
    if prj_id:
        sql += " AND f.prj_id = ?"
        params.append(prj_id)
    if kind:
        sql += " AND f.kind = ?"
        params.append(kind)
    if query:
        sql += " AND (f.title LIKE ? OR f.body LIKE ?)"
        params += [f"%{query}%", f"%{query}%"]
    sql += " ORDER BY f.event_date DESC NULLS LAST, f.rel_path, f.fact_no LIMIT ?"
    params.append(limit)

    rows = db.rows(sql, params)
    if not rows:
        return 0
    print("=== 语义条目 (doc_facts) ===")
    for row in rows:
        status = f" [{row['status']}]" if row["status"] else ""
        date = f" {row['event_date']}" if row["event_date"] else ""
        print(f"\n[{row['kind']}]{status}{date} {row['title']}")
        print(f"  来源 {row['prj_id']}:{row['rel_path']} (第 {row['extraction_gen']} 代抽取)")
        body = row["body"].strip().replace("\n", "\n  ")
        print(f"  {body[:400]}{'…' if len(row['body']) > 400 else ''}")
    return len(rows)


def _print_sections(db: Database, query: str, prj_id: str | None, limit: int) -> int:
    """章节级命中：比整篇文档精确，直接告诉你该回读哪一节。"""
    sql = (
        "SELECT s.prj_id, s.rel_path, s.section_no, s.heading, s.level, s.body_md "
        "FROM note_sections s JOIN ("
        "  SELECT prj_id, rel_path, content_sha256 FROM notes_revisions r1 "
        "  WHERE imported_at = (SELECT max(imported_at) FROM notes_revisions r2 "
        "                       WHERE r2.prj_id=r1.prj_id AND r2.rel_path=r1.rel_path)"
        ") latest ON latest.prj_id=s.prj_id AND latest.rel_path=s.rel_path "
        "        AND latest.content_sha256=s.content_sha256 "
        "WHERE s.body_md LIKE ?"
    )
    params: list[object] = [f"%{query}%"]
    if prj_id:
        sql += " AND s.prj_id = ?"
        params.append(prj_id)
    sql += " ORDER BY s.prj_id, s.rel_path, s.section_no LIMIT ?"
    params.append(limit)

    rows = db.rows(sql, params)
    if not rows:
        return 0
    print("\n=== 文档章节 (note_sections，仅最新修订) ===")
    for row in rows:
        heading = row["heading"] or "（前言）"
        print(f"\n{row['prj_id']}:{row['rel_path']} § {row['section_no']} {heading}")
        body = row["body_md"]
        pos = body.lower().find(query.lower())
        start = max(0, pos - 120) if pos >= 0 else 0
        snippet = body[start : start + 400].strip().replace("\n", "\n  ")
        print(f"  …{snippet}…")
    return len(rows)


def _print_events(db: Database, query: str, prj_id: str | None, limit: int) -> int:
    sql = "SELECT ts, machine_id, prj_id, event_type, result, summary FROM events WHERE 1=1"
    params: list[object] = []
    if prj_id:
        sql += " AND prj_id = ?"
        params.append(prj_id)
    if query:
        sql += " AND summary LIKE ?"
        params.append(f"%{query}%")
    sql += " ORDER BY ts DESC LIMIT ?"
    params.append(limit)

    rows = db.rows(sql, params)
    if not rows:
        return 0
    print("\n=== 事件 (events) ===")
    for row in rows:
        result = f" -> {row['result']}" if row["result"] else ""
        print(
            f"  {row['ts'][:16]} {row['machine_id']} {row['prj_id']} "
            f"[{row['event_type']}{result}] {row['summary'][:100]}"
        )
    return len(rows)


def main() -> int:
    parser = argparse.ArgumentParser(description="在本机 bmem 镜像里只读检索历史")
    parser.add_argument("query", nargs="?", default="", help="关键词；中文用两到四个字的词组")
    parser.add_argument("--project", help="限定 prjID")
    parser.add_argument("--kind", help="限定语义条目类型：decision/issue/requirement/result/param/…")
    parser.add_argument("--limit", type=int, default=8)
    parser.add_argument("--recent", type=int, metavar="N", help="不检索，直接看最近 N 条事件")
    args = parser.parse_args()

    try:
        cfg = config_mod.load()
    except Exception as exc:  # noqa: BLE001
        print(f"本机还没加入 bmem：{exc}\n先跑 `bmem install`。", file=sys.stderr)
        return 1

    if not cfg.db_path.exists():
        print(f"本地镜像不存在：{cfg.db_path}\n跑一次 `bmem sync` 会从中心库重建。", file=sys.stderr)
        return 1

    with Database.open(cfg) as db:
        if args.recent:
            _print_events(db, "", args.project, args.recent)
            return 0

        if not args.query and not args.kind:
            parser.error("给一个关键词，或用 --kind 列举某类条目，或用 --recent N 看最近发生了什么")

        hits = _print_facts(db, args.query, args.project, args.kind, args.limit)
        if args.query:
            # 空关键词配 LIKE '%%' 会把整库倒出来，只在真的有关键词时查这两类
            hits += _print_sections(db, args.query, args.project, args.limit)
            hits += _print_events(db, args.query, args.project, args.limit)

        if hits == 0:
            print(f"没有命中「{args.query}」。")
            print("中文按 bigram 索引，换更短的词组再试；也可能这条历史确实没记过。")
            print(f"（FTS 等价查询串：{fts_query(args.query)}）")
        else:
            print(f"\n共 {hits} 条命中。命中只是线索——回读原文再下结论。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())