可视化工具 MCP / 在线平台 收录库

#!/usr/bin/env bash
# ============================================================
# Bizard MCP 一键安装 / 注册脚本（幂等，可重复跑）
#
# 做的事：
#   1. npm 全局安装 @openbiox/bizard-mcp（已装则跳过，有新版则升级）
#   2. 注册进 DSH web profile（~/.dsh/profiles/web/cordis.patch.yml）
#   3. 注册进 Codex CLI（~/.codex/config.toml）
#   4. 冒烟测试：起 stdio 进程跑一次 tools/list
#
# 用法:
#   ./01_install.sh            # 安装 + 注册全部
#   ./01_install.sh --dsh      # 只注册 DSH
#   ./01_install.sh --codex    # 只注册 Codex
#   ./01_install.sh --check    # 只做体检，不安装不写入
# ============================================================
set -euo pipefail

PKG="@openbiox/bizard-mcp"
DSH_PATCH="$HOME/.dsh/profiles/web/cordis.patch.yml"
CODEX_CFG="$HOME/.codex/config.toml"

ONLY_DSH=0; ONLY_CODEX=0; CHECK_ONLY=0
for a in "$@"; do
  case "$a" in
    --dsh)   ONLY_DSH=1 ;;
    --codex) ONLY_CODEX=1 ;;
    --check) CHECK_ONLY=1 ;;
    *) echo "未知参数: $a"; exit 1 ;;
  esac
done

say() { printf '\n=== %s ===\n' "$1"; }

# ---------- 1. 安装 ----------
if [ "$CHECK_ONLY" -eq 1 ]; then
  say "体检"
  command -v bizard-mcp >/dev/null && echo "✅ 可执行: $(command -v bizard-mcp)" || echo "❌ 未安装 bizard-mcp"
  grep -q "mcp-bizard" "$DSH_PATCH" 2>/dev/null && echo "✅ DSH 已注册" || echo "❌ DSH 未注册"
  grep -q "mcp_servers.bizard" "$CODEX_CFG" 2>/dev/null && echo "✅ Codex 已注册" || echo "❌ Codex 未注册"
  exit 0
fi

say "1/4 安装 $PKG"
if npm ls -g --depth=0 2>/dev/null | grep -q "@openbiox/bizard-mcp"; then
  echo "已安装，跳过（升级请手动：npm update -g $PKG 或 npm i -g $PKG@latest）"
else
  npm install -g "$PKG"
fi
BIN="$(command -v bizard-mcp || true)"
[ -n "$BIN" ] || { echo "❌ 安装后找不到 bizard-mcp，检查 npm prefix 是否在 PATH"; exit 1; }
echo "可执行路径: $BIN"
# 用 bin 软链而不是 readlink 到 dist/index.js：后者是包内实现细节，升级换布局就失效。
# 注意：$BIN 位于 nvm 版本目录（.../versions/node/vX/bin/），切换 node 大版本后需重跑本脚本。
BIN_REAL="$BIN"

# ---------- 2. 注册 DSH ----------
if [ "$ONLY_CODEX" -eq 0 ]; then
  say "2/4 注册 DSH"
  if [ ! -f "$DSH_PATCH" ]; then
    echo "⚠️  找不到 $DSH_PATCH，跳过（非 web profile 用户请手动加）"
  elif grep -q "mcp-bizard" "$DSH_PATCH"; then
    echo "已存在 mcp-bizard 配置，跳过（如需改路径请手动编辑）"
  else
    cp -a "$DSH_PATCH" "$DSH_PATCH.bak-$(date +%Y%m%d%H%M%S)"
    python3 - "$DSH_PATCH" "$BIN_REAL" <<'PY'
import sys, pathlib
patch, binpath = pathlib.Path(sys.argv[1]), sys.argv[2]
s = patch.read_text(encoding="utf-8")
block = f"""
    - id: mcp-bizard
      name: "@deepseek-ai/dsh-mcp-client"
      config:
        serverName: bizard
        transport: stdio
        command: {binpath}
        args: []
"""
anchor = "    - id: mcp-tavily\n"
if anchor not in s:
    sys.exit("未找到插入锚点 mcp-tavily，请手动添加")
patch.write_text(s.replace(anchor, block.lstrip("\n") + "\n" + anchor, 1), encoding="utf-8")
print("已插入 mcp-bizard")
PY
    echo "✅ 写入 $DSH_PATCH（已备份）"
    echo "   验证：dsh --profile web --dump-config | grep -A6 mcp-bizard"
    echo "   profile 设置了 patchReload: live，改动会自动热加载；否则重启 dsh web 生效"
  fi
fi

# ---------- 3. 注册 Codex ----------
if [ "$ONLY_DSH" -eq 0 ]; then
  say "3/4 注册 Codex"
  if [ ! -f "$CODEX_CFG" ]; then
    echo "⚠️  找不到 $CODEX_CFG，跳过"
  elif grep -q "\[mcp_servers.bizard\]" "$CODEX_CFG"; then
    echo "已存在 [mcp_servers.bizard]，跳过"
  else
    cp -a "$CODEX_CFG" "$CODEX_CFG.bak-$(date +%Y%m%d%H%M%S)"
    cat >> "$CODEX_CFG" <<EOF

[mcp_servers.bizard]
command = "$BIN_REAL"
args = []
EOF
    echo "✅ 追加到 $CODEX_CFG（已备份）"
    echo "   验证：codex mcp list | grep bizard"
  fi
fi

# ---------- 4. 冒烟测试 ----------
say "4/4 冒烟测试"
python3 "$(dirname "$0")/02_bizard_client.py" tools 2>&1 | head -6

say "完成"
echo "命令行检索:  python3 02_bizard_client.py search 小提琴图 --zh -n 5"
echo "Agent 内使用: 直接调 mcp__bizard__bizard_search 等 8 个工具"

#!/usr/bin/env python3
"""
Bizard MCP 本地 stdio 客户端（仅用标准库，无需装 mcp SDK）。

用途：脱离 Agent 直接查图。Agent 里用 mcp__bizard__* 工具；命令行/脚本里用这个。

用法：
    python3 02_bizard_client.py cats                     # 列出 40 个类别 + 图数
    python3 02_bizard_client.py search 小提琴图 --zh -n 5  # 中文检索
    python3 02_bizard_client.py search volcano -n 10     # 英文检索
    python3 02_bizard_client.py search heatmap -c Correlation   # 限定类别
    python3 02_bizard_client.py tutorial ViolinPlot-5    # 教程元信息
    python3 02_bizard_client.py code ViolinPlot-5 --lang r      # 可运行代码
    python3 02_bizard_client.py adapt ViolinPlot-5 --lang r --ctx "长表 gene/group/expr"
    python3 02_bizard_client.py index                    # 导出 charts_index.csv（982 图全量索引）

也可直接内联调用：
    from importlib import import_module
    bz = import_module("02_bizard_client")   # 目录名含数字，需按路径导入
    print(bz.call_tool("bizard_search", {"query": "violin", "limit": 3}))
"""

from __future__ import annotations

import argparse
import json
import os
import select
import shutil
import subprocess
import sys

CMD = os.environ.get("BIZARD_MCP_CMD") or shutil.which("bizard-mcp") or "bizard-mcp"
DEFAULT_TIMEOUT = float(os.environ.get("BIZARD_MCP_TIMEOUT", "60"))


