BioScientist/agent_system/toolbase/prepare_converter_inputs.py

#!/usr/bin/env python3
"""
Prepare crawler JSON outputs for BioinfoMCP converter and optionally collect help docs.

Input JSON files:
- bioconda_t0_core_tools.json
- bioconda_t1_domain_tools.json
- bioconda_t2_on_demand_tools.json

Output files:
- converter_input_t0.json
- converter_input_t1.json
- converter_input_t2.json
- converter_input_all.json
- converter_jobs.json
- help_index.json
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import shutil
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Tuple
import requests
import time
from pathlib import Path
from typing import Dict, List, Tuple
from datetime import datetime


@dataclass
class ToolRow:
    source_file: str
    tier: str
    domain: str
    software_name: str
    package_name: str
    summary: str
    description: str
    dependencies: List[str]
    downloads: int
    home_url: str
    doc_url: str
    dev_url: str
    execution_environment: str
    execution_environment_reason: str


def read_rows(file_path: Path) -> List[dict]:
    if not file_path.exists():
        return []
    with file_path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, list):
        raise ValueError(f"JSON root must be list: {file_path}")
    return [x for x in data if isinstance(x, dict)]


def normalize_rows(file_path: Path, rows: List[dict]) -> List[ToolRow]:
    out: List[ToolRow] = []
    for row in rows:
        package_name = str(row.get("package_name", "")).strip()
        software_name = str(row.get("software_name", "")).strip() or package_name
        if not package_name and not software_name:
            continue
        deps = list(row.get("dependencies", []) or [])
        env = str(row.get("execution_environment", "")).strip()
        env_reason = str(row.get("execution_environment_reason", "")).strip()
        if not env:
            env, env_reason = infer_runtime_from_fields(package_name=package_name or software_name, dependencies=deps)
        out.append(
            ToolRow(
                source_file=file_path.name,
                tier=str(row.get("tier", "")).strip() or infer_tier(file_path.name),
                domain=str(row.get("domain", "")).strip(),
                software_name=software_name,
                package_name=package_name or software_name,
                summary=str(row.get("summary", "")).strip(),
                description=str(row.get("description", "")).strip(),
                dependencies=deps,
                downloads=safe_int(row.get("downloads", -1)),
                home_url=str(row.get("home_url", "")).strip(),
                doc_url=str(row.get("doc_url", "")).strip(),
                dev_url=str(row.get("dev_url", "")).strip(),
                execution_environment=env,
                execution_environment_reason=env_reason,
            )
        )
    return out


def infer_tier(filename: str) -> str:
    low = filename.lower()
    if "t0" in low:
        return "T0"
    if "t1" in low:
        return "T1"
    if "t2" in low:
        return "T2"
    return ""


def safe_int(v, default: int = -1) -> int:
    try:
        return int(v)
    except Exception:
        return default


def parse_bool(v, default: bool = False) -> bool:
    if isinstance(v, bool):
        return v
    if v is None:
        return default
    return str(v).strip().lower() in ("1", "true", "yes", "y", "on")


def infer_runtime_from_fields(package_name: str, dependencies: List[str]) -> Tuple[str, str]:
    pkg = package_name.lower()
    deps = [str(d).lower() for d in dependencies]

    if pkg.startswith("bioconductor-") or pkg.startswith("r-") or any("r-base" in d or d.startswith("r-") for d in deps):
        return "R", "inferred from package/dependencies (R ecosystem)"
    if pkg.startswith("perl-") or any(d == "perl" or d.startswith("perl-") for d in deps):
        return "Perl", "inferred from package/dependencies (Perl ecosystem)"
    if any("openjdk" in d or "default-jre" in d or d == "java" for d in deps):
        return "Java", "inferred from Java runtime dependencies"
    if any("python" in d for d in deps):
        return "Python", "inferred from python dependency"
    if any(k in " ".join(deps) for k in ("libgcc", "libstdcxx", "htslib")):
        return "Compiled", "inferred from native/compiled dependencies"
    return "Other", "fallback runtime classification"


def normalize_env_name(name: str) -> str:
    x = re.sub(r"[^A-Za-z0-9._-]+", "-", name.strip())
    x = re.sub(r"-{2,}", "-", x).strip("-")
    return x[:64] if len(x) > 64 else x


def route_env_name(base_env: str, row: ToolRow) -> str:
    pkg = row.package_name.lower()
    runtime = row.execution_environment.lower()
    dep_text = " ".join(str(d).lower() for d in row.dependencies)
    domain = (row.domain or "").lower()

    if pkg.startswith("bioconductor-") or pkg.startswith("r-") or runtime == "r":
        suffix = "r_bioc"
    elif any(k in dep_text for k in ("pytorch", "torch", "jax", "cuda", "scvi")):
        suffix = "py_torch"
    elif runtime == "python" and any(
        k in dep_text or k in pkg
        for k in ("scanpy", "anndata", "scvelo", "scarches", "squidpy", "scikit-learn")
    ):
        suffix = "py_sc"
    elif runtime == "perl":
        suffix = "perl"
    elif runtime == "java":
        suffix = "java"
    elif "single" in domain and runtime == "python":
        suffix = "py_sc"
    elif "spatial" in domain and runtime == "python":
        suffix = "py_spatial"
    else:
        suffix = "cli"
    return normalize_env_name(f"{base_env}_{suffix}")


def dedup_keep_best(rows: List[ToolRow]) -> List[ToolRow]:
    best: Dict[str, ToolRow] = {}
    for row in rows:
        key = row.package_name.lower()
        if key not in best:
            best[key] = row
            continue
        old = best[key]
        # Prefer higher download count.
        if row.downloads > old.downloads:
            best[key] = row
    return list(best.values())


def to_converter_row(row: ToolRow) -> dict:
    # This format is directly consumable by tool2mcp --tools_json,
    # because it keeps package_name/software_name/domain.
    return {
        "package_name": row.package_name,
        "software_name": row.software_name,
        "domain": row.domain,
        "tier": row.tier,
        "summary": row.summary,
        "description": row.description,
        "downloads": row.downloads, 
        "dependencies": row.dependencies,
        "home_url": row.home_url,
        "doc_url": row.doc_url,
        "dev_url": row.dev_url,
        "execution_environment": row.execution_environment,
        "execution_environment_reason": row.execution_environment_reason,
        "source_file": row.source_file,
    }


def write_json(path: Path, data) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def read_json_if_exists(path: Path, default):
    if not path.exists():
        return default
    try:
        with path.open("r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default


def _first_nonempty(paths: List[Path]) -> str:
    for p in paths:
        if p.exists() and p.stat().st_size > 0:
            return str(p)
    return ""


def existing_help_outputs(help_dir: Path, package_name: str) -> Dict[str, str]:
    """
    Check whether the tool already has generated help artifacts in output/help_docs.
    A tool is considered already processed if manual_bundle/help file exists and is non-empty.
    """
    manual_bundle = help_dir / f"{package_name}.manual_bundle.txt"
    manual_bundle_sub = help_dir / "manual_bundle_txt" / f"{package_name}.manual_bundle.txt"
    help_txt = help_dir / f"{package_name}.help.txt"
    help_txt_sub = help_dir / "help_txt" / f"{package_name}.help.txt"
    help_log = help_dir / f"{package_name}.help.log"
    install_log = help_dir / f"{package_name}.install.log"

    result = {
        "manual_bundle_file": _first_nonempty([manual_bundle_sub, manual_bundle]),
        "help_file": _first_nonempty([help_txt_sub, help_txt]),
        "help_log_file": str(help_log) if help_log.exists() and help_log.stat().st_size > 0 else "",
        "install_log_file": str(install_log) if install_log.exists() and install_log.stat().st_size > 0 else "",
    }
    return result


def executable_candidates(row: ToolRow) -> List[str]:
    pkg = row.package_name.strip()
    sw = row.software_name.strip()
    cands = [sw, pkg]
    # common name normalizations
    cands.append(pkg.replace("bioconductor-", ""))
    cands.append(sw.replace("_", "-"))
    cands.append(sw.replace("-", "_"))
    cands.append(pkg.replace("_", "-"))
    cands.append(pkg.replace("-", "_"))

    cleaned = []
    seen = set()
    for c in cands:
        c = c.strip()
        if not c:
            continue
        if not re.match(r"^[A-Za-z0-9._+-]+$", c):
            continue
        if c not in seen:
            seen.add(c)
            cleaned.append(c)
    return cleaned


def module_candidates(row: ToolRow) -> List[str]:
    pkg = row.package_name.strip().replace("-", "_")
    sw = row.software_name.strip().replace("-", "_")
    cands = [sw, pkg]
    cleaned = []
    seen = set()
    for c in cands:
        c = c.strip().strip(".")
        if not c:
            continue
        if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", c):
            continue
        if c not in seen:
            seen.add(c)
            cleaned.append(c)
    return cleaned


def r_package_candidates(row: ToolRow) -> List[str]:
    pkg = row.package_name.strip()
    sw = row.software_name.strip()
    candidates = []
    for raw in [pkg, sw]:
        if not raw:
            continue
        x = raw
        if x.startswith("bioconductor-"):
            x = x[len("bioconductor-") :]
        if x.startswith("r-"):
            x = x[len("r-") :]
        # R package names often use dot notation
        candidates.append(x.replace("-", "."))
        candidates.append(x.replace("-", ""))
    # keep order and unique
    seen = set()
    out = []
    for c in candidates:
        if c and c not in seen:
            seen.add(c)
            out.append(c)
    return out


def perl_module_candidates(row: ToolRow) -> List[str]:
    pkg = row.package_name.strip()
    sw = row.software_name.strip()
    candidates = []
    for raw in [pkg, sw]:
        x = raw
        if x.startswith("perl-"):
            x = x[len("perl-") :]
        x = x.replace("-", "::")
        candidates.append(x)
    seen = set()
    out = []
    for c in candidates:
        if c and c not in seen:
            seen.add(c)
            out.append(c)
    return out


def run_command(cmd: List[str], timeout: int = 1800) -> Tuple[int, str, str]:
    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        return proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired as exc:
        stdout = exc.stdout if isinstance(exc.stdout, str) else (exc.stdout.decode("utf-8", errors="ignore") if exc.stdout else "")
        stderr = exc.stderr if isinstance(exc.stderr, str) else (exc.stderr.decode("utf-8", errors="ignore") if exc.stderr else "")
        stderr = (stderr or "") + f"\n[TimeoutExpired] command exceeded {timeout}s"
        return 124, stdout or "", stderr


def resolve_solver(solver: str) -> str:
    if shutil.which(solver):
        return solver
    return "conda"


def conda_env_exists(conda_env: str) -> bool:
    rc, out, _err = run_command(["conda", "env", "list", "--json"], timeout=120)
    if rc != 0:
        return False
    try:
        payload = json.loads(out)
    except Exception:
        return False
    envs = payload.get("envs", []) or []
    marker = f"/envs/{conda_env}"
    return any(str(p).endswith(marker) or str(p).endswith(f"\\envs\\{conda_env}") for p in envs)


def ensure_conda_env(conda_env: str, python_version: str = "3.10") -> Tuple[bool, str]:
    if conda_env_exists(conda_env):
        return True, f"Conda env '{conda_env}' already exists."
    cmd = ["conda", "create", "-y", "-n", conda_env, f"python={python_version}"]
    rc, out, err = run_command(cmd, timeout=1800)
    ok = rc == 0
    msg = f"$ {' '.join(cmd)}\n[rc={rc}]\n{(out or '')[:6000]}\n{(err or '')[:6000]}"
    return ok, msg


def build_install_cmd(
    solver: str,
    conda_env: str,
    package_name: str,
    dry_run: bool = False,
    strict_channel_priority: bool = False,
) -> List[str]:
    cmd = [
        solver,
        "install",
        "-y",
        "-n",
        conda_env,
        "-c",
        "bioconda",
        "-c",
        "conda-forge",
        package_name,
    ]
    if dry_run:
        cmd.append("--dry-run")
    if strict_channel_priority:
        cmd.append("--strict-channel-priority")
    return cmd


def install_tool(
    conda_env: str,
    package_name: str,
    timeout: int = 1800,
    solver: str = "conda",
    strict_channel_priority: bool = False,
) -> Tuple[bool, str]:
    cmd = build_install_cmd(
        solver=solver,
        conda_env=conda_env,
        package_name=package_name,
        dry_run=False,
        strict_channel_priority=strict_channel_priority,
    )
    rc, out, err = run_command(cmd, timeout=timeout)
    ok = rc == 0
    msg = f"$ {' '.join(cmd)}\n[rc={rc}]\n{(out or '')[:4000]}\n{(err or '')[:4000]}"
    return ok, msg


def dry_run_install_tool(
    conda_env: str,
    package_name: str,
    timeout: int = 600,
    solver: str = "conda",
    strict_channel_priority: bool = False,
) -> Tuple[bool, str]:
    cmd = build_install_cmd(
        solver=solver,
        conda_env=conda_env,
        package_name=package_name,
        dry_run=True,
        strict_channel_priority=strict_channel_priority,
    )
    rc, out, err = run_command(cmd, timeout=timeout)
    ok = rc == 0
    msg = f"$ {' '.join(cmd)}\n[rc={rc}]\n{(out or '')[:5000]}\n{(err or '')[:5000]}"
    return ok, msg


def try_capture_help(conda_env: str, executable: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "run", "-n", conda_env, executable, "--help"]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    # Help sometimes exits non-zero, but still provides usage.
    success = ("usage" in lower or "help" in lower) and ("not found" not in lower)
    success = success or rc == 0
    log = f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"
    return success, log


def try_capture_help_module(conda_env: str, module_name: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "run", "-n", conda_env, "python", "-m", module_name, "--help"]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    success = ("usage" in lower or "help" in lower) and ("no module named" not in lower)
    success = success or rc == 0
    log = f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"
    return success, log


def try_capture_help_rscript(conda_env: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "run", "-n", conda_env, "Rscript", "--help"]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    ok = ("usage" in lower or "help" in lower) and ("not found" not in lower)
    ok = ok or rc == 0
    return ok, f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"


def try_capture_help_r_package(conda_env: str, pkg_name: str, timeout: int = 180) -> Tuple[bool, str]:
    expr = (
        f"if (requireNamespace('{pkg_name}', quietly=TRUE)) "
        f"{{library('{pkg_name}', character.only=TRUE); help(package='{pkg_name}')}} "
        f"else {{stop('package not installed: {pkg_name}')}}"
    )
    cmd = ["conda", "run", "-n", conda_env, "R", "-q", "-e", expr]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    ok = ("package:" in lower or "help pages" in lower or "index" in lower) and ("not installed" not in lower)
    ok = ok or rc == 0
    return ok, f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"


def try_capture_help_perl(conda_env: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "run", "-n", conda_env, "perl", "-h"]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    ok = ("usage" in lower or "perl" in lower) and ("not found" not in lower)
    ok = ok or rc == 0
    return ok, f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"


def try_capture_help_perldoc_module(conda_env: str, module_name: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "run", "-n", conda_env, "perldoc", module_name]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    ok = ("name" in lower or "description" in lower or "synopsis" in lower) and ("no documentation found" not in lower)
    ok = ok or rc == 0
    return ok, f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"


def try_capture_help_java(conda_env: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "run", "-n", conda_env, "java", "-help"]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    lower = text.lower()
    ok = ("usage" in lower or "java" in lower) and ("not found" not in lower)
    ok = ok or rc == 0
    return ok, f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"


def capture_help_by_runtime(conda_env: str, row: ToolRow, help_timeout: int = 120) -> Tuple[bool, str, str, List[str]]:
    runtime = row.execution_environment.lower()
    logs: List[str] = []

    # Python: CLI -> python -m
    if runtime == "python":
        for cand in executable_candidates(row):
            ok, log = try_capture_help(conda_env=conda_env, executable=cand, timeout=help_timeout)
            logs.append(log)
            if ok:
                return True, f"cli:{cand}", log, logs
        for mod in module_candidates(row):
            ok, log = try_capture_help_module(conda_env=conda_env, module_name=mod, timeout=help_timeout)
            logs.append(log)
            if ok:
                return True, f"module:{mod}", log, logs
        return False, "", "", logs

    # R: Rscript --help -> R package help -> generic CLI
    if runtime == "r":
        ok, log = try_capture_help_rscript(conda_env=conda_env, timeout=help_timeout)
        logs.append(log)
        if ok:
            return True, "rscript:--help", log, logs
        for rpkg in r_package_candidates(row):
            ok, log = try_capture_help_r_package(conda_env=conda_env, pkg_name=rpkg, timeout=max(help_timeout, 180))
            logs.append(log)
            if ok:
                return True, f"r_package:{rpkg}", log, logs
        for cand in executable_candidates(row):
            ok, log = try_capture_help(conda_env=conda_env, executable=cand, timeout=help_timeout)
            logs.append(log)
            if ok:
                return True, f"cli:{cand}", log, logs
        return False, "", "", logs

    # Perl: perldoc module -> perl -h -> generic CLI
    if runtime == "perl":
        for mod in perl_module_candidates(row):
            ok, log = try_capture_help_perldoc_module(conda_env=conda_env, module_name=mod, timeout=help_timeout)
            logs.append(log)
            if ok:
                return True, f"perldoc:{mod}", log, logs
        ok, log = try_capture_help_perl(conda_env=conda_env, timeout=help_timeout)
        logs.append(log)
        if ok:
            return True, "perl:-h", log, logs
        for cand in executable_candidates(row):
            ok, log = try_capture_help(conda_env=conda_env, executable=cand, timeout=help_timeout)
            logs.append(log)
            if ok:
                return True, f"cli:{cand}", log, logs
        return False, "", "", logs

    # Java: java -help -> generic CLI
    if runtime == "java":
        ok, log = try_capture_help_java(conda_env=conda_env, timeout=help_timeout)
        logs.append(log)
        if ok:
            return True, "java:-help", log, logs
        for cand in executable_candidates(row):
            ok, log = try_capture_help(conda_env=conda_env, executable=cand, timeout=help_timeout)
            logs.append(log)
            if ok:
                return True, f"cli:{cand}", log, logs
        return False, "", "", logs

    # Compiled/Other: generic CLI first, then python -m as last fallback.
    for cand in executable_candidates(row):
        ok, log = try_capture_help(conda_env=conda_env, executable=cand, timeout=help_timeout)
        logs.append(log)
        if ok:
            return True, f"cli:{cand}", log, logs
    for mod in module_candidates(row):
        ok, log = try_capture_help_module(conda_env=conda_env, module_name=mod, timeout=help_timeout)
        logs.append(log)
        if ok:
            return True, f"module:{mod}", log, logs
    return False, "", "", logs


def fetch_url_text(url: str, timeout: int = 20) -> Tuple[bool, str]:
    if not url:
        return False, ""
    try:
        resp = requests.get(url, timeout=timeout)
        if resp.status_code >= 400:
            return False, f"[HTTP {resp.status_code}] {url}"
        text = resp.text or ""
        text = re.sub(r"<script[\s\S]*?</script>", " ", text, flags=re.IGNORECASE)
        text = re.sub(r"<style[\s\S]*?</style>", " ", text, flags=re.IGNORECASE)
        text = re.sub(r"<[^>]+>", " ", text)
        text = re.sub(r"\s+", " ", text).strip()
        return True, text[:20000]
    except Exception as exc:
        return False, f"[ERROR] {url} -> {exc}"


def conda_search_info(package_name: str, timeout: int = 120) -> Tuple[bool, str]:
    cmd = ["conda", "search", "-c", "bioconda", "-c", "conda-forge", package_name, "--info"]
    rc, out, err = run_command(cmd, timeout=timeout)
    text = (out or "") + ("\n" + err if err else "")
    ok = rc == 0 and bool(text.strip())
    log = f"$ {' '.join(cmd)}\n[rc={rc}]\n{text[:12000]}"
    return ok, log


def build_manual_bundle(
    row: ToolRow,
    cli_help: str,
    cli_source: str,
    url_docs: List[Tuple[str, str]],
    conda_info: str,
) -> str:
    parts = [
        f"# Tool: {row.package_name}",
        f"software_name: {row.software_name}",
        f"tier: {row.tier}",
        f"domain: {row.domain}",
        f"downloads: {row.downloads}",
        f"summary: {row.summary}",
        f"description: {row.description}",
        f"dependencies: {', '.join(row.dependencies)}",
        f"execution_environment: {row.execution_environment}",
        f"execution_environment_reason: {row.execution_environment_reason}",
        "",
        "## URLs",
        f"home_url: {row.home_url}",
        f"doc_url: {row.doc_url}",
        f"dev_url: {row.dev_url}",
        "",
    ]
    if cli_help:
        parts += ["## CLI Help Source", cli_source or "unknown", "## CLI Help Content", cli_help, ""]
    if url_docs:
        parts += ["## URL Docs Extract"]
        for url, text in url_docs:
            parts += [f"### {url}", text, ""]
    if conda_info:
        parts += ["## Conda Search Info", conda_info, ""]
    return "\n".join(parts).strip() + "\n"


def collect_help_for_rows(
    rows: List[ToolRow],
    output_dir: Path,
    conda_env: str,
    do_install: bool,
    python_version: str = "3.10",
    install_timeout: int = 1800,
    dry_run_timeout: int = 600,
    help_timeout: int = 120,
    conda_info_timeout: int = 120,
    skip_processed: bool = True,
    use_env_routing: bool = True,
    enable_dry_run: bool = True,
    solver: str = "conda",
    strict_channel_priority: bool = False,
) -> Dict[str, dict]:
    """
    为一批工具收集帮助文档，生成 manual bundle。
    
    增强日志功能：
    - 分阶段进度输出
    - 每个工具的详细处理日志
    - 成功/失败统计
    - 耗时统计
    """
    start_time = time.time()
    help_dir = output_dir / "help_docs"
    help_dir.mkdir(parents=True, exist_ok=True)
    help_txt_dir = help_dir / "help_txt"
    manual_bundle_dir = help_dir / "manual_bundle_txt"
    help_txt_dir.mkdir(parents=True, exist_ok=True)
    manual_bundle_dir.mkdir(parents=True, exist_ok=True)
    
    # 日志文件：记录整个批处理的汇总信息
    batch_log_file = help_dir / f"batch_collect_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
    batch_log_lines = []
    
    def log_batch(message: str, also_print: bool = True):
        """记录批处理日志"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_line = f"[{timestamp}] {message}"
        batch_log_lines.append(log_line)
        if also_print:
            print(log_line)
    
    def log_to_file(file_path: Path, content: str):
        """写入文件并同时记录到批处理日志"""
        file_path.write_text(content, encoding="utf-8")
        log_batch(f"  -> 写入文件: {file_path.name} ({len(content)} bytes)", also_print=False)
    
    log_batch("=" * 80)
    log_batch(f"开始批量收集工具帮助文档")
    log_batch(f"  - 工具总数: {len(rows)}")
    log_batch(f"  - Conda 环境: {conda_env}")
    log_batch(f"  - 是否安装工具: {do_install}")
    log_batch(f"  - 输出目录: {output_dir}")
    log_batch(f"  - install_timeout: {install_timeout}s")
    log_batch(f"  - dry_run_timeout: {dry_run_timeout}s")
    log_batch(f"  - help_timeout: {help_timeout}s")
    log_batch(f"  - conda_info_timeout: {conda_info_timeout}s")
    log_batch(f"  - use_env_routing: {use_env_routing}")
    log_batch(f"  - enable_dry_run: {enable_dry_run}")
    log_batch(f"  - solver: {solver}")
    log_batch(f"  - strict_channel_priority: {strict_channel_priority}")
    log_batch("=" * 80)
    
    index_path = output_dir / "help_index.json"
    existing_index = read_json_if_exists(index_path, {})
    if not isinstance(existing_index, dict):
        existing_index = {}
    index: Dict[str, dict] = dict(existing_index)
    stats = {
        "total": len(rows),
        "installed": 0,
        "install_failed": 0,
        "help_success": 0,
        "help_failed": 0,
        "manual_bundle_generated": 0,
        "skipped_by_no_install": 0,
        "skipped_processed": 0,
        "dry_run_failed": 0,
        "docs_only_fallback": 0,
    }
    ensured_envs: Dict[str, bool] = {}
    
    for idx, row in enumerate(rows, 1):
        tool_start_time = time.time()
        log_batch(f"\n[{idx}/{len(rows)}] 处理工具: {row.package_name}")
        log_batch(f"  - Tier: {row.tier}, Domain: {row.domain}")
        
        key = row.package_name
        item = {
            "tool": key,
            "tier": row.tier,
            "domain": row.domain,
            "installed": False,
            "install_log_file": "",
            "help_ok": False,
            "help_file": "",
            "help_log_file": "",
            "executable_used": "",
            "manual_bundle_file": "",
            "manual_source": "",
            "reason": "",
            "processing_time_seconds": 0,
        }

        if skip_processed:
            existing = existing_index.get(key, {})
            existing_fs = existing_help_outputs(help_dir=help_dir, package_name=key)
            existing_bundle = existing_fs.get("manual_bundle_file", "")
            existing_help = existing_fs.get("help_file", "")
            bundle_ok = bool(existing_bundle)
            help_ok = bool(existing_help)
            if bundle_ok or help_ok:
                stats["skipped_processed"] += 1
                reason = "already processed (existing manual/help file found)"
                log_batch(f"  ⏭️  跳过: {reason}")
                existing["manual_bundle_file"] = existing_bundle or existing.get("manual_bundle_file", "")
                existing["help_file"] = existing_help or existing.get("help_file", "")
                existing["help_log_file"] = existing_fs.get("help_log_file", "") or existing.get("help_log_file", "")
                existing["install_log_file"] = existing_fs.get("install_log_file", "") or existing.get("install_log_file", "")
                existing["help_ok"] = True if (existing_bundle or existing_help) else bool(existing.get("help_ok", False))
                existing["reason"] = existing.get("reason") or reason
                existing["processing_time_seconds"] = existing.get("processing_time_seconds", 0)
                index[key] = existing
                continue
        
        target_env = route_env_name(conda_env, row) if use_env_routing else conda_env
        item["target_env"] = target_env
        logs: List[str] = []
        success = False
        chosen = ""
        chosen_source = ""
        chosen_help_text = ""
        capture_time = 0.0
        docs_only = False

        # ========== 阶段1: 安装工具（带可解性预检） ==========
        if do_install:
            if target_env not in ensured_envs:
                env_ok, env_msg = ensure_conda_env(target_env, python_version=python_version)
                ensured_envs[target_env] = env_ok
                pre_file = help_dir / f"{target_env}.env_preflight.log"
                log_to_file(pre_file, env_msg)
            if not ensured_envs.get(target_env, False):
                stats["install_failed"] += 1
                item["reason"] = f"conda env preflight failed: {target_env}"
                log_batch(f"  ❌ 环境不可用: {target_env}")
                docs_only = True
                stats["docs_only_fallback"] += 1
            else:
                if enable_dry_run:
                    log_batch(f"  [阶段1A] dry-run 预检到环境 '{target_env}' ...")
                    dry_ok, dry_log = dry_run_install_tool(
                        conda_env=target_env,
                        package_name=row.package_name,
                        timeout=dry_run_timeout,
                        solver=solver,
                        strict_channel_priority=strict_channel_priority,
                    )
                    dry_file = help_dir / f"{row.package_name}.dryrun.log"
                    log_to_file(dry_file, dry_log)
                    if not dry_ok:
                        stats["dry_run_failed"] += 1
                        stats["docs_only_fallback"] += 1
                        item["reason"] = "conda dry-run unsatisfiable/timeout -> docs_only fallback"
                        log_batch(f"  ⚠️ dry-run 失败，进入 docs_only 回退")
                        docs_only = True

                if not docs_only:
                    log_batch(f"  [阶段1B] 安装工具到环境 '{target_env}' ...")
                    install_start = time.time()
                    ok, install_log = install_tool(
                        conda_env=target_env,
                        package_name=row.package_name,
                        timeout=install_timeout,
                        solver=solver,
                        strict_channel_priority=strict_channel_priority,
                    )
                    install_time = time.time() - install_start

                    install_log_file = help_dir / f"{row.package_name}.install.log"
                    log_to_file(install_log_file, install_log)
                    item["install_log_file"] = str(install_log_file)
                    item["installed"] = ok

                    if ok:
                        stats["installed"] += 1
                        log_batch(f"  ✅ 安装成功 (耗时: {install_time:.2f}s)")
                    else:
                        stats["install_failed"] += 1
                        stats["docs_only_fallback"] += 1
                        item["reason"] = "conda install failed -> docs_only fallback"
                        log_batch(f"  ❌ 安装失败 (耗时: {install_time:.2f}s)，进入 docs_only 回退")
                        docs_only = True
        else:
            item["reason"] = "install skipped by --skip-install"
            stats["skipped_by_no_install"] += 1
            log_batch(f"  [阶段1] 跳过安装 (--skip-install)")
            docs_only = True
        
        # ========== 阶段2: 捕获 CLI 帮助 ==========
        if docs_only:
            logs.append(f"[docs_only] skip cli help capture for {row.package_name}")
            chosen_source = "docs_only"
            log_batch("  [阶段2] 跳过 CLI 捕获（docs_only 模式）")
        else:
            log_batch(f"  [阶段2] 捕获 CLI 帮助文档 ...")
            capture_start = time.time()
        
            try:
                success, chosen_source, chosen_help_text, capture_logs = capture_help_by_runtime(
                    conda_env=target_env,
                    row=row,
                    help_timeout=help_timeout,
                )
                logs.extend(capture_logs)
                capture_time = time.time() - capture_start
            except Exception as e:
                capture_time = time.time() - capture_start
                log_batch(f"  ⚠️ 捕获异常: {str(e)} (耗时: {capture_time:.2f}s)")
                success = False
                chosen_source = f"exception:{str(e)}"
                logs.append(f"Exception during capture: {str(e)}")
        
        if success:
            chosen = chosen_source.split(":", 1)[1] if ":" in chosen_source else chosen_source
            help_file = help_txt_dir / f"{row.package_name}.help.txt"
            log_to_file(help_file, chosen_help_text)
            item["help_file"] = str(help_file)
            item["help_ok"] = True
            stats["help_success"] += 1
            log_batch(f"  ✅ CLI 帮助捕获成功 (耗时: {capture_time:.2f}s)")
            log_batch(f"     - 可执行文件: {chosen}")
            log_batch(f"     - 帮助文本长度: {len(chosen_help_text)} 字符")
        else:
            if not docs_only:
                stats["help_failed"] += 1
                log_batch(f"  ⚠️ CLI 帮助捕获失败 (耗时: {capture_time:.2f}s)")
                log_batch(f"     - 来源: {chosen_source}")
        
        # ========== 阶段3: 收集外部文档 ==========
        log_batch(f"  [阶段3] 收集外部文档 (URL + conda info) ...")
        url_start = time.time()
        
        url_docs: List[Tuple[str, str]] = []
        url_success_count = 0
        for url in [row.doc_url, row.home_url, row.dev_url]:
            if not url:
                continue
            log_batch(f"     - 抓取 URL: {url[:80]}...", also_print=False)
            ok, text = fetch_url_text(url)
            logs.append(f"[url_fetch] {url}\n{(text or '')[:2000]}")
            if ok and text.strip():
                url_docs.append((url, text))
                url_success_count += 1
                log_batch(f"       ✅ 成功 (长度: {len(text)} 字符)", also_print=False)
            else:
                log_batch(f"       ❌ 失败", also_print=False)
        
        conda_ok, conda_info = conda_search_info(row.package_name, timeout=conda_info_timeout)
        logs.append(f"[conda_search_info]\n{conda_info}")
        conda_info_text = conda_info if conda_ok else ""
        log_batch(f"     - conda search --info: {'✅ 成功' if conda_ok else '❌ 失败'} (信息长度: {len(conda_info_text)} 字符)", also_print=False)
        
        url_time = time.time() - url_start
        log_batch(f"  [阶段3] 完成 (耗时: {url_time:.2f}s, 成功 URL: {url_success_count}/{len([u for u in [row.doc_url, row.home_url, row.dev_url] if u])})")
        
        # ========== 阶段4: 构建手册包 ==========
        log_batch(f"  [阶段4] 构建 manual bundle ...")
        bundle_start = time.time()
        
        manual_bundle = build_manual_bundle(
            row=row,
            cli_help=chosen_help_text,
            cli_source=chosen_source,
            url_docs=url_docs,
            conda_info=conda_info_text,
        )
        bundle_file = manual_bundle_dir / f"{row.package_name}.manual_bundle.txt"
        log_to_file(bundle_file, manual_bundle)
        item["manual_bundle_file"] = str(bundle_file)
        stats["manual_bundle_generated"] += 1
        
        if chosen_help_text:
            item["manual_source"] = chosen_source
            manual_source_desc = f"CLI help (via {chosen})"
        elif url_docs:
            item["manual_source"] = "url_docs"
            manual_source_desc = "URL docs"
        elif conda_info_text:
            item["manual_source"] = "conda_search_info"
            manual_source_desc = "conda search --info"
        else:
            item["manual_source"] = ""
            manual_source_desc = "无可用来源"
        
        bundle_time = time.time() - bundle_start
        log_batch(f"  [阶段4] 完成 (耗时: {bundle_time:.2f}s, 手册包大小: {len(manual_bundle)} 字符)")
        log_batch(f"     - 手册来源: {manual_source_desc}")
        
        # ========== 阶段5: 保存日志 ==========
        help_log_file = help_dir / f"{row.package_name}.help.log"
        full_log_content = "\n\n" + ("\n" + "=" * 80 + "\n\n").join(logs)
        log_to_file(help_log_file, full_log_content)
        item["help_log_file"] = str(help_log_file)
        
        if not success and not item["reason"]:
            item["reason"] = "cannot determine runnable executable for --help"
        
        tool_elapsed = time.time() - tool_start_time
        item["processing_time_seconds"] = round(tool_elapsed, 2)
        
        index[key] = item
        
        # 工具处理完成汇总
        status_icon = "✅" if item["help_ok"] else "⚠️"
        log_batch(f"  [完成] {status_icon} 工具 {row.package_name} 处理完成 (总耗时: {tool_elapsed:.2f}s)")
    
    # ========== 最终汇总 ==========
    total_elapsed = time.time() - start_time
    log_batch("\n" + "=" * 80)
    log_batch("批量收集完成 - 统计报告")
    log_batch("=" * 80)
    log_batch(f"  📊 总工具数:           {stats['total']}")
    log_batch(f"  📦 安装成功:           {stats['installed']}")
    log_batch(f"  ❌ 安装失败:           {stats['install_failed']}")
    log_batch(f"  ⏭️  跳过安装:           {stats['skipped_by_no_install']}")
    log_batch(f"  ⏩ 已处理跳过:         {stats['skipped_processed']}")
    log_batch(f"  🧪 dry-run 失败:        {stats['dry_run_failed']}")
    log_batch(f"  📚 docs_only 回退:      {stats['docs_only_fallback']}")
    log_batch(f"  📖 CLI 帮助成功:       {stats['help_success']}")
    log_batch(f"  ⚠️  CLI 帮助失败:       {stats['help_failed']}")
    log_batch(f"  📄 手册包生成:         {stats['manual_bundle_generated']}")
    log_batch(f"  ⏱️  总耗时:             {total_elapsed:.2f} 秒")
    if stats['total'] > 0:
        log_batch(f"  📈 平均每工具耗时:     {total_elapsed / stats['total']:.2f} 秒")
    log_batch(f"  📁 输出目录:           {output_dir}")
    log_batch("=" * 80)
    
    # 写入汇总日志文件
    log_to_file(batch_log_file, "\n".join(batch_log_lines))
    print(f"\n📋 批处理日志已保存至: {batch_log_file}")
    
    # 如果有失败的工具，列出它们以便排查
    failed_items = [
        (k, v)
        for k, v in index.items()
        if (not v.get("help_ok", False)) and (not v.get("manual_bundle_file", ""))
    ]
    if failed_items:
        print(f"\n⚠️ 以下 {len(failed_items)} 个工具的帮助文档收集失败:")
        for pkg_name, item in failed_items:
            reason = item.get("reason", "未知原因")
            print(f"   - {pkg_name}: {reason}")
    
    return index


