app.py

# app.py
# ============================================================
# 类型：界面层（乙负责）
# 功能：Gradio 界面 + 研究方向全景研报格式化
# 用法：python app.py → 浏览器打开 http://127.0.0.1:7860
# 公网：设置 NGROK_AUTH_TOKEN 环境变量后自动生成公网链接
# ============================================================
import sys
import os
import gradio as gr

# 自动加载 .env 文件
_ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
if os.path.isfile(_ENV_PATH):
    with open(_ENV_PATH, "r", encoding="utf-8") as _f:
        for _line in _f:
            _line = _line.strip()
            if _line and not _line.startswith("#") and "=" in _line:
                _key, _val = _line.split("=", 1)
                if _key not in os.environ:
                    os.environ[_key] = _val.strip().strip('"').strip("'")

try:
    from run import run
except (ImportError, ModuleNotFoundError) as e:
    print(f"[错误] 无法导入 run.py，请确认甲已交付所有 Workflow 模块。")
    print(f"  {e}")
    sys.exit(1)


def diagnose(arxiv_url: str, progress: gr.Progress = gr.Progress()) -> str:
    """Gradio 回调：用户输入 arXiv URL → 返回 Markdown 研报"""
    url = arxiv_url.strip()
    if not url:
        return "## ⚠️ 请输入 arXiv 论文链接\n\n示例: `https://arxiv.org/abs/2011.08785`"

    try:
        result = run(url, progress=progress)
    except ValueError as e:
        return f"## ❌ 输入格式错误\n\n{str(e)}\n\n请确认输入的是有效的 arXiv 链接。"
    except RuntimeError as e:
        return f"## ❌ API 请求失败\n\n{str(e)}\n\n请检查网络连接或稍后重试。"
    except Exception as e:
        return f"## ❌ 运行出错\n\n```\n{type(e).__name__}: {str(e)}\n```\n\n请将以上错误信息反馈给开发者。"

    # 处理部分失败的情况
    if result.get("error") and not result.get("repos"):
        paper = result.get("paper", {})
        direction = result.get("direction", {})
        title = paper.get("title", "未知") if paper else "未知"

        md = f"""# ResearchRadar 研究方向全景研报

---

## 论文信息

**{title}**

---

## ⚠️ 未能完成完整评估

**原因**: {result['error']}

"""
        if direction:
            md += f"""## 研究方向解析（部分结果）

**子领域**: {direction.get('subfield', 'N/A')}

**趋势**: {direction.get('subfield_trend', 'N/A')}
"""
        md += """
---
> 该论文方向可能太新或太冷门，暂无高质量开源实现。这恰好帮助你判断——在这个方向开展研究需要从头实现。
"""
        return md

    return format_report(result)


