wqb-server/blueprints/paper_analysis.py

"""
论文分析蓝图 - 使用 Deepseek AI 分析研究论文的 Flask 蓝图
"""

from flask import Blueprint, render_template, request, jsonify
import requests
import json
import os
import tempfile
from werkzeug.utils import secure_filename

# 创建蓝图
paper_analysis_bp = Blueprint('paper_analysis', __name__, url_prefix='/paper-analysis')

@paper_analysis_bp.route('/')
def paper_analysis():
    """论文分析页面"""
    return render_template('paper_analysis.html')

@paper_analysis_bp.route('/api/test-deepseek', methods=['POST'])
def test_deepseek():
    """测试 Deepseek API 连接"""
    try:
        api_key = request.headers.get('X-API-Key')
        if not api_key:
            return jsonify({'error': '需要 API 密钥'}), 401

        # 使用简单提示测试 API
        headers = {
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }

        test_response = requests.post(
            'https://api.deepseek.com/v1/chat/completions',  # 使用聊天补全端点
            headers=headers,
            json={
                'model': 'deepseek-chat',
                'messages': [
                    {'role': 'user', 'content': '打个招呼'}
                ],
                'max_tokens': 10
            },
            timeout=10
        )

        if test_response.ok:
            return jsonify({
                'success': True,
                'message': 'Deepseek API 连接成功',
                'response': test_response.json()
            })
        else:
            return jsonify({
                'success': False,
                'error': f'API 错误: {test_response.status_code}',
                'details': test_response.text
            }), test_response.status_code

    except requests.exceptions.RequestException as e:
        return jsonify({
            'success': False,
            'error': '连接错误',
            'details': str(e)
        }), 500
    except Exception as e:
        return jsonify({
            'success': False,
            'error': '意外错误',
            'details': str(e)
        }), 500

@paper_analysis_bp.route('/api/analyze-paper', methods=['POST'])
def analyze_paper():
    """使用 Deepseek API 分析论文"""
    try:
        # 从请求头获取 API 密钥
        api_key = request.headers.get('X-API-Key')
        if not api_key:
            return jsonify({'error': '需要 API 密钥'}), 401

        # 获取分析选项
        extract_keywords = request.form.get('extract_keywords') == 'true'
        generate_summary = request.form.get('generate_summary') == 'true'
        find_related = request.form.get('find_related') == 'true'

        # 获取上传的文件
        if 'file' not in request.files:
            return jsonify({'error': '没有上传文件'}), 400

        file = request.files['file']
        if file.filename == '':
            return jsonify({'error': '没有选择文件'}), 400

        # 检查文件大小（限制为 50MB）
        file.seek(0, 2)  # 定位到文件末尾
        file_size = file.tell()
        file.seek(0)  # 重置到开头

        if file_size > 50 * 1024 * 1024:  # 50MB 限制
            return jsonify({'error': '文件过大。最大大小为 50MB'}), 400

        if file_size == 0:
            return jsonify({'error': '文件为空'}), 400

        # 临时保存文件
        filename = secure_filename(file.filename)
        print(f"正在处理文件: {filename} (大小: {file_size} 字节)")

        with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(filename)[1]) as temp_file:
            file.save(temp_file.name)
            file_path = temp_file.name

        try:
            # 初始化结果字典
            results = {
                'keywords': [],
                'summary': '',
                'related_works': []
            }

            # 从文件提取文本
            text = extract_text_from_file(file_path, filename)

            if not text or not text.strip():
                return jsonify({'error': '无法从文件中提取文本。文件可能为空或格式不受支持。'}), 400

            # 清理文本
            text = text.strip()
            print(f"截断前的最终文本长度: {len(text)}")

            # 检查是否有足够的文本
            if len(text) < 100:
                return jsonify({
                    'error': '提取的文本过短。这可能是没有 OCR 文本的扫描 PDF。请确保您的 PDF 包含可选择的文本，而不仅仅是图像。'
                }), 400

            # 处理大型文档
            text = process_large_document(text)

            # 为每个请求的分析调用 Deepseek API
            headers = {
                'Authorization': f'Bearer {api_key}',
                'Content-Type': 'application/json'
            }

            if extract_keywords:
                results['keywords'] = extract_keywords_with_deepseek(text, headers)

            if generate_summary:
                results['summary'] = generate_summary_with_deepseek(text, headers)

            if find_related:
                results['related_works'] = extract_formulas_with_deepseek(text, headers)

            return jsonify(results)

        finally:
            # 清理临时文件
            try:
                os.unlink(file_path)
            except Exception as e:
                print(f"删除临时文件时出错: {str(e)}")

    except Exception as e:
        print(f"分析论文时出错: {str(e)}")
        return jsonify({'error': str(e)}), 500

