镜像自地址
https://github.com/binary-husky/gpt_academic.git
已同步 2025-12-06 06:26:47 +00:00
360 行
15 KiB
Python
360 行
15 KiB
Python
import os
|
||
import time
|
||
import glob
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from dataclasses import dataclass
|
||
from typing import Dict, List, Generator, Tuple
|
||
from crazy_functions.crazy_utils import request_gpt_model_in_new_thread_with_ui_alive
|
||
from toolbox import update_ui, promote_file_to_downloadzone, write_history_to_file, CatchException, report_exception
|
||
from shared_utils.fastapi_server import validate_path_safety
|
||
from crazy_functions.paper_fns.paper_download import extract_paper_id, extract_paper_ids, get_arxiv_paper, format_arxiv_id
|
||
|
||
|
||
|
||
@dataclass
|
||
class PaperQuestion:
|
||
"""论文分析问题类"""
|
||
id: str # 问题ID
|
||
question: str # 问题内容
|
||
importance: int # 重要性 (1-5，5最高)
|
||
description: str # 问题描述
|
||
|
||
|
||
class PaperAnalyzer:
|
||
"""论文快速分析器"""
|
||
|
||
def __init__(self, llm_kwargs: Dict, plugin_kwargs: Dict, chatbot: List, history: List, system_prompt: str):
|
||
"""初始化分析器"""
|
||
self.llm_kwargs = llm_kwargs
|
||
self.plugin_kwargs = plugin_kwargs
|
||
self.chatbot = chatbot
|
||
self.history = history
|
||
self.system_prompt = system_prompt
|
||
self.paper_content = ""
|
||
self.results = {}
|
||
|
||
# 定义论文分析问题库（已合并为4个核心问题）
|
||
self.questions = [
|
||
PaperQuestion(
|
||
id="research_and_methods",
|
||
question="这篇论文的主要研究问题、目标和方法是什么？请分析：1)论文的核心研究问题和研究动机；2)论文提出的关键方法、模型或理论框架；3)这些方法如何解决研究问题。",
|
||
importance=5,
|
||
description="研究问题与方法"
|
||
),
|
||
PaperQuestion(
|
||
id="findings_and_innovation",
|
||
question="论文的主要发现、结论及创新点是什么？请分析：1)论文的核心结果与主要发现；2)作者得出的关键结论；3)研究的创新点与对领域的贡献；4)与已有工作的区别。",
|
||
importance=4,
|
||
description="研究发现与创新"
|
||
),
|
||
PaperQuestion(
|
||
id="methodology_and_data",
|
||
question="论文使用了什么研究方法和数据？请详细分析：1)研究设计与实验设置；2)数据收集方法与数据集特点；3)分析技术与评估方法；4)方法学上的合理性。",
|
||
importance=3,
|
||
description="研究方法与数据"
|
||
),
|
||
PaperQuestion(
|
||
id="limitations_and_impact",
|
||
question="论文的局限性、未来方向及潜在影响是什么？请分析：1)研究的不足与限制因素；2)作者提出的未来研究方向；3)该研究对学术界和行业可能产生的影响；4)研究结果的适用范围与推广价值。",
|
||
importance=2,
|
||
description="局限性与影响"
|
||
),
|
||
]
|
||
|
||
# 按重要性排序
|
||
self.questions.sort(key=lambda q: q.importance, reverse=True)
|
||
|
||
def _load_paper(self, paper_path: str) -> Generator:
|
||
from crazy_functions.doc_fns.text_content_loader import TextContentLoader
|
||
"""加载论文内容"""
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
|
||
# 使用TextContentLoader读取文件
|
||
loader = TextContentLoader(self.chatbot, self.history)
|
||
|
||
yield from loader.execute_single_file(paper_path)
|
||
|
||
# 获取加载的内容
|
||
if len(self.history) >= 2 and self.history[-2]:
|
||
self.paper_content = self.history[-2]
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
return True
|
||
else:
|
||
self.chatbot.append(["错误", "无法读取论文内容，请检查文件是否有效"])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
return False
|
||
|
||
def _analyze_question(self, question: PaperQuestion) -> Generator:
|
||
"""分析单个问题 - 直接显示问题和答案"""
|
||
try:
|
||
# 创建分析提示
|
||
prompt = f"请基于以下论文内容回答问题：\n\n{self.paper_content}\n\n问题：{question.question}"
|
||
|
||
# 使用单线程版本的请求函数
|
||
response = yield from request_gpt_model_in_new_thread_with_ui_alive(
|
||
inputs=prompt,
|
||
inputs_show_user=question.question, # 显示问题本身
|
||
llm_kwargs=self.llm_kwargs,
|
||
chatbot=self.chatbot,
|
||
history=[], # 空历史，确保每个问题独立分析
|
||
sys_prompt="你是一个专业的科研论文分析助手，需要仔细阅读论文内容并回答问题。请保持客观、准确，并基于论文内容提供深入分析。"
|
||
)
|
||
|
||
if response:
|
||
self.results[question.id] = response
|
||
return True
|
||
return False
|
||
|
||
except Exception as e:
|
||
self.chatbot.append(["错误", f"分析问题时出错: {str(e)}"])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
return False
|
||
|
||
def _generate_summary(self) -> Generator:
|
||
"""生成最终总结报告"""
|
||
self.chatbot.append(["生成报告", "正在整合分析结果，生成最终报告..."])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
|
||
summary_prompt = "请基于以下对论文的各个方面的分析，生成一份全面的论文解读报告。报告应该简明扼要地呈现论文的关键内容，并保持逻辑连贯性。"
|
||
|
||
for q in self.questions:
|
||
if q.id in self.results:
|
||
summary_prompt += f"\n\n关于{q.description}的分析:\n{self.results[q.id]}"
|
||
|
||
try:
|
||
# 使用单线程版本的请求函数，可以在前端实时显示生成结果
|
||
response = yield from request_gpt_model_in_new_thread_with_ui_alive(
|
||
inputs=summary_prompt,
|
||
inputs_show_user="生成论文解读报告",
|
||
llm_kwargs=self.llm_kwargs,
|
||
chatbot=self.chatbot,
|
||
history=[],
|
||
sys_prompt="你是一个科研论文解读专家，请将多个方面的分析整合为一份完整、连贯、有条理的报告。报告应当重点突出，层次分明，并且保持学术性和客观性。"
|
||
)
|
||
|
||
if response:
|
||
return response
|
||
return "报告生成失败"
|
||
|
||
except Exception as e:
|
||
self.chatbot.append(["错误", f"生成报告时出错: {str(e)}"])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
return "报告生成失败: " + str(e)
|
||
|
||
def save_report(self, report: str) -> Generator:
|
||
"""保存分析报告"""
|
||
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
||
|
||
# 保存为Markdown文件
|
||
try:
|
||
md_content = f"# 论文快速解读报告\n\n{report}"
|
||
for q in self.questions:
|
||
if q.id in self.results:
|
||
md_content += f"\n\n## {q.description}\n\n{self.results[q.id]}"
|
||
|
||
result_file = write_history_to_file(
|
||
history=[md_content],
|
||
file_basename=f"论文解读_{timestamp}.md"
|
||
)
|
||
|
||
if result_file and os.path.exists(result_file):
|
||
promote_file_to_downloadzone(result_file, chatbot=self.chatbot)
|
||
self.chatbot.append(["保存成功", f"解读报告已保存至: {os.path.basename(result_file)}"])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
else:
|
||
self.chatbot.append(["警告", "保存报告成功但找不到文件"])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
except Exception as e:
|
||
self.chatbot.append(["警告", f"保存报告失败: {str(e)}"])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
|
||
def analyze_paper(self, paper_path: str) -> Generator:
|
||
"""分析论文主流程"""
|
||
# 加载论文
|
||
success = yield from self._load_paper(paper_path)
|
||
if not success:
|
||
return
|
||
|
||
# 分析关键问题 - 直接询问每个问题，不显示进度信息
|
||
for question in self.questions:
|
||
yield from self._analyze_question(question)
|
||
|
||
# 生成总结报告
|
||
final_report = yield from self._generate_summary()
|
||
|
||
# 显示最终报告
|
||
# self.chatbot.append(["论文解读报告", final_report])
|
||
yield from update_ui(chatbot=self.chatbot, history=self.history)
|
||
|
||
# 保存报告
|
||
yield from self.save_report(final_report)
|
||
|
||
|
||
def _find_paper_file(path: str) -> str:
|
||
"""查找路径中的论文文件（简化版）"""
|
||
if os.path.isfile(path):
|
||
return path
|
||
|
||
# 支持的文件扩展名（按优先级排序）
|
||
extensions = ["pdf", "docx", "doc", "txt", "md", "tex"]
|
||
|
||
# 简单地遍历目录
|
||
if os.path.isdir(path):
|
||
try:
|
||
for ext in extensions:
|
||
# 手动检查每个可能的文件，而不使用glob
|
||
potential_file = os.path.join(path, f"paper.{ext}")
|
||
if os.path.exists(potential_file) and os.path.isfile(potential_file):
|
||
return potential_file
|
||
|
||
# 如果没找到特定命名的文件，检查目录中的所有文件
|
||
for file in os.listdir(path):
|
||
file_path = os.path.join(path, file)
|
||
if os.path.isfile(file_path):
|
||
file_ext = file.split('.')[-1].lower() if '.' in file else ""
|
||
if file_ext in extensions:
|
||
return file_path
|
||
except Exception:
|
||
pass # 忽略任何错误
|
||
|
||
return None
|
||
|
||
|
||
def download_paper_by_id(paper_info, chatbot, history) -> str:
|
||
"""下载论文并返回保存路径
|
||
|
||
Args:
|
||
paper_info: 元组，包含论文ID类型（arxiv或doi）和ID值
|
||
chatbot: 聊天机器人对象
|
||
history: 历史记录
|
||
|
||
Returns:
|
||
str: 下载的论文路径或None
|
||
"""
|
||
from crazy_functions.review_fns.data_sources.scihub_source import SciHub
|
||
id_type, paper_id = paper_info
|
||
|
||
# 创建保存目录 - 使用时间戳创建唯一文件夹
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
user_name = chatbot.get_user() if hasattr(chatbot, 'get_user') else "default"
|
||
from toolbox import get_log_folder, get_user
|
||
base_save_dir = get_log_folder(get_user(chatbot), plugin_name='paper_download')
|
||
save_dir = os.path.join(base_save_dir, f"papers_{timestamp}")
|
||
if not os.path.exists(save_dir):
|
||
os.makedirs(save_dir)
|
||
save_path = Path(save_dir)
|
||
|
||
chatbot.append([f"下载论文", f"正在下载{'arXiv' if id_type == 'arxiv' else 'DOI'} {paper_id} 的论文..."])
|
||
update_ui(chatbot=chatbot, history=history)
|
||
|
||
pdf_path = None
|
||
|
||
try:
|
||
if id_type == 'arxiv':
|
||
# 使用改进的arxiv查询方法
|
||
formatted_id = format_arxiv_id(paper_id)
|
||
paper_result = get_arxiv_paper(formatted_id)
|
||
|
||
if not paper_result:
|
||
chatbot.append([f"下载失败", f"未找到arXiv论文: {paper_id}"])
|
||
update_ui(chatbot=chatbot, history=history)
|
||
return None
|
||
|
||
# 下载PDF
|
||
filename = f"arxiv_{paper_id.replace('/', '_')}.pdf"
|
||
pdf_path = str(save_path / filename)
|
||
paper_result.download_pdf(filename=pdf_path)
|
||
|
||
else: # doi
|
||
# 下载DOI
|
||
sci_hub = SciHub(
|
||
doi=paper_id,
|
||
path=save_path
|
||
)
|
||
pdf_path = sci_hub.fetch()
|
||
|
||
# 检查下载结果
|
||
if pdf_path and os.path.exists(pdf_path):
|
||
promote_file_to_downloadzone(pdf_path, chatbot=chatbot)
|
||
chatbot.append([f"下载成功", f"已成功下载论文: {os.path.basename(pdf_path)}"])
|
||
update_ui(chatbot=chatbot, history=history)
|
||
return pdf_path
|
||
else:
|
||
chatbot.append([f"下载失败", f"论文下载失败: {paper_id}"])
|
||
update_ui(chatbot=chatbot, history=history)
|
||
return None
|
||
|
||
except Exception as e:
|
||
chatbot.append([f"下载错误", f"下载论文时出错: {str(e)}"])
|
||
update_ui(chatbot=chatbot, history=history)
|
||
return None
|
||
|
||
|
||
@CatchException
|
||
def 快速论文解读(txt: str, llm_kwargs: Dict, plugin_kwargs: Dict, chatbot: List,
|
||
history: List, system_prompt: str, user_request: str):
|
||
"""主函数 - 论文快速解读"""
|
||
# 初始化分析器
|
||
chatbot.append(["函数插件功能及使用方式", "论文快速解读：通过分析论文的关键要素，帮助您迅速理解论文内容，适用于各学科领域的科研论文。 <br><br>📋 使用方式：<br>1、直接上传PDF文件或者输入DOI号（仅针对SCI hub存在的论文）或arXiv ID（如2501.03916）<br>2、点击插件开始分析"])
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
|
||
paper_file = None
|
||
|
||
# 检查输入是否为论文ID（arxiv或DOI）
|
||
paper_info = extract_paper_id(txt)
|
||
|
||
if paper_info:
|
||
# 如果是论文ID，下载论文
|
||
chatbot.append(["检测到论文ID", f"检测到{'arXiv' if paper_info[0] == 'arxiv' else 'DOI'} ID: {paper_info[1]}，准备下载论文..."])
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
|
||
# 下载论文 - 完全重新实现
|
||
paper_file = download_paper_by_id(paper_info, chatbot, history)
|
||
|
||
if not paper_file:
|
||
report_exception(chatbot, history, a=f"下载论文失败", b=f"无法下载{'arXiv' if paper_info[0] == 'arxiv' else 'DOI'}论文: {paper_info[1]}")
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
return
|
||
else:
|
||
# 检查输入路径
|
||
if not os.path.exists(txt):
|
||
report_exception(chatbot, history, a=f"解析论文: {txt}", b=f"找不到文件或无权访问: {txt}")
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
return
|
||
|
||
# 验证路径安全性
|
||
user_name = chatbot.get_user()
|
||
validate_path_safety(txt, user_name)
|
||
|
||
# 查找论文文件
|
||
paper_file = _find_paper_file(txt)
|
||
|
||
if not paper_file:
|
||
report_exception(chatbot, history, a=f"解析论文", b=f"在路径 {txt} 中未找到支持的论文文件")
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
return
|
||
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
|
||
# 增加调试信息，检查paper_file的类型和值
|
||
chatbot.append(["文件类型检查", f"paper_file类型: {type(paper_file)}, 值: {paper_file}"])
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
chatbot.pop() # 移除调试信息
|
||
|
||
# 确保paper_file是字符串
|
||
if paper_file is not None and not isinstance(paper_file, str):
|
||
# 尝试转换为字符串
|
||
try:
|
||
paper_file = str(paper_file)
|
||
except:
|
||
report_exception(chatbot, history, a=f"类型错误", b=f"论文路径不是有效的字符串: {type(paper_file)}")
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
return
|
||
|
||
# 分析论文
|
||
chatbot.append(["开始分析", f"正在分析论文: {os.path.basename(paper_file)}"])
|
||
yield from update_ui(chatbot=chatbot, history=history)
|
||
|
||
analyzer = PaperAnalyzer(llm_kwargs, plugin_kwargs, chatbot, history, system_prompt)
|
||
yield from analyzer.analyze_paper(paper_file) |