def format_report(result: dict) -> str:
    """将 run() 的结果格式化为"研究方向全景研报" Markdown。

    研报结构：
    1. 论文信息
    2. 研究方向全景（子领域 + 趋势 + 方法族谱系 + 开源覆盖度）
    3. 方法族开源覆盖度（哪些已有实现、哪些是空白）
    4. 开源实现评估排名（六维度评分明细）
    5. 对比实验推荐总结（按就绪度分组）
    """
    paper = result.get("paper", {})
    direction = result.get("direction", {})
    repos = result.get("repos", [])

    # ===== 第一部分：论文信息 =====
    arxiv_id = paper.get("arxiv_id", "")
    arxiv_url = f"https://arxiv.org/abs/{arxiv_id}" if arxiv_id else ""

    md = "# ResearchRadar 研究方向全景研报\n\n---\n\n## 一、论文信息\n\n"
    md += "| 项目 | 内容 |\n|------|------|\n"
    md += f"| **标题** | {paper.get('title', 'N/A')} |\n"
    if arxiv_url:
        md += f"| **arXiv** | [{arxiv_id}]({arxiv_url}) |\n"
    else:
        md += f"| **arXiv ID** | {arxiv_id or 'N/A'} |\n"
    md += f"| **作者** | {', '.join(paper.get('authors', ['N/A'])[:5])}"
    if len(paper.get("authors", [])) > 5:
        md += f" 等 {len(paper.get('authors', []))} 人"
    md += " |\n"
    published = paper.get("published", "")
    if published:
        md += f"| **发表时间** | {published[:10]} |\n"
    categories = paper.get("categories", [])
    if categories:
        md += f"| **分类** | {', '.join(categories)} |\n"

    # ===== 论文自身代码评估 =====
    own_repo = None
    other_repos = []
    for r in repos:
        if r.get("method_family") == "本文代码":
            own_repo = r
        else:
            other_repos.append(r)

    if own_repo is not None:
        ev = own_repo.get("evaluation", {})
        md += "\n---\n\n## 二、本文代码复现评估\n\n"
        md += "> 该论文是否提供了官方代码？代码质量如何？能否直接复现论文实验结果？\n\n"
        md += f"| 仓库 | Stars | 综合评分 | 可复现性 | 对比实验适配度 |\n"
        md += f"|------|:----:|:----:|:----:|:----:|\n"
        verdict_icon = {"reproducible": "✅", "partially": "⚠️", "not_reproducible": "❌", "error": "💥"}.get(ev.get("verdict", ""), "❓")
        readiness_icon = {"ready": "🟢", "partial": "🟡", "not_ready": "🔴"}.get(ev.get("benchmark_readiness", ""), "⚪")
        md += f"| [{own_repo['full_name']}]({own_repo.get('html_url', '#')}) | {own_repo.get('stars', 0)} | **{ev.get('overall_score', 'N/A')}/100** | {verdict_icon} | {readiness_icon} |\n\n"
        md += f"### 📋 详细评估\n\n{ev.get('reasoning', 'N/A')}\n\n"

        risks = ev.get("risks", [])
        if risks:
            md += "### ⚠️ 风险点\n\n"
            for r_ in risks:
                md += f"- {r_}\n"
            md += "\n"
        md += f"### 💡 使用建议\n\n{ev.get('suggested_use', 'N/A')}\n\n"
    else:
        md += "\n---\n\n## 二、本文代码复现评估\n\n"
        md += "> 🔴 **未找到该论文的官方开源代码。**\n>\n"
        md += "> 这意味着复现该论文的实验结果需要从头实现，难度较高。建议在对比实验中使用下面列出的同一方法族的开源实现作为替代。\n\n"

    # ===== 第三部分：研究方向全景 =====
    md += "\n---\n\n## 三、研究方向全景\n\n"
    md += f"### 🎯 子领域定位\n\n**{direction.get('subfield', '未知')}**\n\n"
    
    trend = direction.get('subfield_trend', '暂无分析')
    md += f"### 📈 技术趋势与前沿动态\n\n{trend}\n\n"
    
    md += "### 🏗️ 方法族谱系\n\n"
    md += "> 以下方法族基于 GitHub 上该领域实际开源仓库的分析归纳，反映了当前学术界的组织结构和开源生态。\n\n"

    families = direction.get("method_families", [])
    for i, mf in enumerate(families):
        family_name = mf.get("family_name", f"方法族 {i+1}")
        desc = mf.get("description", "暂无详细描述")
        rep = mf.get("representative_work", "暂无代表工作")
        matched = mf.get("matched_repos", [])
        queries = mf.get("search_queries", [])
        
        md += f"#### {i+1}. {family_name}\n\n"
        md += f"**核心特点**: {desc}\n\n"
        md += f"**代表工作**: {rep}\n\n"
        if matched:
            md += f"**开源覆盖**: 🟢 {len(matched)} 个仓库 — {', '.join(matched)}\n\n"
        else:
            md += "**开源覆盖**: 🔴 暂无开源实现（可能是研究空白）\n\n"
        if queries:
            md += f"**精准搜索词**: {', '.join(queries)}\n\n"

    if not families:
        md += "| - | 未能识别出明确的方法族 | - | - |\n"

    md += "\n"

    # ===== 第四部分：仓库评估排名 =====
    md += "\n---\n\n## 四、开源实现评估排名\n\n"
    md += "> 评分逻辑：可复现性（80分）评估代码能否被他人成功跑通；对比实验适配度（20分）评估该仓库能否直接用于论文 Experiment section。\n\n"

    if not other_repos and not own_repo:
        md += "**未找到相关开源仓库。**\n\n"
        return md

    if not other_repos:
        md += "**未找到其他开源仓库（仅找到论文自身代码）。**\n\n"

    for i, repo in enumerate(other_repos):
        full_name = repo.get("full_name", f"仓库 {i+1}")
        html_url = repo.get("html_url", "")
        ev = repo.get("evaluation", {})
        mf = repo.get("method_family", "")

        verdict_icon = {
            "reproducible": "✅", "partially": "⚠️",
            "not_reproducible": "❌", "error": "💥",
        }.get(ev.get("verdict", ""), "❓")

        readiness_icon = {
            "ready": "🟢", "partial": "🟡", "not_ready": "🔴",
        }.get(ev.get("benchmark_readiness", ""), "⚪")

        family_tag = f" `[{mf}]`" if mf else ""

        md += f"### {i+1}. [{full_name}]({html_url}){family_tag}\n\n"

        repro_score = ev.get("reproducibility_score", ev.get("overall_score", "N/A"))
        bench_score = ev.get("benchmark_fitness_score", ev.get("benchmark_score", "N/A"))
        overall = ev.get("overall_score", "N/A")

        # 评分概要
        md += f"| 指标 | 评分 |\n"
        md += f"|------|:----:|\n"
        md += f"| {verdict_icon} 可复现性 | **{repro_score}/80** |\n"
        md += f"| {readiness_icon} 对比实验适配度 | **{bench_score}/20** |\n"
        md += f"| 📊 综合评分 | **{overall}/100** |\n\n"

        # 详细分析（核心内容，放在最显眼位置）
        reasoning = ev.get("reasoning", "N/A")
        md += f"### 📋 详细分析\n\n{reasoning}\n\n"

        # 六维度评分表
        md += "<details>\n<summary>📊 六维度评分明细（点击展开）</summary>\n\n"
        md += "| 维度 | 得分 | 满分 | 评估标准 |\n"
        md += "|------|:----:|:----:|----------|\n"
        md += f"| 环境配置 | {_fmt(ev.get('env_score'))} | 15 | 依赖文件完整性 |\n"
        md += f"| 文档质量 | {_fmt(ev.get('doc_score'))} | 20 | 安装/训练/数据说明 |\n"
        md += f"| 代码可用性 | {_fmt(ev.get('code_score'))} | 20 | 训练+推理脚本 |\n"
        md += f"| 社区活跃度 | {_fmt(ev.get('community_score'))} | 10 | Star + 更新频率 |\n"
        md += f"| 依赖健康度 | {_fmt(ev.get('dep_score'))} | 15 | 依赖兼容性 |\n"
        md += f"| **对比实验** | **{_fmt(ev.get('benchmark_score'))}** | **20** | **benchmark + 预训练权重** |\n"
        md += "\n</details>\n\n"

        # 对比实验建议
        md += f"### 💡 对比实验建议\n\n{ev.get('suggested_use', 'N/A')}\n\n"

        # 风险点
        risks = ev.get("risks", [])
        if risks:
            md += "### ⚠️ 风险点\n\n"
            for r in risks:
                md += f"- {r}\n"
            md += "\n"

        md += f"⭐ Star: {repo.get('stars', 0)}"
        if repo.get("match_keyword"):
            md += f" | 匹配词: `{repo.get('match_keyword', '')}`"
        md += "\n\n---\n\n"

    # ===== 第五部分：对比实验推荐总结 =====
    md += "## 五、对比实验推荐总结\n\n"

    all_evaluated = other_repos + ([own_repo] if own_repo else [])
    ready = [r for r in all_evaluated if r["evaluation"].get("benchmark_readiness") == "ready"]
    partial = [r for r in all_evaluated if r["evaluation"].get("benchmark_readiness") == "partial"]
    not_ready = [
        r for r in all_evaluated
        if r["evaluation"].get("benchmark_readiness") not in ("ready", "partial")
    ]

    if ready:
        md += "### 🟢 可直接用于对比实验\n\n"
        for r in ready:
            url = r.get("html_url", "#")
            name = r.get("full_name", "未知")
            suggestion = r["evaluation"].get("suggested_use", "")
            md += f"- **[{name}]({url})** — {suggestion}\n"
        md += "\n"

    if partial:
        md += "### 🟡 需要少量修改后可用\n\n"
        for r in partial:
            url = r.get("html_url", "#")
            name = r.get("full_name", "未知")
            suggestion = r["evaluation"].get("suggested_use", "")
            md += f"- **[{name}]({url})** — {suggestion}\n"
        md += "\n"

    if not_ready:
        md += "### 🔴 仅适合参考代码实现\n\n"
        for r in not_ready:
            url = r.get("html_url", "#")
            name = r.get("full_name", "未知")
            reason = r["evaluation"].get("suggested_use") or r["evaluation"].get("reasoning", "")
            md += f"- **[{name}]({url})** — {reason}\n"
        md += "\n"

    # ===== 审核质量报告 =====
    audit = result.get("audit", {})
    if audit:
        md += "\n---\n\n## 六、质量审核报告\n\n"
        md += f"**综合质量评分**: {audit.get('overall_score', 'N/A')}/100\n\n"
        md += f"{audit.get('summary', '')}\n\n"

        if audit.get("actions"):
            md += "### 改进建议\n\n"
            for action in audit["actions"]:
                md += f"- {action}\n"
            md += "\n"

        src = audit.get("source_audit", {})
        if src.get("unreliable_repos"):
            md += "### ⚠️ 来源可靠性提醒\n\n"
            md += "以下仓库来源可靠性较低，评估结果仅供参考：\n\n"
            for r in src["unreliable_repos"]:
                md += f"- `{r}`\n"
            md += "\n"

    md += "---\n\n> ResearchRadar · 论文研究方向全景 + 开源代码评估 + 对比实验推荐\n"

    return md