class BizardClient:
    """最小 MCP stdio 客户端：initialize → 调用工具 → 关闭。"""

    def __init__(self, command: str = CMD, timeout: float = DEFAULT_TIMEOUT):
        self.timeout = timeout
        self._id = 0
        self.proc = subprocess.Popen(
            [command],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )
        self._handshake()

    def _readline(self):
        r, _, _ = select.select([self.proc.stdout], [], [], self.timeout)
        if not r:
            raise TimeoutError(f"bizard-mcp 在 {self.timeout}s 内无响应")
        return self.proc.stdout.readline()

    def _send(self, payload: dict) -> None:
        self.proc.stdin.write(json.dumps(payload) + "\n")
        self.proc.stdin.flush()

    def _rpc(self, method: str, params: dict | None = None) -> dict:
        self._id += 1
        rid = self._id
        self._send({"jsonrpc": "2.0", "id": rid, "method": method, "params": params or {}})
        while True:
            line = self._readline()
            if not line:
                raise RuntimeError("bizard-mcp 进程提前退出；检查是否已安装：npm i -g @openbiox/bizard-mcp")
            try:
                msg = json.loads(line)
            except json.JSONDecodeError:
                continue  # 非 JSON 诊断行，跳过
            if msg.get("id") != rid:
                continue  # 服务端通知（如内容包热更新提示），跳过
            if "error" in msg:
                raise RuntimeError(msg["error"].get("message", str(msg["error"])))
            return msg["result"]

    def _handshake(self) -> None:
        result = self._rpc(
            "initialize",
            {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "hz-bizard-client", "version": "1.0"},
            },
        )
        self.server_info = result.get("serverInfo", {})
        self._send({"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}})

    def call_tool(self, name: str, arguments: dict | None = None) -> str:
        """调用工具，返回原始文本块。工具报错时抛 RuntimeError。"""
        result = self._rpc("tools/call", {"name": name, "arguments": arguments or {}})
        text = "".join(b.get("text", "") for b in result.get("content", []) if b.get("type") == "text")
        if result.get("isError"):
            raise RuntimeError(text.strip() or "未知错误")
        return text

    def list_tools(self) -> list[dict]:
        return self._rpc("tools/list").get("tools", [])

    def read_resource(self, uri: str) -> str:
        result = self._rpc("resources/read", {"uri": uri})
        return "".join(c.get("text", "") for c in result.get("contents", []))

    def close(self) -> None:
        try:
            self.proc.terminate()
            self.proc.wait(timeout=5)
        except Exception:
            self.proc.kill()

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()


def call_tool(name: str, arguments: dict | None = None) -> str:
    """一次性调用的便捷入口。"""
    with BizardClient() as c:
        return c.call_tool(name, arguments)


def load_json(text: str):
    """工具返回的是 JSON 字符串；纯文本错误信息原样返回字符串。"""
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return text


# ---------------- 子命令 ----------------

def cmd_cats(args) -> int:
    cats = load_json(call_tool("bizard_list_categories"))
    if isinstance(cats, dict):
        cats = cats.get("items", [])
    cats = sorted(cats, key=lambda c: -c["count"])
    print(f"{'类别':<26}{'中文名':<10}{'图数':>5}")
    print("-" * 43)
    for c in cats:
        print(f"{c['name']:<26}{c.get('nameZh') or '-':<10}{c['count']:>5}")
    print("-" * 43)
    print(f"{'合计':<26}{'':<10}{sum(c['count'] for c in cats):>5}")
    nz = [c["name"] for c in cats if not c.get("nameZh")]
    if nz:
        print(f"\n注意：{len(nz)}/{len(cats)} 个类别无中文名，中文检索时这些类别可能漏召回，改用英文关键词。")
    return 0


def cmd_search(args) -> int:
    payload = {"query": args.query, "limit": args.num}
    if args.zh:
        payload["language"] = "zh"
    if args.category:
        payload["category"] = args.category
    data = load_json(call_tool("bizard_search", payload))
    if isinstance(data, str):
        print(data)
        return 1
    results = data.get("results") or []
    if not results:
        print("无命中。相邻类别建议：")
        for s in (data.get("suggestions") or []):
            print(f"  - {s['id']:<22}{s['name']}  {s['tutorialUrl']}")
        return 0
    for i, r in enumerate(results, 1):
        print(f"[{i}] {r['id']}  |  {r['name']}")
        print(f"    type: {r.get('type', '-')}")
        print(f"    {r.get('tutorialUrl', '')}")
        print(f"    img: {r.get('imageUrl', '')}")
        if r.get("description"):
            print(f"    {r['description'][:120]}")
        print()
    return 0


def cmd_tutorial(args) -> int:
    print(json.dumps(load_json(call_tool("bizard_get_tutorial", {"chartId": args.chart_id})),
                     ensure_ascii=False, indent=2))
    return 0


def cmd_code(args) -> int:
    text = call_tool("bizard_get_code", {"chartId": args.chart_id, "language": args.lang})
    data = load_json(text)
    if isinstance(data, str):
        print(data)
        return 1
    if not args.raw:
        print(f"# chart: {args.chart_id}  language: {args.lang}")
        print(f"# packages: {', '.join(data.get('packages') or [])}")
        for u in (data.get("dataUrls") or []):
            print(f"# data: {u}")
        print("# " + "-" * 60)
    print(data.get("code", ""))
    if not args.raw and data.get("license"):
        print(f"\n# License: {data['license']}")
    return 0


def cmd_adapt(args) -> int:
    payload = {"chartId": args.chart_id, "language": args.lang}
    if args.ctx:
        payload["userDataContext"] = args.ctx
    data = load_json(call_tool("bizard_adapt_guidance", payload))
    if isinstance(data, str):
        print(data)
        return 1
    for sec in data.get("sections", []):
        print(f"## {sec['title']}")
        for item in sec.get("items", []):
            print(f"  - {item}")
        print()
    return 0


def _find_snapshot(bundle_path: str | None = None) -> str | None:
    """定位安装包内的内容快照 bundle.json（离线全量数据，含语言/依赖）。"""
    if bundle_path:
        return bundle_path
    if os.environ.get("BIZARD_BUNDLE"):
        return os.environ["BIZARD_BUNDLE"]
    try:
        root = subprocess.run(["npm", "root", "-g"], capture_output=True, text=True, timeout=30)
        cand = os.path.join(root.stdout.strip(), "@openbiox/bizard-mcp/dist/snapshot/bundle.json")
        if os.path.isfile(cand):
            return cand
    except Exception:
        pass
    return None


def cmd_index(args) -> int:
    """导出全量图例索引 CSV。

    优先读安装包内快照（字段最全，含语言/依赖/图片），
    找不到时退回 catalog 资源（仅 id/名称/类别/链接）。
    """
    import csv

    snap = _find_snapshot(getattr(args, "bundle", None))
    if snap:
        with open(snap, encoding="utf-8") as fh:
            bundle = json.load(fh)
        langs: dict[str, set] = {}
        for s in bundle.get("codeSnippets", []):
            langs.setdefault(s["chartId"], set()).add(s["lang"])
        rows = []
        for ch in bundle["charts"]:
            rows.append([
                ch["id"], ch["name"], ch.get("nameZh") or "",
                ch.get("level1", ""), ch.get("level2", ""), ch.get("type", ""),
                "/".join(sorted(langs.get(ch["id"], []))),
                ch.get("tutorialUrl", ""), ch.get("imageUrl", ""), ch.get("description", ""),
            ])
        header = ["id", "name", "nameZh", "level1", "level2", "type",
                  "languages", "tutorialUrl", "imageUrl", "description"]
        source = f"快照 {snap}"
    else:
        with BizardClient() as c:
            entries = json.loads(c.read_resource("bizard://catalog"))
        rows = [[e["id"], e["name"], "", e.get("level1", ""), e.get("level2", ""),
                 "", "", e.get("tutorialUrl", ""), "", ""] for e in entries]
        header = ["id", "name", "nameZh", "level1", "level2", "type",
                  "languages", "tutorialUrl", "imageUrl", "description"]
        source = "catalog 资源（精简字段）"

    with open(args.out, "w", newline="", encoding="utf-8") as fh:
        w = csv.writer(fh)
        w.writerow(header)
        w.writerows(rows)
    print(f"已写入 {args.out}（{len(rows)} 图，来源：{source}）")
    return 0


def cmd_tools(args) -> int:
    with BizardClient() as c:
        print(f"server: {c.server_info}")
        for t in c.list_tools():
            req = (t.get("inputSchema") or {}).get("required") or []
            props = list(((t.get("inputSchema") or {}).get("properties") or {}))
            print(f"\n{t['name']}")
            print(f"  {t.get('description', '').strip()}")
            print(f"  参数: {props}  必填: {req}")
    return 0


def main() -> int:
    p = argparse.ArgumentParser(description="Bizard MCP 本地客户端")
    sub = p.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("cats", help="列出全部类别与图数")
    sp.set_defaults(func=cmd_cats)

    sp = sub.add_parser("search", help="关键词检索")
    sp.add_argument("query")
    sp.add_argument("-n", "--num", type=int, default=5, help="返回条数（1-20，默认 5）")
    sp.add_argument("--zh", action="store_true", help="中文结果优先")
    sp.add_argument("-c", "--category", help="限定类别，须为 cats 里的 name（英文）")
    sp.set_defaults(func=cmd_search)

    sp = sub.add_parser("tutorial", help="教程元信息")
    sp.add_argument("chart_id")
    sp.set_defaults(func=cmd_tutorial)

    sp = sub.add_parser("code", help="取可运行代码")
    sp.add_argument("chart_id")
    sp.add_argument("--lang", choices=["r", "python", "julia"], default="r")
    sp.add_argument("--raw", action="store_true", help="只打印代码，不打印依赖头")
    sp.set_defaults(func=cmd_code)

    sp = sub.add_parser("adapt", help="适配自己数据的检查清单")
    sp.add_argument("chart_id")
    sp.add_argument("--lang", choices=["r", "python", "julia"], default="r")
    sp.add_argument("--ctx", help="描述你的数据形状，如 '长表 gene/group/expr'")
    sp.set_defaults(func=cmd_adapt)

    sp = sub.add_parser("index", help="导出 charts_index.csv")
    sp.add_argument("-o", "--out", default="charts_index.csv")
    sp.set_defaults(func=cmd_index)

    sp = sub.add_parser("tools", help="列出工具与参数")
    sp.set_defaults(func=cmd_tools)

    args = p.parse_args()
    try:
        return args.func(args)
    except (RuntimeError, TimeoutError) as e:
        print(f"错误: {e}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# ============================================================
# FigureYa 代码拉取（blobless + sparse checkout）
#
# 为什么要特殊处理：上游仓库 6.4 GB，其中 318 个模块的 .Rmd + .R 只有 7 MB，
# 其余全是 .txt/.csv/.gz/RData 数据与渲染产物。直接 git clone 会拉 3.5 GB+。
#
# 本脚本用三层过滤把体积压到 ~76 MB：
#   1. --filter=blob:none  延迟下载文件内容（只取提交与目录树）
#   2. --depth 1           只要最新一次提交，不要历史
#   3. sparse-checkout     只检出代码 + 元数据 + 教程文本（规则见 sparse-checkout.rules）
#
# 用法:
#   ./01_clone.sh              # 首次克隆（约 76 MB）
#   ./01_clone.sh --update     # 已有则更新到上游最新
#   ./01_clone.sh --full       # 需要数据文件时全量检出（约 3.5 GB，谨慎）
# ============================================================
set -euo pipefail
cd "$(dirname "$0")"

REPO="git@github.com:Han-Z02/FigureYa.git"
UPSTREAM="git@github.com:ying-ge/FigureYa.git"
DIR="reference"
RULES="sparse-checkout.rules"

MODE="init"
for a in "$@"; do
  case "$a" in
    --update) MODE="update" ;;
    --full)   MODE="full" ;;
    -h|--help) sed -n '2,20p' "$0"; exit 0 ;;
    *) echo "未知参数: $a"; exit 1 ;;
  esac