def extract_text_from_file(file_path, filename):
    """从各种文件格式中提取文本"""
    text = ''
    file_ext = os.path.splitext(filename)[1].lower()

    try:
        if file_ext == '.pdf':
            text = extract_pdf_text(file_path)
        elif file_ext in ['.docx', '.doc']:
            text = extract_word_text(file_path, file_ext)
        elif file_ext == '.rtf':
            text = extract_rtf_text(file_path)
        elif file_ext in ['.tex', '.latex']:
            text = extract_latex_text(file_path)
        elif file_ext in ['.md', '.markdown']:
            text = extract_markdown_text(file_path)
        else:
            text = extract_plain_text(file_path)

    except Exception as e:
        print(f"文件处理错误: {str(e)}")
        raise Exception(f"读取文件时出错: {str(e)}")

    return text

def extract_pdf_text(file_path):
    """从 PDF 文件提取文本"""
    try:
        from PyPDF2 import PdfReader
        reader = PdfReader(file_path)
        text = ''
        num_pages = len(reader.pages)
        print(f"PDF 有 {num_pages} 页")

        for i, page in enumerate(reader.pages):
            try:
                page_text = page.extract_text()
                if page_text:
                    text += page_text + '\n'
                print(f"已提取第 {i+1}/{num_pages} 页")
            except Exception as page_error:
                print(f"提取第 {i+1} 页时出错: {str(page_error)}")
                continue

        print(f"提取的总文本长度: {len(text)}")
        return text

    except ImportError:
        # 尝试替代的 PDF 库
        try:
            import pdfplumber
            text = ''
            with pdfplumber.open(file_path) as pdf:
                for page in pdf.pages:
                    page_text = page.extract_text()
                    if page_text:
                        text += page_text + '\n'
            return text
        except ImportError:
            raise Exception('PDF 处理不可用。请安装 PyPDF2 或 pdfplumber。')
    except Exception as pdf_error:
        print(f"PDF 提取错误: {str(pdf_error)}")
        # 尝试使用 PyMuPDF 作为后备方案
        try:
            import fitz  # PyMuPDF
            pdf_document = fitz.open(file_path)
            text = ''
            for page_num in range(pdf_document.page_count):
                page = pdf_document[page_num]
                text += page.get_text() + '\n'
            pdf_document.close()
            return text
        except ImportError:
            raise Exception(f'无法从 PDF 提取文本: {str(pdf_error)}。请尝试安装 PyMuPDF。')
        except Exception as mupdf_error:
            raise Exception(f'PDF 提取失败: {str(pdf_error)}')

def extract_word_text(file_path, file_ext):
    """从 Word 文档提取文本"""
    try:
        if file_ext == '.docx':
            from docx import Document
            doc = Document(file_path)
            return '\n'.join([paragraph.text for paragraph in doc.paragraphs])
        else:
            # .doc 文件
            try:
                import docx2txt
                return docx2txt.process(file_path)
            except ImportError:
                raise Exception('DOC 文件支持需要 docx2txt。请使用以下命令安装: pip install docx2txt')
    except ImportError:
        raise Exception('Word 文档支持需要 python-docx。请使用以下命令安装: pip install python-docx')
    except Exception as docx_error:
        raise Exception(f'读取 Word 文档时出错: {str(docx_error)}')

def extract_rtf_text(file_path):
    """从 RTF 文件提取文本"""
    try:
        import striprtf
        with open(file_path, 'r', encoding='utf-8') as f:
            rtf_content = f.read()
        return striprtf.rtf_to_text(rtf_content)
    except ImportError:
        raise Exception('RTF 支持需要 striprtf。请使用以下命令安装: pip install striprtf')
    except Exception as rtf_error:
        raise Exception(f'读取 RTF 文件时出错: {str(rtf_error)}')

def extract_latex_text(file_path):
    """从 LaTeX 文件提取文本"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            tex_content = f.read()
        # 基础 LaTeX 清理 - 移除常见命令
        import re
        text = tex_content
        # 移除注释
        text = re.sub(r'%.*$', '', text, flags=re.MULTILINE)
        # 移除常见 LaTeX 命令但保留内容
        text = re.sub(r'\\(begin|end)\{[^}]+\}', '', text)
        text = re.sub(r'\\[a-zA-Z]+\*?\{([^}]+)\}', r'\1', text)
        text = re.sub(r'\\[a-zA-Z]+\*?', '', text)
        return text
    except Exception as tex_error:
        raise Exception(f'读取 LaTeX 文件时出错: {str(tex_error)}')

def extract_markdown_text(file_path):
    """从 Markdown 文件提取文本"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            text = f.read()
        # 清理 markdown 语法
        import re
        # 移除图片链接
        text = re.sub(r'!\[[^\]]*\]\([^)]+\)', '', text)
        # 将链接转换为纯文本
        text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', text)
        return text
    except Exception as md_error:
        raise Exception(f'读取 Markdown 文件时出错: {str(md_error)}')