def _fmt(value) -> str:
    """格式化数值，None 显示为 N/A"""
    if value is None:
        return "N/A"
    return str(value)


# ============================================================
# Gradio 启动
# ============================================================
DEMO_EXAMPLES = [
    ["https://arxiv.org/abs/2011.08785"],   # PaDiM — 工业缺陷检测（制造工程）
    ["https://arxiv.org/abs/2301.06230"],   # Swarm-SLAM — 多机器人协同SLAM（机器人工程）
    ["https://arxiv.org/abs/2304.07794"],   # NDP-NMPC — 四旋翼非线性控制（航空航天）
    ["https://arxiv.org/abs/2306.05889"],   # C(NN)FD — 涡轮机械CFD（机械工程）
]

with gr.Blocks(title="ResearchRadar") as demo:
    gr.HTML("""
    <div style="text-align: center; margin-bottom: 20px;">
        <h1>🔬 ResearchRadar</h1>
        <h3>研究方向全景分析 + 开源代码评估 + 对比实验推荐</h3>
        <p style="color: #666; max-width: 700px; margin: 0 auto;">
            输入一篇工科论文的 arXiv 链接，系统自动完成：<br>
            ① <strong>摸清方向</strong> — 该领域有哪些方法族？各自什么特点？当前趋势是什么？<br>
            ② <strong>找到代码</strong> — GitHub 上哪些开源仓库实现了这些方法？覆盖度如何？<br>
            ③ <strong>评估可用性</strong> — 每个仓库能不能跑通？能不能直接用于论文对比实验？
        </p>
    </div>
    """)

    with gr.Row():
        with gr.Column(scale=4):
            url_input = gr.Textbox(
                label="📄 论文 arXiv 链接",
                placeholder="https://arxiv.org/abs/2011.08785",
                lines=1,
                show_label=True,
            )
        with gr.Column(scale=1):
            submit_btn = gr.Button("🔍 开始分析", variant="primary", size="lg")

    with gr.Accordion("📚 示例论文（点击展开，选择后自动填入）", open=True):
        gr.Examples(
            examples=DEMO_EXAMPLES,
            inputs=url_input,
            examples_per_page=4,
        )

    with gr.Row():
        report_output = gr.Markdown(
            label="研究方向全景研报",
            value="""
---
## 等待输入

请输入一篇工科论文的 **arXiv 链接**（如 `https://arxiv.org/abs/2011.08785`），点击「开始分析」。

系统将在 3-5 分钟内生成一份完整的 **研究方向全景研报**，包含：

| 章节 | 内容 |
|------|------|
| **一、论文信息** | 标题、作者、分类、发表时间 |
| **二、研究方向全景** | 子领域定位、技术趋势、方法族谱系（每个方法族含核心特点/代表工作/开源覆盖度/精准搜索词） |
| **三、开源实现评估排名** | 每个仓库的详细分析 + 六维度评分 + 风险点 |
| **四、对比实验推荐总结** | 🟢可直接使用 / 🟡需少量修改 / 🔴仅适合参考 |

---
            """,
        )

    submit_btn.click(fn=diagnose, inputs=url_input, outputs=report_output)
    url_input.submit(fn=diagnose, inputs=url_input, outputs=report_output)

    demo.queue(default_concurrency_limit=2, max_size=20, status_update_rate=2.0)

if __name__ == "__main__":
    public_url = None

    # 尝试 ngrok 公网隧道（国内可用）
    ngrok_token = os.getenv("NGROK_AUTH_TOKEN", "")
    if ngrok_token:
        try:
            from pyngrok import ngrok, conf
            conf.get_default().auth_token = ngrok_token
            tunnel = ngrok.connect(7860, "http")
            public_url = tunnel.public_url
            print(f"\n{'='*60}")
            print(f"  公网链接: {public_url}")
            print(f"  任何人可以打开此链接使用 ResearchRadar")
            print(f"{'='*60}\n")
        except Exception as e:
            print(f"  [WARN] ngrok 启动失败: {e}")
            print(f"  降级为仅本地访问")

    if not public_url:
        print("\n  未配置公网隧道，仅本地访问: http://127.0.0.1:7860")
        print("  如需公网访问，请设置 NGROK_AUTH_TOKEN 环境变量")
        print("  免费获取 token: https://dashboard.ngrok.com/get-started/your-authtoken\n")

    demo.launch(server_name="0.0.0.0", share=False, theme="soft", show_error=True)