done

[ -f "$RULES" ] || { echo "❌ 缺少 $RULES"; exit 1; }

apply_sparse() {
  git sparse-checkout init --no-cone
  cp "../$RULES" .git/info/sparse-checkout
  git checkout main >/dev/null 2>&1 || git read-tree -mu HEAD
}

case "$MODE" in
  init)
    if [ -d "$DIR/.git" ]; then
      echo "⚠️  $DIR 已存在，改用 --update 增量更新"
      exit 0
    fi
    echo "=== 克隆 $REPO（blobless, depth=1）==="
    git clone --filter=blob:none --no-checkout --depth 1 "$REPO" "$DIR"
    cd "$DIR"
    echo "=== 应用 sparse 规则 ==="
    apply_sparse
    ;;
  update)
    [ -d "$DIR/.git" ] || { echo "❌ 还没克隆过，先跑 ./01_clone.sh"; exit 1; }
    cd "$DIR"
    echo "=== 拉取上游最新 ==="
    git fetch --filter=blob:none --depth 1 origin main
    git reset --hard origin/main
    apply_sparse
    ;;
  full)
    [ -d "$DIR/.git" ] || { echo "❌ 还没克隆过，先跑 ./01_clone.sh"; exit 1; }
    cd "$DIR"
    echo "=== 全量检出（约 3.5 GB，会下载所有数据文件）==="
    read -rp "确认继续？[y/N] " ans
    [ "$ans" = "y" ] || exit 0
    git sparse-checkout disable
    git checkout main
    ;;
esac

cd - >/dev/null
echo
echo "=== 体积 ==="
du -sh "$DIR"
echo
echo "=== 内容 ==="
echo "模块数      : $(ls -d "$DIR"/FigureYa*/ 2>/dev/null | wc -l)"
echo "Rmd         : $(find "$DIR" -name '*.Rmd' -not -path '*/.git/*' | wc -l)"
echo "R 脚本      : $(find "$DIR" -name '*.R' -not -path '*/.git/*' | wc -l)"
echo "教程文本    : $(ls "$DIR"/texts 2>/dev/null | wc -l)"
echo
echo "下一步: python3 03_build_index.py     # 重建索引"