def extract_plain_text(file_path):
    """从纯文本文件提取文本"""
    encodings = ['utf-8', 'utf-16', 'gbk', 'gb2312', 'big5', 'latin-1']
    text = None

    for encoding in encodings:
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                text = f.read()
                print(f"使用 {encoding} 编码成功读取文件")
                break
        except UnicodeDecodeError:
            continue
        except Exception as e:
            print(f"使用 {encoding} 读取时出错: {str(e)}")
            continue

    if text is None:
        # 尝试以二进制方式读取并解码
        with open(file_path, 'rb') as f:
            binary_content = f.read()
            try:
                text = binary_content.decode('utf-8', errors='ignore')
            except:
                text = str(binary_content)

    return text

def process_large_document(text):
    """通过优先处理公式提取来处理大型文档"""
    if len(text) > 98000:
        print("检测到大型文档，优先处理内容以进行公式提取")
        # 尝试找到包含公式的章节（常见模式）
        import re
        # 查找数学内容指示符
        math_sections = []
        lines = text.split('\n')
        for i, line in enumerate(lines):
            if re.search(r'[=+\-*/∫∑∏√∂∇∆λμσπ]|equation|formula|theorem|lemma|proof', line, re.IGNORECASE):
                # 包含周围上下文
                start = max(0, i-5)
                end = min(len(lines), i+6)
                math_sections.extend(lines[start:end])

        if math_sections:
            # 使用数学内容丰富的章节以更好地提取公式
            math_text = '\n'.join(math_sections)
            if len(math_text) > 50000:  # 仍然太长
                text = math_text[:98000]
            else:
                # 将数学章节与文档开头结合
                remaining_space = 98000 - len(math_text)
                text = text[:remaining_space] + '\n\n[数学内容章节:]\n' + math_text
        else:
            # 未找到数学指示符，使用第一部分
            text = text[:98000]

    return text

def extract_keywords_with_deepseek(text, headers):
    """使用 Deepseek API 提取关键词"""
    try:
        keyword_messages = [
            {
                'role': 'system',
                'content': '你是一个有助于从学术论文中提取关键词的助手。始终以有效的 JSON 格式响应。'
            },
            {
                'role': 'user',
                'content': f"""分析以下学术论文并提取关键术语和概念。
对于每个关键词，提供一个介于 0 到 1 之间的相关性分数。
仅返回具有 'text' 和 'score' 属性的有效 JSON 对象数组。
示例格式: [{{"text": "machine learning", "score": 0.95}}, {{"text": "neural networks", "score": 0.85}}]

论文文本:
{text}"""
            }
        ]

        keyword_response = requests.post(
            'https://api.deepseek.com/v1/chat/completions',
            headers=headers,
            json={
                'model': 'deepseek-chat',
                'messages': keyword_messages,
                'temperature': 0.3,
                'max_tokens': 4000
            },
            timeout=60
        )

        if keyword_response.ok:
            response_content = keyword_response.json()['choices'][0]['message']['content']
            try:
                # 尝试从响应中提取 JSON
                import re
                json_match = re.search(r'\[.*\]', response_content, re.DOTALL)
                if json_match:
                    return json.loads(json_match.group())
                else:
                    return json.loads(response_content)
            except json.JSONDecodeError:
                print(f"来自关键词 API 的无效 JSON: {response_content}")
                return []
        else:
            print(f"关键词 API 错误: {keyword_response.text}")
            return []

    except Exception as e:
        print(f"关键词提取错误: {str(e)}")
        return []

def generate_summary_with_deepseek(text, headers):
    """使用 Deepseek API 生成摘要"""
    try:
        summary_messages = [
            {
                'role': 'system',
                'content': '你是一个有助于总结学术论文的助手。'
            },
            {
                'role': 'user',
                'content': f"""提供以下学术论文的全面摘要。
重点关注主要贡献、方法和关键发现。
保持回应简洁且结构良好。

论文文本:
{text}"""
            }
        ]

        summary_response = requests.post(
            'https://api.deepseek.com/v1/chat/completions',
            headers=headers,
            json={
                'model': 'deepseek-chat',
                'messages': summary_messages,
                'temperature': 0.3,
                'max_tokens': 4000
            },
            timeout=60
        )

        if summary_response.ok:
            return summary_response.json()['choices'][0]['message']['content']
        else:
            print(f"摘要 API 错误: {summary_response.text}")
            return "生成摘要时出错"

    except Exception as e:
        print(f"摘要生成错误: {str(e)}")
        return "生成摘要时出错"

