构建高效科研文献管理工作流：从Zotero到AI翻译的实践探索

文献采集 → 元数据管理 → 分类标注 → 阅读笔记 → 引用输出

-- 核心数据表结构（简化版）
CREATE TABLE items (
    itemID INTEGER PRIMARY KEY,
    itemTypeID INT,
    dateAdded TIMESTAMP,
    dateModified TIMESTAMP
);

CREATE TABLE itemData (
    itemID INT,
    fieldID INT,
    valueID INT
);

# 伪代码示例：PDF结构化解析
def parse_academic_pdf(pdf_path):
    doc = PDFDocument(pdf_path)
    
    # 识别文档结构
    sections = identify_sections(doc)  # 标题、正文、公式、表格
    
    # 提取数学公式
    formulas = extract_latex_formulas(doc)
    
    # 识别表格和图表
    tables = extract_tables(doc)
    figures = extract_figures(doc)
    
    return {
        'text_blocks': sections,
        'formulas': formulas,
        'tables': tables,
        'figures': figures
    }

# 伪代码：重建PDF文档
def reconstruct_pdf(translated_data, original_style):
    pdf = PDFBuilder()
    
    # 应用原文档样式
    pdf.set_style(original_style)
    
    # 逐块插入翻译内容
    for block in translated_data:
        if block.type == 'text':
            pdf.add_paragraph(block.content)
        elif block.type == 'formula':
            pdf.add_latex(block.content)  # 保持原公式
        elif block.type == 'table':
            pdf.add_table(block.content)  # 保持表格结构
    
    return pdf.build()

// Zotero插件核心逻辑（简化版）
class TranslationPlugin {
    async translateDocument(item) {
        // 获取PDF附件
        const attachment = item.getAttachments()[0];
        const pdfPath = await attachment.getFilePath();
        
        // 上传到翻译服务
        const taskId = await this.uploadToTranslationService(pdfPath);
        
        // 轮询翻译状态
        const translatedPdf = await this.pollTranslationStatus(taskId);
        
        // 下载并附加到Zotero条目
        await this.attachTranslatedFile(item, translatedPdf);
        
        // 更新元数据
        item.setField('note', `已翻译 (任务ID: ${taskId})`);
        await item.saveTx();
    }
}

# FastAPI示例：翻译服务接口
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/api/translate")
async def create_translation(
    file: UploadFile,
    source_lang: str = "en",
    target_lang: str = "zh",
    optimize_formulas: bool = True
):
    # 保存上传文件
    task_id = generate_task_id()
    file_path = save_upload(file, task_id)
    
    # 异步处理翻译任务
    translation_task.delay(
        task_id=task_id,
        file_path=file_path,
        source_lang=source_lang,
        target_lang=target_lang,
        optimize_formulas=optimize_formulas
    )
    
    return {"task_id": task_id, "status": "INIT"}

@app.get("/api/translate/{task_id}")
async def get_translation_status(task_id: str):
    task = get_task_from_db(task_id)
    
    return {
        "task_id": task_id,
        "status": task.status,  # INIT/PROGRESS/DONE/ERROR
        "progress": task.progress,
        "result_url": task.result_url if task.status == "DONE" else None
    }

# Celery任务队列
@celery_app.task
def translation_task(task_id, file_path, source_lang, target_lang, optimize_formulas):
    try:
        # 1. 文档解析
        update_task_status(task_id, status="PROGRESS", progress=0.1)
        parsed_doc = parse_academic_pdf(file_path)
        
        # 2. 分块翻译
        update_task_status(task_id, status="PROGRESS", progress=0.3)
        translated_blocks = translate_blocks(
            parsed_doc, 
            source_lang, 
            target_lang
        )
        
        # 3. 格式重建
        update_task_status(task_id, status="PROGRESS", progress=0.7)
        output_pdf = reconstruct_pdf(translated_blocks, parsed_doc.style)
        
        # 4. 上传结果
        update_task_status(task_id, status="PROGRESS", progress=0.9)
        result_url = upload_to_storage(output_pdf)
        
        # 5. 完成任务
        update_task_status(
            task_id, 
            status="DONE", 
            progress=1.0, 
            result_url=result_url
        )
        
    except Exception as e:
        update_task_status(task_id, status="ERROR", error_msg=str(e))

# 个人术语表示例（CSV格式）
English,Chinese,Domain,Note
"mitochondrial dysfunction",线粒体功能障碍,医学,统一使用"障碍"而非"失调"
"single-cell sequencing",单细胞测序,生物,简写为scRNA-seq
"deep learning",深度学习,AI,不翻译为"深层学习"