#!/usr/bin/env python3
"""
FigureYa 索引生成器 —— 从 clone 的 reference/ 解析 317 个模块，输出可检索索引。

输入（均在 reference/ 下）：
    chapters.json      模块清单：folder / html / thumb / text
    file_list.json     每模块的文件列表（判断有无 Rmd / R 脚本）
    texts/*.txt        每个模块的教程全文（含需求描述、应用场景、完整代码）

输出：
    modules_index.csv  一行一个模块：编号、Rmd、脚本数、需求描述、教程与缩略图链接
    categories.md      按 categories.txt 的分组盘点

用法：
    python3 03_build_index.py                       # 在 02_figureya/ 下执行
    python3 03_build_index.py --ref reference --out .
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import re
import sys

SITE = "https://ying-ge.github.io/FigureYa"
REPO = "https://github.com/Han-Z02/FigureYa"

# 需求段落的结束标志（下一小节标题）
_END_MARKERS = re.compile(
    r"(应用场景|Application\s*scenarios?|环境设置|Environment\s*settings?|"
    r"数据准备|Data\s*preparation|代码|Code\b|\n\d+\n)",
    re.I,
)
# 正文开头要跳过的元信息行
_SKIP_LINE = re.compile(
    r"^(FigureYa\S*|Code|Show All Code|Hide All Code|Author\(s\)|Reviewer\(s\)|"
    r"Date|Academic Citation|If you use this code|Xiaofan Lu|我们的宗旨|:|"
    r"\d+\s*$|\d+\s*[A-Z])",
    re.I,
)
# 引用/致谢段落（出现在需求段之前，需先剔除）
_CITATION_BLOCK = re.compile(
    r"(?:If you use this code.*?https://doi\.org/\S+|"
    r"Academic Citation.*?(?=需求描述|Requirement|$)|"
    r"Xiaofan Lu,\s*et al\..*?iMetaMed\.?|"
    r"\d+\s*Framework for Enhancing.*?iMetaMed\.?)",
    re.S | re.I,
)


def _norm(text: str) -> str:
    return re.sub(r"\s+", " ", text).strip()


def _clean_head(raw: str) -> str:
    """砍掉正文开头的引用块与编号小节标号，避免污染需求描述。"""
    # 先把首个「需求描述」之前的内容整体裁掉前缀噪声
    idx = raw.lower().find("需求描述")
    if idx == -1:
        idx = raw.lower().find("requirement")
    if idx > 0:
        head, tail = raw[:idx], raw[idx:]
        # 若前半段包含引用信息，直接丢弃；否则保留（老模块的需求常写在开头）
        if _CITATION_BLOCK.search(head) or "doi.org" in head.lower() or "iMetaMed" in head:
            prefix = "需求描述 " if "需求描述" in tail[:12] else ""
            raw = prefix + tail
    return _CITATION_BLOCK.sub("", raw)


def extract_needs(path: str) -> str:
    """从教程全文里抠出「需求描述 / Requirement」段落。

    优先按小节标题切片；没有该标题的老模块，跳过元信息后取正文前几行。
    """
    if not path or not os.path.isfile(path):
        return ""
    raw = _clean_head(open(path, encoding="utf-8", errors="replace").read())

    m = re.search(
        r"需求描述\s*(?:Requirement)?\s*(?:description)?(.*)", raw, re.S | re.I
    )
    if m:
        body = m.group(1)
        end = _END_MARKERS.search(body)
        if end:
            body = body[: end.start()]
        body = _norm(body)
        # 反复剥离开头的标题词（原文常连续出现「需求描述 需求描述 Requirement description」）
        _HEAD = re.compile(
            r"^(?:需求描述|Requirement(?:\s*Description)?|Description[s]?|描述)\s*",
            re.I,
        )
        for _ in range(6):
            stripped = _HEAD.sub("", body)
            if stripped == body:
                break
            body = stripped
        if len(body) > 20:
            return body[:400]
    # 兜底：跳过元信息，取正文
    picked: list[str] = []
    for line in raw.split("\n")[:120]:
        ls = line.strip()
        if not ls or _SKIP_LINE.search(ls):
            continue
        if "doi.org" in ls.lower() or "cite our publication" in ls.lower():
            continue
        picked.append(ls)
        if len(" ".join(picked)) > 320:
            break
    return _norm(" ".join(picked))[:400]


def load_modules(ref: str) -> list[dict]:
    """按模块（folder）聚合 chapters。

    chapters.json 里一个模块可能有多条记录（多分册 Rmd，如 FigureYa63pubmedMiningV2
    有 3 章），索引应按模块去重，分册数单独记一列。
    """
    ch_path = os.path.join(ref, "chapters.json")
    fl_path = os.path.join(ref, "file_list.json")
    if not os.path.isfile(ch_path):
        sys.exit(f"找不到 {ch_path}；请先在 {os.path.dirname(ref)} 执行 01_clone.sh")
    chapters = json.load(open(ch_path, encoding="utf-8"))
    file_list = json.load(open(fl_path, encoding="utf-8")) if os.path.isfile(fl_path) else {}

    grouped: dict[str, list[dict]] = {}
    for c in chapters:
        grouped.setdefault(c["folder"], []).append(c)

    rows = []
    for folder, chapters_of in grouped.items():
        files = [f["name"] for f in file_list.get(folder, [])]
        rmd = next((f for f in files if f.lower().endswith(".rmd")), "")
        scripts = [
            f for f in files
            if f.lower().endswith((".r", ".py")) and f != "install_dependencies.R"
        ]
        # 需求描述取第一份非空的分册
        needs, site = "", ""
        for c in chapters_of:
            if not needs and c.get("text"):
                needs = extract_needs(os.path.join(ref, c["text"]))
            if not site and c.get("html"):
                site = f"{SITE}/{c['html']}"
        # html 字段缺失时用实际 Rmd 文件名拼（有 7 个模块 Rmd 名与目录名不同，
        # 如 FigureYa8radar/FigureYa8radarChart.Rmd，按目录名拼会 404）
        if not site and rmd:
            site = f"{SITE}/{folder}/{os.path.splitext(rmd)[0]}.html"
        thumb = next((c["thumb"] for c in chapters_of if c.get("thumb")), "")

        rows.append(
            {
                "id": folder,
                "rmd": rmd,
                "n_scripts": len(scripts),
                "has_install": "install_dependencies.R" in files,
                "n_files": len(files),
                "n_chapters": len(chapters_of),
                "needs": needs,
                "site": site,
                "thumb": f"{REPO}/raw/main/{thumb}" if thumb else "",
            }
        )
    rows.sort(key=lambda r: r["id"])
    return rows


def write_csv(rows: list[dict], out: str) -> str:
    path = os.path.join(out, "modules_index.csv")
    with open(path, "w", newline="", encoding="utf-8") as fh:
        w = csv.writer(fh)
        w.writerow(["id", "rmd", "n_scripts", "has_install", "n_files",
                    "n_chapters", "needs", "site", "thumb"])
        for r in rows:
            w.writerow([
                r["id"], r["rmd"], r["n_scripts"], "Y" if r["has_install"] else "N",
                r["n_files"], r["n_chapters"], r["needs"], r["site"], r["thumb"],
            ])
    return path


def write_categories(ref: str, out: str, rows: list[dict]) -> str:
    """把 categories.txt 的主题分组转成 markdown。

    文件结构：开头是 6 个主题的「标题 + 副标题」总览（各占一行，无链接），
    空行之后进入正文，每组以 `主题，公众号链接` 开头，下面跟模块行。

    注意：分隔符是全角逗号「，」，而主题标题自身也含它
    （如「圆圈图小结｜美图如云，理性选择，https://...」），
    所以不能用 split(',')，必须先按 "http" 切开再取前半段作标题。
    """
    src = os.path.join(ref, "categories.txt")
    if not os.path.isfile(src):
        return ""
    ids = {r["id"] for r in rows}

    sections: list[tuple[str, str, list[str]]] = []
    cur_title, cur_link, cur_mods = None, "", []

    for raw in open(src, encoding="utf-8", errors="replace"):
        line = raw.strip().lstrip("\ufeff")
        if not line:
            continue
        if line.startswith("FigureYa"):
            if cur_title is not None:
                cur_mods.append(line)
            continue
        if "http" not in line:
            continue        # 开头的总览双行标题，跳过
        # 主题行
        if cur_title is not None:
            sections.append((cur_title, cur_link, cur_mods))
        head, _, url = line.partition("http")
        url = "http" + url
        title = head.rstrip("，, ").strip()
        cur_title, cur_link, cur_mods = title, url.strip(), []
    if cur_title is not None:
        sections.append((cur_title, cur_link, cur_mods))

    lines = [
        "# FigureYa 主题分组盘点",
        "",
        "> 由 `03_build_index.py` 从仓库 `categories.txt` 生成；分组与顺序遵循上游，"
        "模块名已链接到在线教程页。",
        "",
    ]
    total_listed = 0
    for title, link, mods in sections:
        present = [m for m in mods if m in ids]
        missing = [m for m in mods if m not in ids]
        total_listed += len(present)
        lines.append(f"## {title}")
        lines.append("")
        if link:
            lines.append(f"> 主题小结：{link}")
            lines.append("")
        lines.append(f"**{len(present)} 个模块**")
        lines.append("")
        for m in present:
            lines.append(f"- [`{m}`]({SITE}/{m}/{m}.html)")
        if missing:
            lines.append("")
            lines.append(f"> 仓库中未找到（可能已改名/删除）：{', '.join(missing)}")
        lines.append("")

    covered = {m for _, _, mods in sections for m in mods}
    leftover = sorted(ids - covered)
    lines.append("## 未归入主题的模块")
    lines.append("")
    lines.append(
        f"**{len(leftover)} 个模块** —— `categories.txt` 只整理了 6 个主题，"
        "其余模块未归类（多为后期新增或专项工具），可按需自行分组。"
    )
    lines.append("")
    for m in leftover:
        lines.append(f"- [`{m}`]({SITE}/{m}/{m}.html)")
    lines.append("")

    path = os.path.join(out, "categories.md")
    open(path, "w", encoding="utf-8").write("\n".join(lines) + "\n")
    print(f"主题分组    : {len(sections)} 组覆盖 {total_listed} 模块；未归类 {len(leftover)}")
    return path


def main() -> int:
    ap = argparse.ArgumentParser(description="生成 FigureYa 模块索引")
    ap.add_argument("--ref", default="reference", help="clone 出来的仓库目录")
    ap.add_argument("--out", default=".", help="索引输出目录")
    a = ap.parse_args()

    rows = load_modules(a.ref)
    csv_path = write_csv(rows, a.out)
    cat_path = write_categories(a.ref, a.out, rows)

    with_rmd = sum(1 for r in rows if r["rmd"])
    with_needs = sum(1 for r in rows if r["needs"])
    print(f"模块总数    : {len(rows)}")
    print(f"有 Rmd 主脚本: {with_rmd}")
    print(f"有需求描述  : {with_needs}")
    print(f"平均脚本数  : {sum(r['n_scripts'] for r in rows) / max(len(rows), 1):.2f}")
    print(f"已写出      : {csv_path}")
    if cat_path:
        print(f"            : {cat_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
FigureYa 内容打包器 —— 把 reference/ 下的 Rmd 与教程全文压成单个 JSON 快照。

为什么要打包：MCP server 每次启动都去扫 313 个 Rmd（328 个文件、4569 个 chunk）
要 1-2 秒且依赖 reference/ 目录存在。打包成一个 bundle.json 后可随包分发，
server 启动只需 json.load，且离线可用 —— 与 Bizard MCP 的做法一致。

输入（reference/ 下，由 01_clone.sh 拉取）：
    FigureYa*/…Rmd      主脚本 + 代码块
    FigureYa*/install_dependencies.R
    chapters.json       模块 → html/thumb/text 映射
    texts/*.txt         教程全文（含需求描述、应用场景）
    categories.txt      6 个主题分组

输出：
    snapshot/bundle.json

用法：
    python3 build_bundle.py                 # 默认读 ./reference，写 ./snapshot
    python3 build_bundle.py --ref reference --out snapshot
"""

from __future__ import annotations

import argparse
import glob
import json
import os
import re
import sys

# ---- Rmd 解析正则 ----
# ```{r ...} 或 ```{r} —— 支持 chunk 选项里的任意内容（含引号、逗号）
CHUNK_RE = re.compile(r"^```\{r([^}]*)\}[ \t]*\r?\n(.*?)^```[ \t]*$", re.M | re.S)
HEADING_RE = re.compile(r"^(#{1,4})[ \t]+(.+?)[ \t]*#*[ \t]*$", re.M)
PKG_RE = re.compile(
    r'(?:library|require|requireNamespace)\(\s*["\']?([A-Za-z][A-Za-z0-9._]*)["\']?\s*[,)]'
)
# install_dependencies.R 里的包清单
DEPS_CRAN_RE = re.compile(r"cran_packages\s*<-\s*c\(([^)]*)\)", re.S)
DEPS_BIOC_RE = re.compile(r"bioc_packages\s*<-\s*c\(([^)]*)\)", re.S)