def extract_formulas_with_deepseek(text, headers):
    """使用 Deepseek API 提取公式"""
    try:
        related_messages = [
            {
                'role': 'system',
                'content': '''你是一位专业的数学家和 AI 助手，专门从学术论文中提取数学公式。
你的任务是识别并提取给定文本中所有的数学公式、方程式和数学表达式，尽可能多地提取。

重要说明:
1. 提取你找到的每一个数学公式、方程式或表达式
2. 包括内联公式、显示方程和数学定义
3. 尽可能保留原始符号
4. 对于每个公式，提供其表示内容的上下文
5. 始终以有效的 JSON 格式响应

你必须彻底并提取所有公式，而不仅仅是主要的公式。'''
            },
            {
                'role': 'user',
                'content': f"""从以下论文文本中提取所有的数学公式和方程式。

对于找到的每个公式，请提供:
- 公式本身（尽可能使用 LaTeX 表示法）
- 详细描述，解释公式表示的内容以及每个变量的含义
- 出现位置的上下文或章节
- 它是定义、定理、引理还是一般方程
- 解释公式目的的中文描述

返回一个 JSON 数组，其中每个元素具有以下属性:
- "formula": 数学表达式（使用 LaTeX 表示法）
- "description": 公式表示或计算的内容
- "variables": 对公式中每个变量/符号含义的详细解释
- "variables_chinese": 变量解释的中文翻译（与 variables 结构相同）
- "type": 其中之一 ["definition", "theorem", "lemma", "equation", "inequality", "identity", "other"]
- "context": 关于其使用位置/方式的简要上下文
- "chinese_description": 关于公式及其目的的综合中文描述

示例格式:
[
  {{
    "formula": "E = mc^2",
    "description": "爱因斯坦质能等价关系",
    "variables": {{"E": "energy (joules)", "m": "mass (kilograms)", "c": "speed of light in vacuum (≈3×10^8 m/s)"}},
    "variables_chinese": {{"E": "能量 (焦耳)", "m": "质量 (千克)", "c": "真空中的光速 (≈3×10^8 m/s)"}},
    "type": "equation",
    "context": "狭义相对论基本方程",
    "chinese_description": "爱因斯坦质能等价公式，表示质量和能量之间的等价关系"
  }},
  {{
    "formula": "F = ma",
    "description": "牛顿第二运动定律",
    "variables": {{"F": "net force (newtons)", "m": "mass (kilograms)", "a": "acceleration (m/s²)"}},
    "variables_chinese": {{"F": "净力 (牛顿)", "m": "质量 (千克)", "a": "加速度 (m/s²)"}},
    "type": "equation", 
    "context": "经典力学基本定律",
    "chinese_description": "牛顿第二定律，描述物体受力与加速度的关系"
  }}
]

论文文本:
{text}

重要说明:
1. 提取每一个公式，即使是简单的如 "x + y = z" 或 "f(x) = ax + b"
2. 对于公式中的每个变量或符号，解释其代表什么
3. 相关时包括测量单位
4. 提供解释公式重要性的全面中文描述
5. 在变量解释中要彻底且详细"""
            }
        ]

        related_response = requests.post(
            'https://api.deepseek.com/v1/chat/completions',
            headers=headers,
            json={
                'model': 'deepseek-chat',
                'messages': related_messages,
                'temperature': 0.1,  # 较低的温度以获得更一致的提取
                'max_tokens': 4000   # 增加令牌限制以获取更多公式
            },
            timeout=120  # 增加超时时间以处理大型文档
        )

        if related_response.ok:
            response_content = related_response.json()['choices'][0]['message']['content']
            try:
                # 尝试从响应中提取 JSON
                import re
                # 在响应中查找 JSON 数组
                json_match = re.search(r'\[[\s\S]*\]', response_content)
                if json_match:
                    formulas = json.loads(json_match.group())
                    return formulas
                else:
                    # 尝试直接 JSON 解析
                    return json.loads(response_content)
            except json.JSONDecodeError as e:
                print(f"来自公式 API 的无效 JSON: {response_content}")
                print(f"JSON 错误: {str(e)}")
                return []
        else:
            print(f"公式 API 错误: {related_response.text}")
            return []

    except Exception as e:
        print(f"公式提取错误: {str(e)}")
        return []