def build_converter_jobs(rows: List[ToolRow], help_index: Dict[str, dict]) -> List[dict]:
    jobs = []
    for row in rows:
        h = help_index.get(row.package_name, {})
        bundle_file = h.get("manual_bundle_file", "")
        help_file = h.get("help_file", "")
        if bundle_file:
            manual = bundle_file
            run_help_command = False
        elif help_file:
            manual = help_file
            run_help_command = False
        else:
            manual = "--help"
            run_help_command = True
        jobs.append(
            {
                "name": row.package_name,
                "manual": manual,
                "run_help_command": run_help_command,
                "tier": row.tier,
                "domain": row.domain,
            }
        )
    return jobs


def main() -> None:
    parser = argparse.ArgumentParser(description="Prepare crawler outputs for BioinfoMCP converter.")
    parser.add_argument(
        "--input-dir",
        default="/225040511/project/BioScientist/agent_system/toolbase/output",
        help="Directory containing bioconda_t0/t1/t2 JSON files.",
    )
    parser.add_argument(
        "--output-dir",
        default="/225040511/project/BioScientist/agent_system/toolbase/output",
        help="Directory to write converter-ready files.",
    )
    parser.add_argument(
        "--conda-env",
        default="bioinfomcp-env",
        help="Base conda environment name. With --use-env-routing, derived envs are auto-created from this prefix.",
    )
    parser.add_argument(
        "--skip-install",
        action="store_true",
        help="Do not run conda install; only transform JSON and attempt help on existing env.",
    )
    parser.add_argument(
        "--max-tools",
        type=int,
        default=0,
        help="Limit number of tools for help collection (0 means all).",
    )
    parser.add_argument(
        "--python-version",
        default="3.10",
        help="Python version for auto-created conda env.",
    )
    parser.add_argument(
        "--install-timeout",
        type=int,
        default=1800,
        help="Timeout (seconds) for each conda install command.",
    )
    parser.add_argument(
        "--dry-run-timeout",
        type=int,
        default=600,
        help="Timeout (seconds) for conda install --dry-run precheck.",
    )
    parser.add_argument(
        "--help-timeout",
        type=int,
        default=120,
        help="Timeout (seconds) for each help capture command.",
    )
    parser.add_argument(
        "--conda-info-timeout",
        type=int,
        default=120,
        help="Timeout (seconds) for conda search --info.",
    )
    parser.add_argument(
        "--shard-total",
        type=int,
        default=1,
        help="Total shard count for multi-terminal execution.",
    )
    parser.add_argument(
        "--shard-index",
        type=int,
        default=0,
        help="Current shard index (0-based).",
    )
    parser.add_argument(
        "--skip-processed",
        type=str,
        default="True",
        help="Skip tools that already have existing manual/help outputs (True/False).",
    )
    parser.add_argument(
        "--use-env-routing",
        type=str,
        default="True",
        help="Auto route tools to category-specific conda envs (True/False).",
    )
    parser.add_argument(
        "--enable-dry-run",
        type=str,
        default="True",
        help="Run conda install --dry-run before real install (True/False).",
    )
    parser.add_argument(
        "--solver",
        type=str,
        default="conda",
        choices=["conda", "mamba"],
        help="Package solver executable for install steps.",
    )
    parser.add_argument(
        "--strict-channel-priority",
        type=str,
        default="True",
        help="Use --strict-channel-priority for install/dry-run (True/False).",
    )
    args = parser.parse_args()
    use_env_routing = parse_bool(args.use_env_routing, default=True)
    enable_dry_run = parse_bool(args.enable_dry_run, default=True)
    strict_channel_priority = parse_bool(args.strict_channel_priority, default=True)
    skip_processed = parse_bool(args.skip_processed, default=True)
    selected_solver = resolve_solver(args.solver)

    input_dir = Path(args.input_dir)
    output_dir = Path(args.output_dir)

    files = {
        "t0": input_dir / "bioconda_t0_core_tools.json",
        "t1": input_dir / "bioconda_t1_domain_tools.json",
        "t2": input_dir / "bioconda_t2_on_demand_tools.json",
    }

    normalized_by_tier: Dict[str, List[ToolRow]] = {}
    all_rows: List[ToolRow] = []
    for tier, file_path in files.items():
        rows = normalize_rows(file_path, read_rows(file_path))
        normalized_by_tier[tier] = rows
        all_rows.extend(rows)

    dedup_rows = dedup_keep_best(all_rows)
    if args.max_tools > 0:
        dedup_rows = dedup_rows[: args.max_tools]
    if args.shard_total < 1:
        raise ValueError("--shard-total must be >= 1")
    if args.shard_index < 0 or args.shard_index >= args.shard_total:
        raise ValueError("--shard-index must be in [0, shard_total)")
    if args.shard_total > 1:
        dedup_rows = [row for idx, row in enumerate(dedup_rows) if idx % args.shard_total == args.shard_index]
        print(f"Shard mode enabled: shard {args.shard_index}/{args.shard_total}, tools in this shard: {len(dedup_rows)}")

    # Write converter-friendly JSON per tier and merged.
    for tier, rows in normalized_by_tier.items():
        write_json(output_dir / f"converter_input_{tier}.json", [to_converter_row(r) for r in rows])
    write_json(output_dir / "converter_input_all.json", [to_converter_row(r) for r in dedup_rows])

    # Optional install + help capture
    preflight = {
        "conda_env": args.conda_env,
        "env_ready": True,
        "mode": "skip_install",
        "solver": selected_solver,
        "message": "skip install mode",
    }
    if not args.skip_install:
        if use_env_routing:
            preflight = {
                "conda_env": args.conda_env,
                "env_ready": True,
                "mode": "env_routing",
                "solver": selected_solver,
                "message": "routing mode enabled; per-category envs will be created lazily.",
            }
        else:
            env_ok, env_msg = ensure_conda_env(args.conda_env, python_version=args.python_version)
            preflight = {
                "conda_env": args.conda_env,
                "env_ready": env_ok,
                "mode": "single_env",
                "solver": selected_solver,
                "message": env_msg,
            }
            write_json(output_dir / "help_preflight.json", preflight)
            if not env_ok:
                # Environment failed to create; stop before per-tool install.
                write_json(output_dir / "help_index.json", {})
                write_json(output_dir / "converter_jobs.json", [])
                print("Conda environment preparation failed. See help_preflight.json")
                return
    write_json(output_dir / "help_preflight.json", preflight)

    help_index = collect_help_for_rows(
        rows=dedup_rows,
        output_dir=output_dir,
        conda_env=args.conda_env,
        do_install=not args.skip_install,
        python_version=args.python_version,
        install_timeout=args.install_timeout,
        dry_run_timeout=args.dry_run_timeout,
        help_timeout=args.help_timeout,
        conda_info_timeout=args.conda_info_timeout,
        skip_processed=skip_processed,
        use_env_routing=use_env_routing,
        enable_dry_run=enable_dry_run,
        solver=selected_solver,
        strict_channel_priority=strict_channel_priority,
    )
    write_json(output_dir / "help_index.json", help_index)

    # Build converter jobs with per-tool manual strategy.
    jobs = build_converter_jobs(rows=dedup_rows, help_index=help_index)
    write_json(output_dir / "converter_jobs.json", jobs)

    print(f"Prepared {len(dedup_rows)} tools.")
    print(f"- converter_input_all.json: {output_dir / 'converter_input_all.json'}")
    print(f"- converter_jobs.json: {output_dir / 'converter_jobs.json'}")
    print(f"- help_index.json: {output_dir / 'help_index.json'}")


if __name__ == "__main__":
    main()