# 需求描述 / 应用场景这些标准节，用来给 section 打标签
SECTION_ALIASES = {
    "requirement": ("需求描述", "requirement", "需求"),
    "scenario": ("应用场景", "使用场景", "application scenario", "usage scenario", "scenario"),
    "environment": ("环境设置", "environment setting"),
    "input": ("输入文件", "输入数据", "input data", "input file", "输入文件的获得"),
    "draw": ("开始画图", "start to draw", "draw"),
    "postprocess": ("后期处理", "post-processing", "postprocess"),
}


def _strip_chunks(text: str) -> str:
    """移除代码块，只留说明性 Markdown。

    必须做这一步：R chunk 里的 `# 注释` 会被标题正则误判成 Markdown 标题
    （FigureYa101PCA 有 106 个「标题」，其中绝大多数是 chunk 内的注释）。
    """
    return CHUNK_RE.sub("", text)


def split_frontmatter(text: str) -> tuple[dict, str]:
    """拆 YAML front matter。只做浅解析 —— 不引入 PyYAML 依赖。"""
    if not text.startswith("---"):
        return {}, text
    end = text.find("\n---", 3)
    if end == -1:
        return {}, text
    head = text[3:end]
    body = text[end + 4 :]
    fm: dict = {}
    m = re.search(r'^title:\s*["\']?(.+?)["\']?\s*$', head, re.M)
    if m:
        fm["title"] = m.group(1)
    m = re.search(r'^\s*author:\s*["\']?(.+?)["\']?\s*$', head, re.M)
    if m:
        fm["author"] = m.group(1)
    m = re.search(r'^\s*reviewer:\s*["\']?(.+?)["\']?\s*$', head, re.M)
    if m:
        fm["reviewer"] = m.group(1)
    return fm, body


def parse_sections(body_no_chunks: str) -> list[dict]:
    """按 ## 级标题切分说明性章节。返回 [{title, level, content}]。"""
    marks = [(m.start(), len(m.group(1)), m.group(2).strip()) for m in HEADING_RE.finditer(body_no_chunks)]
    if not marks:
        return []
    out = []
    for i, (pos, level, title) in enumerate(marks):
        end = marks[i + 1][0] if i + 1 < len(marks) else len(body_no_chunks)
        content = body_no_chunks[pos:end]
        content = content.split("\n", 1)[1] if "\n" in content else ""
        content = re.sub(r"\n{3,}", "\n\n", content).strip()
        if not content or not title:
            continue
        out.append({"title": title, "level": level, "content": content[:4000]})
    return out


def tag_section(title: str) -> str | None:
    """给标准节打标签，便于按语义取「需求描述」而不用猜标题写法。"""
    low = title.lower()
    for tag, keys in SECTION_ALIASES.items():
        if any(k.lower() in low for k in keys):
            return tag
    return None


def parse_chunks(text: str) -> list[dict]:
    """抽取 R 代码块。skip setup/eval=FALSE 这类不产生内容的块单列标记。"""
    chunks = []
    for i, m in enumerate(CHUNK_RE.finditer(text)):
        header = m.group(1).strip().strip(",")
        code = m.group(2).rstrip()
        if not code.strip():
            continue
        label = ""
        lm = re.match(r"^([A-Za-z0-9_.-]+)", header)
        if lm:
            label = lm.group(1)
        chunks.append(
            {
                "i": len(chunks),
                "label": label,
                "header": header[:200],
                "setup": bool(re.search(r"include\s*=\s*FALSE|eval\s*=\s*FALSE", header, re.I)),
                "code": code[:20000],
                "lines": code.count("\n") + 1,
            }
        )
    return chunks


def extract_packages(text: str, deps_script: str) -> dict:
    """从 Rmd 的 library() 调用 + install_dependencies.R 的清单合并出依赖。"""
    pkgs = set(PKG_RE.findall(text))
    cran, bioc = [], []
    for pat, bucket in ((DEPS_CRAN_RE, cran), (DEPS_BIOC_RE, bioc)):
        m = pat.search(deps_script or "")
        if m:
            bucket.extend(
                x.strip().strip("\"'")
                for x in m.group(1).split(",")
                if x.strip().strip("\"'")
            )
    pkgs.update(cran)
    pkgs.update(bioc)
    # 过滤明显不是包名的
    bad = {"install", "library", "require", "c", "TRUE", "FALSE", "T", "F"}
    return {
        "all": sorted(p for p in pkgs if p and p not in bad),
        "cran": sorted(set(cran)),
        "bioc": sorted(set(bioc)),
    }


def build(ref: str) -> dict:
    ch_path = os.path.join(ref, "chapters.json")
    chapters = json.load(open(ch_path, encoding="utf-8")) if os.path.isfile(ch_path) else []

    # folder -> 章节记录（多分册）
    by_folder: dict[str, list[dict]] = {}
    for c in chapters:
        by_folder.setdefault(c["folder"], []).append(c)

    # 目录名集合以实际 Rmd 所在目录为准
    folders = sorted(
        {os.path.basename(os.path.dirname(p)) for p in glob.glob(os.path.join(ref, "FigureYa*", "*.Rmd"))}
    )

    modules = []
    for folder in folders:
        mdir = os.path.join(ref, folder)
        rmd_paths = sorted(glob.glob(os.path.join(mdir, "*.Rmd")))
        if not rmd_paths:
            continue
        rmd_path = rmd_paths[0]
        raw = open(rmd_path, encoding="utf-8", errors="replace").read()
        fm, body = split_frontmatter(raw)
        body_no_chunks = _strip_chunks(body)

        deps_path = os.path.join(mdir, "install_dependencies.R")
        deps_script = open(deps_path, encoding="utf-8", errors="replace").read() if os.path.isfile(deps_path) else ""

        chapters_of = by_folder.get(folder, [])
        thumb = next((c["thumb"] for c in chapters_of if c.get("thumb")), "")
        html = next((c["html"] for c in chapters_of if c.get("html")), "")

        sections = parse_sections(body_no_chunks)
        for s in sections:
            s["tag"] = tag_section(s["title"])

        modules.append(
            {
                "id": folder,
                "rmd": os.path.basename(rmd_path),
                "title": fm.get("title", folder),
                "author": fm.get("author", ""),
                "reviewer": fm.get("reviewer", ""),
                "site": f"https://ying-ge.github.io/FigureYa/{html}" if html else "",
                "thumb": f"https://raw.githubusercontent.com/Han-Z02/FigureYa/main/{thumb}" if thumb else "",
                "n_chapters": len(chapters_of),
                "sections": sections,
                "chunks": parse_chunks(raw),
                "packages": extract_packages(raw, deps_script),
                "has_install_script": bool(deps_script),
                "deps_script": deps_script[:8000],
                "extra_scripts": sorted(
                    os.path.basename(p)
                    for p in glob.glob(os.path.join(mdir, "*.R"))
                    if os.path.basename(p) != "install_dependencies.R"
                )[:40],
            }
        )

    # 主题分组
    categories = []
    cat_path = os.path.join(ref, "categories.txt")
    if os.path.isfile(cat_path):
        cur = None
        for raw_line in open(cat_path, encoding="utf-8", errors="replace"):
            line = raw_line.strip().lstrip("\ufeff")
            if not line:
                continue
            if line.startswith("FigureYa"):
                if cur:
                    cur["modules"].append(line)
                continue
            if "http" not in line:
                continue
            if cur:
                categories.append(cur)
            head, _, url = line.partition("http")
            cur = {"name": head.rstrip("，, ").strip(), "url": ("http" + url).strip(), "modules": []}
        if cur:
            categories.append(cur)

    return {
        "meta": {
            "source": "https://github.com/ying-ge/FigureYa",
            "fork": "https://github.com/Han-Z02/FigureYa",
            "site": "https://ying-ge.github.io/FigureYa/",
            "license": "CC BY-NC-SA 4.0",
            "citation": "Lu X, et al. (2025). FigureYa: A Standardized Visualization Framework "
                        "for Enhancing Biomedical Data Interpretation and Research Efficiency. "
                        "iMetaMed. https://doi.org/10.1002/imm3.70005",
            "doi": "10.1002/imm3.70005",
            "n_modules": len(modules),
        },
        "categories": categories,
        "modules": modules,
    }


def main() -> int:
    ap = argparse.ArgumentParser(description="打包 FigureYa 内容快照")
    ap.add_argument("--ref", default="reference")
    ap.add_argument("--out", default="snapshot")
    a = ap.parse_args()

    if not os.path.isdir(a.ref):
        sys.exit(f"找不到 {a.ref}/；先跑 ./01_clone.sh")

    bundle = build(a.ref)
    os.makedirs(a.out, exist_ok=True)
    path = os.path.join(a.out, "bundle.json")
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(bundle, fh, ensure_ascii=False, separators=(",", ":"))

    size = os.path.getsize(path) / 1024 / 1024
    n_chunk = sum(len(m["chunks"]) for m in bundle["modules"])
    n_sec = sum(len(m["sections"]) for m in bundle["modules"])
    n_pkg = sum(len(m["packages"]["all"]) for m in bundle["modules"])
    print(f"模块      : {len(bundle['modules'])}")
    print(f"代码块    : {n_chunk}")
    print(f"说明章节  : {n_sec}")
    print(f"依赖声明项: {n_pkg}")
    print(f"主题分组  : {len(bundle['categories'])}")
    print(f"快照      : {path}  ({size:.1f} MB)")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
FigureYa MCP Server —— 把 313 个 R Markdown 图表模板变成可检索、可取码的 MCP 服务。

设计取舍（与 Bizard MCP 对齐，便于两者并列使用）：
  - 内容随包分发（snapshot/bundle.json，5.8 MB），启动只需 json.load，离线可用
  - 只读服务，不执行任何 R 代码；取回的代码由宿主 Agent/用户自己跑
  - 工具名 figureya_*，与 bizard_* 不冲突

工具：
  figureya_search           关键词检索模块（中英双语，命中标题/需求/依赖/代码）
  figureya_list_categories  6 个主题分组
  figureya_get_module       模块元信息：章节结构、依赖、代码块清单
  figureya_get_code         取可运行 R 代码（整篇 Rmd 或指定代码块）
  figureya_adapt_guidance   把模板适配到自己数据的检查清单
  figureya_cite             引用与许可

资源：
  figureya://catalog        模块清单 JSON
  figureya://module/{id}    单个模块完整内容

用法：
  python3 05_mcp_server.py                 # stdio（默认）
  python3 05_mcp_server.py --selftest      # 自检，不启动服务
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys

try:
    from mcp.server.fastmcp import FastMCP
except ImportError:  # pragma: no cover
    sys.exit("缺少 mcp SDK：pip install 'mcp[cli]'")

# FastMCP/mcp SDK 用标准 logging 往 stderr 打每个请求的 INFO（Processing request of type
# ...）。DSH/Codex 把 MCP 的 stderr 当错误日志收，逐条刷屏却看不出真问题。
# 这里把 mcp 相关 logger 抬到 WARNING，只保留异常。
import logging as _logging

for _name in ("mcp", "mcp.server", "mcp.server.lowlevel", "fastmcp", "httpx"):
    _logging.getLogger(_name).setLevel(_logging.WARNING)

HERE = os.path.dirname(os.path.abspath(__file__))
SNAPSHOT = os.environ.get("FIGUREYA_BUNDLE") or os.path.join(HERE, "snapshot", "bundle.json")
SERVER_NAME = "figureya"
SEARCH_LIMIT_MAX = 25

mcp = FastMCP("figureya")

_bundle: dict | None = None


def get_bundle() -> dict:
    """惰性加载快照。找不到时给出可操作的报错，而不是空列表。"""
    global _bundle
    if _bundle is None:
        if not os.path.isfile(SNAPSHOT):
            raise RuntimeError(
                f"找不到内容快照 {SNAPSHOT}。请先执行：\n"
                "  ./01_clone.sh              # 拉取 FigureYa 源码\n"
                "  python3 04_build_bundle.py # 生成 snapshot/bundle.json"
            )
        with open(SNAPSHOT, encoding="utf-8") as fh:
            _bundle = json.load(fh)
    return _bundle


def _mods() -> list[dict]:
    return get_bundle()["modules"]


def _by_id(mid: str) -> dict:
    for m in _mods():
        if m["id"] == mid:
            return m
    raise ValueError(f"未知模块 ID: {mid}（用 figureya_search 查可用 ID）")


# ---------------------------------------------------------------- 检索

def _haystack(m: dict) -> str:
    """拼出检索文本。需求描述权重最高，代码只取前 2000 字避免噪声。"""
    parts = [m["id"], m["title"], m.get("author", "")]
    for s in m["sections"]:
        if s.get("tag") in ("requirement", "scenario"):
            parts.append(s["title"])
            parts.append(s["content"])
    for s in m["sections"]:
        if s.get("tag") not in ("requirement", "scenario"):
            parts.append(s["title"])
    parts.extend(m["packages"]["all"])
    for c in m["chunks"][:6]:
        parts.append(c["code"][:2000])
    return "\n".join(parts).lower()


def _score(m: dict, terms: list[str], hay: str | None = None) -> int:
    """加权打分：模块 ID 命中权重最高，其次标题/需求，最后正文与代码。"""
    hay = hay if hay is not None else _haystack(m)
    low_id = m["id"].lower()
    low_title = m["title"].lower()
    req = "\n".join(s["content"] for s in m["sections"] if s.get("tag") == "requirement").lower()
    score = 0
    for t in terms:
        if t in low_id:
            score += 10
        if t in low_title:
            score += 6
        if t in req:
            score += 4
        if t in hay:
            score += 1
    return score


@mcp.tool()
def figureya_search(query: str, limit: int = 5, category: str = "") -> str:
    """检索 FigureYa 图表模板（中英文均可）。

    在模块编号、标题、需求描述、依赖包名和代码里做加权匹配。
    零命中时会按主题分组给出相邻建议。

    Args:
        query: 检索词，如「生存曲线」「circos」「WGCNA」「radar」
        limit: 返回条数 1-25，默认 5
        category: 限定主题分组名（先跑 figureya_list_categories 看可选值）
    """
    limit = max(1, min(int(limit), SEARCH_LIMIT_MAX))
    terms = [t for t in re.split(r"[\s,，、]+", query.strip().lower()) if len(t) >= 2]
    if not terms:
        return json.dumps({"error": "query 太短，至少 2 个字符"}, ensure_ascii=False)

    pool = _mods()
    if category:
        cats = get_bundle()["categories"]
        hit = next((c for c in cats if category.lower() in c["name"].lower()), None)
        if hit:
            allowed = set(hit["modules"])
            pool = [m for m in pool if m["id"] in allowed]
        else:
            return json.dumps(
                {
                    "error": f"未知主题 {category!r}",
                    "available": [c["name"] for c in cats],
                },
                ensure_ascii=False,
            )

    scored = []
    for m in pool:
        s = _score(m, terms)
        if s > 0:
            scored.append((s, m))
    scored.sort(key=lambda x: (-x[0], x[1]["id"]))

    if not scored:
        req = " ".join(s["content"] for s in [] )  # 占位，保持结构一致
        del req
        return json.dumps(
            {
                "results": [],
                "hint": "换英文关键词，或先 figureya_list_categories 看主题",
                "sample_ids": [m["id"] for m in pool[:5]],
            },
            ensure_ascii=False,
        )

    out = []
    for s, m in scored[:limit]:
        need = next((x["content"] for x in m["sections"] if x.get("tag") == "requirement"), "")
        out.append(
            {
                "id": m["id"],
                "title": m["title"],
                "score": s,
                "requirement": need[:400],
                "packages": m["packages"]["all"][:12],
                "n_chunks": len(m["chunks"]),
                "site": m["site"],
                "thumb": m["thumb"],
            }
        )
    return json.dumps({"results": out, "total_matched": len(scored)}, ensure_ascii=False, indent=2)


@mcp.tool()
def figureya_list_categories() -> str:
    """列出 6 个主题分组及每组模块数与公众号小结链接。"""
    cats = get_bundle()["categories"]
    items = [
        {
            "name": c["name"],
            "url": c["url"],
            "count": len([x for x in c["modules"] if any(m["id"] == x for m in _mods())]),
            "modules": c["modules"],
        }
        for c in cats
    ]
    covered = {x for c in cats for x in c["modules"]}
    leftover = [m["id"] for m in _mods() if m["id"] not in covered]
    return json.dumps(
        {"items": items, "uncategorized_count": len(leftover)}, ensure_ascii=False, indent=2
    )


# ---------------------------------------------------------------- 取内容

@mcp.tool()
def figureya_get_module(module_id: str, include_sections: bool = True) -> str:
    """取单个模块的结构化元信息：章节标题、依赖包、代码块清单、在线链接。

    不返回完整代码（用 figureya_get_code）。

    Args:
        module_id: 模块 ID，如 FigureYa101PCA
        include_sections: 是否返回章节正文（默认 True；只要目录可设 false）
    """
    m = _by_id(module_id)
    payload = {
        "id": m["id"],
        "title": m["title"],
        "author": m["author"],
        "reviewer": m["reviewer"],
        "site": m["site"],
        "thumb": m["thumb"],
        "packages": m["packages"],
        "has_install_script": m["has_install_script"],
        "extra_scripts": m["extra_scripts"],
        "chunks": [
            {"i": c["i"], "label": c["label"], "lines": c["lines"], "setup": c["setup"]}
            for c in m["chunks"]
        ],
    }
    if include_sections:
        payload["sections"] = [
            {"title": s["title"], "tag": s.get("tag"), "content": s["content"][:2000]}
            for s in m["sections"]
        ]
    return json.dumps(payload, ensure_ascii=False, indent=2)


@mcp.tool()
def figureya_get_code(module_id: str, chunk: int = -1, include_deps: bool = True) -> str:
    """取可运行的 R 代码。

    Args:
        module_id: 模块 ID
        chunk: 代码块序号（0 起）。-1 表示返回整篇 Rmd 的全部代码块；也可用 figureya_get_module 查清单
        include_deps: 是否附带 install_dependencies.R（默认 True，照着跑可自动装依赖）
    """
    m = _by_id(module_id)
    chunks = m["chunks"]
    if not chunks:
        return json.dumps({"error": f"{module_id} 没有可提取的 R 代码块"}, ensure_ascii=False)

    if chunk >= 0:
        if chunk >= len(chunks):
            return json.dumps(
                {"error": f"chunk 超出范围，该模块只有 {len(chunks)} 个代码块（0-{len(chunks)-1}）"},
                ensure_ascii=False,
            )
        picked = [chunks[chunk]]
    else:
        picked = chunks

    out = {
        "id": m["id"],
        "rmd": m["rmd"],
        "site": m["site"],
        "packages": m["packages"],
        "license": "CC BY-NC-SA 4.0",
        "code_blocks": [{"i": c["i"], "label": c["label"], "code": c["code"]} for c in picked],
    }
    if include_deps and m["has_install_script"]:
        out["install_dependencies.R"] = m["deps_script"]
    out["run_hint"] = (
        f"把代码存为 {m['rmd']} 同目录下的脚本，先 source(\"install_dependencies.R\"), "
        "再依次运行各代码块；数据文件需从上游仓库取（本地 sparse 检出不含数据）"
    )
    return json.dumps(out, ensure_ascii=False, indent=2)


@mcp.tool()
def figureya_adapt_guidance(module_id: str, user_data_context: str = "") -> str:
    """把模板适配到自己数据的检查清单。

    基于该模块实际用到的包、代码块结构和需求描述给出针对性提示。

    Args:
        module_id: 模块 ID
        user_data_context: 可选，描述你的数据形状，如「长表 gene/group/expr，3 组」
    """
    m = _by_id(module_id)
    need = next((s["content"] for s in m["sections"] if s.get("tag") == "requirement"), "")
    scen = next((s["content"] for s in m["sections"] if s.get("tag") == "scenario"), "")
    inp = [s["content"][:600] for s in m["sections"] if s.get("tag") == "input"]

    pkgs = m["packages"]
    checks = [
        {
            "title": "依赖安装",
            "items": [
                f"CRAN: {', '.join(pkgs['cran']) or '（脚本内自动检测）'}",
                f"Bioconductor: {', '.join(pkgs['bioc']) or '（无）'}",
                '先跑 source("install_dependencies.R")；该脚本用 install.packages + BiocManager 双通道',
                f"代码块共 {len(m['chunks'])} 个，其中 setup 块 {sum(1 for c in m['chunks'] if c['setup'])} 个（改 knitr 选项，非绘图逻辑）",
            ],
        },
        {
            "title": "输入数据",
            "items": (
                [f"模板对输入的要求：{x}" for x in inp]
                or ["该模块未单列「输入文件」章节，看代码块的 read.* 调用确认列名与格式"]
            )
            + [
                "数据文件不在本地 sparse 检出里（占上游 6.4 GB），需要时 "
                "`./01_clone.sh --full` 或从在线教程页下载 easy_input_*.csv"
            ],
        },
        {
            "title": "参数映射",
            "items": [
                "把 read.* 的路径换成你的文件；列名按 aes()/索引里的写法对齐",
                "先原样跑通再改：一次只改一个参数，避免样式与数据错误混在一起",
            ],
        },
        {
            "title": "常见失败",
            "items": [
                "列名不匹配 → 用 colnames()/str() 核对后再套 aes()",
                "分组变量被读成字符 → factor(x, levels=你的顺序)，否则图例与检验顺序错",
                "Bioconductor 包版本与 R 版本冲突 → 先 BiocManager::valid() 再装",
                "字体/中文乱码 → 出 PDF 前设 showtext 或改用 cairo_pdf",
            ],
        },
    ]
    if need:
        checks.insert(
            0,
            {
                "title": "这个模板要解决什么",
                "items": [need[:600]] + ([f"使用场景：{scen[:400]}"] if scen else []),
            },
        )
    if user_data_context:
        checks.append(
            {
                "title": "针对你的数据",
                "items": [
                    f"你的描述：{user_data_context}",
                    "对照上面的「输入数据」逐项确认格式是否满足",
                    f"若结构差异大，先用 figureya_search 找更接近的模块（同主题共 "
                    f"{len(m['sections'])} 个章节可参考）",
                ],
            }
        )
    return json.dumps(
        {"module": m["id"], "sections": checks, "site": m["site"]}, ensure_ascii=False, indent=2
    )


@mcp.tool()
def figureya_cite(module_id: str = "", format: str = "text") -> str:
    """引用信息与许可声明。

    Args:
        module_id: 可选，指定模块则附带该模块的在线地址
        format: text | bibtex
    """
    meta = get_bundle()["meta"]
    if format == "bibtex":
        body = (
            "@article{lu2025figureya,\n"
            "  title={FigureYa: A Standardized Visualization Framework for Enhancing "
            "Biomedical Data Interpretation and Research Efficiency},\n"
            "  author={Lu, Xiaofan and others},\n"
            "  journal={iMetaMed},\n"
            "  year={2025},\n"
            "  doi={10.1002/imm3.70005}\n"
            "}"
        )
    else:
        body = meta["citation"]
    out = {"citation": body, "doi": meta["doi"], "license": meta["license"]}
    if module_id:
        try:
            out["module"] = _by_id(module_id)["site"]
        except ValueError as e:
            out["module_error"] = str(e)
    out["license_note"] = "CC BY-NC-SA 4.0：非商用，转载/改编需署名并以相同许可共享"
    return json.dumps(out, ensure_ascii=False, indent=2)


# ---------------------------------------------------------------- 资源

@mcp.resource("figureya://catalog")
def catalog() -> str:
    """全部模块清单（id / 标题 / 依赖 / 在线链接）。"""
    return json.dumps(
        [
            {
                "id": m["id"],
                "title": m["title"],
                "packages": m["packages"]["all"][:8],
                "site": m["site"],
            }
            for m in _mods()
        ],
        ensure_ascii=False,
    )


@mcp.resource("figureya://module/{module_id}")
def module_resource(module_id: str) -> str:
    """单个模块的完整内容（含全部代码）。"""
    m = _by_id(module_id)
    return json.dumps(m, ensure_ascii=False)


# ---------------------------------------------------------------- 自检

def selftest() -> int:
    b = get_bundle()
    print(f"快照      : {SNAPSHOT} ({os.path.getsize(SNAPSHOT)/1024/1024:.1f} MB)")
    print(f"模块      : {len(b['modules'])}")
    print(f"主题分组  : {len(b['categories'])}")
    print(f"代码块    : {sum(len(m['chunks']) for m in b['modules'])}")

    cases = [("生存曲线", 5), ("circos", 5), ("单细胞", 5), ("volcano", 5)]
    print("\n检索自检:")
    for q, n in cases:
        r = json.loads(figureya_search(q, n))
        ids = [x["id"] for x in r.get("results", [])]
        print(f"  {q:10} -> {len(ids)} 条  {ids[:3]}")

    mid = b["modules"][0]["id"]
    print(f"\n取码自检: {mid}")
    code = json.loads(figureya_get_code(mid, chunk=0, include_deps=False))
    print(f"  chunk0 长度 {len(code['code_blocks'][0]['code'])} 字")
    mod = json.loads(figureya_get_module(mid, include_sections=False))
    print(f"  章节 {len(mod.get('chunks', []))} 个代码块, 依赖 {mod['packages']['all'][:3]}")
    return 0


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description="FigureYa MCP server")
    ap.add_argument("--selftest", action="store_true", help="自检后退出")
    a = ap.parse_args()
    if a.selftest:
        sys.exit(selftest())
    mcp.run()

#!/usr/bin/env bash
# ============================================================
# FigureYa MCP 一键安装 / 注册（幂等，可重复跑）
#
# 做的事：
#   1. 检查 Python 与 mcp SDK，缺则安装
#   2. 若 reference/ 与 snapshot/bundle.json 都不存在，先拉源码再打包
#   3. 注册进 DSH web profile 与 Codex CLI
#   4. 自检：跑一次 tools/list + 检索
#
# 用法:
#   ./06_install_mcp.sh            # 安装 + 注册全部
#   ./06_install_mcp.sh --dsh      # 只注册 DSH
#   ./06_install_mcp.sh --codex    # 只注册 Codex
#   ./06_install_mcp.sh --check    # 只体检，不安装不写入
#   ./06_install_mcp.sh --bundle   # 只重建内容快照
# ============================================================
set -euo pipefail
cd "$(dirname "$0")"
HERE="$(pwd)"

PY="${FIGUREYA_PYTHON:-~/miniforge3/bin/python3}"
[ -x "$PY" ] || PY="$(command -v python3)"
SERVER="$HERE/05_mcp_server.py"
SNAPSHOT="$HERE/snapshot/bundle.json"
DSH_PATCH="$HOME/.dsh/profiles/web/cordis.patch.yml"
CODEX_CFG="$HOME/.codex/config.toml"

ONLY_DSH=0; ONLY_CODEX=0; CHECK=0; BUNDLE_ONLY=0
for a in "$@"; do
  case "$a" in
    --dsh) ONLY_DSH=1 ;;
    --codex) ONLY_CODEX=1 ;;
    --check) CHECK=1 ;;
    --bundle) BUNDLE_ONLY=1 ;;
    -h|--help) sed -n '2,20p' "$0"; exit 0 ;;
    *) echo "未知参数: $a"; exit 1 ;;
  esac
done

say() { printf '\n=== %s ===\n' "$1"; }

# ---------- 0. 体检 ----------
if [ "$CHECK" -eq 1 ]; then
  say "体检"
  echo "Python      : $PY ($("$PY" -V 2>&1))"
  if "$PY" -c "import mcp" 2>/dev/null; then
    echo "mcp SDK     : ✅ $("$PY" -c 'import importlib.metadata as m; print(m.version("mcp"))')"
  else
    echo "mcp SDK     : ❌ 未安装"
  fi
  [ -f "$SERVER" ] && echo "server 脚本 : ✅ $SERVER" || echo "server 脚本 : ❌ 缺失"
  [ -f "$SNAPSHOT" ] && echo "内容快照    : ✅ $(du -h "$SNAPSHOT" | cut -f1)" || echo "内容快照    : ❌ 缺失（跑 --bundle）"
  [ -d reference ] && echo "reference/  : ✅ $(du -sh reference | cut -f1)" || echo "reference/  : ⚠️  未拉取（跑 ../01_clone.sh）"
  grep -q "mcp-figureya" "$DSH_PATCH" 2>/dev/null && echo "DSH 注册    : ✅" || echo "DSH 注册    : ❌"
  grep -q "mcp_servers.figureya" "$CODEX_CFG" 2>/dev/null && echo "Codex 注册  : ✅" || echo "Codex 注册  : ❌"
  exit 0
fi

# ---------- 1. Python 依赖 ----------
say "1/4 检查 Python 与 mcp SDK"
if ! "$PY" -c "import mcp" 2>/dev/null; then
  echo "安装 mcp SDK..."
  "$PY" -m pip install -q "mcp[cli]"
fi
echo "Python: $PY ($("$PY" -V 2>&1))"
echo "mcp   : $("$PY" -c 'import importlib.metadata as m;print(m.version("mcp"))')"

# ---------- 2. 内容快照 ----------
say "2/4 内容快照"
if [ "$BUNDLE_ONLY" -eq 1 ] || [ ! -f "$SNAPSHOT" ]; then
  if [ ! -d reference ]; then
    echo "缺少 reference/，先拉源码："
    ./01_clone.sh
  fi
  "$PY" 04_build_bundle.py
else
  echo "已存在 $(du -h "$SNAPSHOT" | cut -f1)，跳过（--bundle 可强制重建）"
fi

# ---------- 3. 注册 ----------
register_dsh() {
  say "3/4 注册 DSH"
  if [ ! -f "$DSH_PATCH" ]; then
    echo "⚠️  找不到 $DSH_PATCH，跳过"
    return
  fi
  if grep -q "mcp-figureya" "$DSH_PATCH"; then
    echo "已存在 mcp-figureya，跳过"
    return
  fi
  cp -a "$DSH_PATCH" "$DSH_PATCH.bak-$(date +%Y%m%d%H%M%S)"
  "$PY" - "$DSH_PATCH" "$PY" "$SERVER" <<'PYEOF'
import sys, pathlib
patch, pybin, server = pathlib.Path(sys.argv[1]), sys.argv[2], sys.argv[3]
s = patch.read_text(encoding="utf-8")
block = f"""
    - id: mcp-figureya
      name: "@deepseek-ai/dsh-mcp-client"
      config:
        serverName: figureya
        transport: stdio
        command: {pybin}
        args: [ "{server}" ]
"""
anchor = "    - id: mcp-tavily\n"
if anchor not in s:
    sys.exit("未找到插入锚点 mcp-tavily，请手动添加")
patch.write_text(s.replace(anchor, block.lstrip("\n") + "\n" + anchor, 1), encoding="utf-8")
print("已插入 mcp-figureya")
PYEOF
  echo "✅ 写入 $DSH_PATCH（已备份）"
  echo "   验证: dsh --profile web --dump-config | grep -A7 mcp-figureya"
}

register_codex() {
  say "4/4 注册 Codex"
  if [ ! -f "$CODEX_CFG" ]; then
    echo "⚠️  找不到 $CODEX_CFG，跳过"
    return
  fi
  if grep -q "\[mcp_servers.figureya\]" "$CODEX_CFG"; then
    echo "已存在 [mcp_servers.figureya]，跳过"
    return
  fi
  cp -a "$CODEX_CFG" "$CODEX_CFG.bak-$(date +%Y%m%d%H%M%S)"
  cat >> "$CODEX_CFG" <<EOF

[mcp_servers.figureya]
command = "$PY"
args = ["$SERVER"]
EOF
  echo "✅ 追加到 $CODEX_CFG（已备份）"
}

[ "$ONLY_CODEX" -eq 0 ] && register_dsh
[ "$ONLY_DSH" -eq 0 ] && register_codex

# ---------- 5. 自检 ----------
say "自检"
"$PY" 05_mcp_server.py --selftest 2>&1 | tail -12

say "完成"
echo "Agent 内直接用: mcp__figureya__figureya_search 等 6 个工具"
echo "重跑本脚本安全（幂等），--check 看状态"

#!/usr/bin/env Rscript
# Auto-generated R dependency installation script
# This script installs all required R packages for this project

# Set up mirrors for better download performance
options("repos" = c(CRAN = "https://cloud.r-project.org/"))
# Note: BiocManager::install handles its own mirrors, so the BioC_mirror option is often not needed.

# Function to check if a package is installed
is_package_installed <- function(package_name) {
  return(package_name %in% rownames(installed.packages()))
}

# Function to install CRAN packages
install_cran_package <- function(package_name) {
  if (!is_package_installed(package_name)) {
    cat("Installing CRAN package:", package_name, "\n")
    tryCatch({
      install.packages(package_name, dependencies = TRUE)
      cat("Successfully installed:", package_name, "\n")
    }, error = function(e) {
      cat("Failed to install", package_name, ":", e$message, "\n")
    })
  } else {
    cat("Package already installed:", package_name, "\n")
  }
}

# Function to install Bioconductor packages
install_bioc_package <- function(package_name) {
  if (!is_package_installed(package_name)) {
    cat("Installing Bioconductor package:", package_name, "\n")
    tryCatch({
      if (!is_package_installed("BiocManager")) {
        install.packages("BiocManager")
      }
      # Also install Biobase as it's a key dependency
      BiocManager::install(c("Biobase", package_name), update = FALSE, ask = FALSE)
      cat("Successfully installed:", package_name, "\n")
    }, error = function(e) {
      cat("Failed to install", package_name, ":", e$message, "\n")
    })
  } else {
    cat("Package already installed:", package_name, "\n")
  }
}

cat("Starting R package installation...\n")
cat("===========================================\n")


# Installing CRAN packages
cat("\nInstalling CRAN packages...\n")
cran_packages <- c("stringr") # "ClassDiscovery" was removed from here

for (pkg in cran_packages) {
  install_cran_package(pkg)
}

# Installing Bioconductor packages
cat("\nInstalling Bioconductor packages...\n")
bioc_packages <- c("ClassDiscovery") # "ClassDiscovery" is now correctly listed here

for (pkg in bioc_packages) {
  install_bioc_package(pkg)
}


cat("\n===========================================\n")
cat("Package installation completed!\n")
cat("You can now run your R scripts in this directory.\n")