Zotero 8.0.1 英文文献批量下载与自动化管理实战

Zotero 默认不连接 Sci-Hub，但用户手册允许通过配置 resolvers 来自定义 PDF 查找路径。这通常用于处理未在上述渠道找到的文献。

注意：部分第三方镜像站点可能涉及版权风险，请自行评估合规性。插件通常会提供一组镜像 URL，这些链接可能会失效，需要定期维护。

三、实操步骤

1. 导入文献

1. 在 Zotero 左侧栏新建一个分类文件夹。
2. 将导出的 .ris 文件拖入该文件夹。
3. 双击 RIS 文件，勾选'导入到新收藏'，系统会自动根据文件名创建子文件夹并导入文献。

2. 获取全文

在 Zotero 设置中指定本地保存路径。数据保存在 storage 目录下，每个文献对应一个子目录。

• 单篇获取：右键条目 -> 查找全文。
• 批量获取：全选所有文献 (Ctrl + A) -> 右键 -> 查找全文。

如果自动获取失败率较高，可以考虑安装专门的插件来扩展查找源。例如 zotero-scipdf 插件，它可以内置一些镜像站点配置。

安装方法：

1. 从 GitHub Releases 页面下载 .xpi 安装包。
2. 在 Zotero 中：工具 -> 插件 -> 设置 -> 从文件安装。
3. 配置镜像 URL：编辑 -> 设置 -> 插件设置。格式通常为 https://域名/{doi}，多个 URL 用分号分隔。

提示：镜像站点的可用性是动态变化的。可以在 工具 -> 开发者 -> error console 中查看报错信息，及时剔除失效的链接。

3. 处理未下载成功的文献

对于有 DOI 但自动下载失败的文献，可以编写 Python 脚本进行针对性处理。

准备工作：

1. 在 Zotero 中将未下载附件的文献单独归类到一个文件夹。
2. 导出该分类为 RIS 格式，提取 DOI 列表。

Python 脚本示例： 下面是一个基于并行请求的下载脚本，支持多镜像轮询，并具备断点续传和日志记录功能。

# -*- coding = utf-8 -*-
import os
import time
import requests
import pandas as pd
import rispy
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor, as_completed
import warnings
import re

warnings.filterwarnings('ignore')
os.environ['CURL_CA_BUNDLE'] = ''

# ================= 配置区域 =================
RIS_PATH = r"xxx.ris"
DOWNLOAD_DIR = r"xxx\未下载"
LOG_FILE = os.path.join(DOWNLOAD_DIR, "download_status_log.csv")

# 7 个镜像站点模板
SCIHUB_MIRRORS_TEMPLATE = [
    "https://www.pismin.com/{doi}",
    "https://sci-hub.st/{doi}",
    "https://sci-hub.ru/{doi}",
    "https://sci-hub.box/{doi}",
    "https://sci-hub.red/{doi}",
    "https://sci-hub.ren/{doi}",
    "https://sci-hub.ee/{doi}"
]

HEADERS = {
    'User-Agent': 'Mozilla/5.0(Windows NT 10.0; Win64; x64) AppleWebKit/537.36(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

# ================= 核心功能函数 =================
def sanitize_filename(filename):
    if not isinstance(filename, str): return "Unknown_Filename"
    filename = re.sub(r'[\\/*?:"<>|]', '_', filename)
    filename = re.sub(r'\s+', ' ', filename).strip()
    return filename[:200]

def parse_ris_robust(file_path):
    print(f"[解析] 正在读取 RIS 文件：{file_path}")
    with open(file_path, 'r', encoding='utf-8-sig', errors='ignore') as f:
        entries = rispy.load(f)
    parsed_data = []
    for entry in entries:
        doi = entry.get('doi') or entry.get('DO') or entry.get('number')
        if not doi:
            notes = entry.get('notes', [])
            if isinstance(notes, list):
                for n in notes:
                    if '10.' in n and '/' in n:
                        doi = n
                        break
        
        title = entry.get('title') or entry.get('primary_title') or entry.get('TI') or entry.get('T1') or entry.get('T2')
        if not title: title = "Unknown_Title"
        
        year = entry.get('year') or entry.get('publication_year') or entry.get('PY') or entry.get('Y1')
        if not year:
            date = entry.get('date') or entry.get('DA')
            if date: year = date[:4]
            else: year = "NoYear"
        
        doi = str(doi).strip() if doi else ""
        title = str(title).strip()
        year = str(year).strip()
        clean_title = sanitize_filename(title)
        filename = f"[{year}] {clean_title}.pdf"
        status = 'Pending' if doi else 'No_DOI'
        
        if not doi:
            print(f" [警告] 发现无 DOI 文献：{title[:30]}...")
        
        parsed_data.append({
            'DOI': doi, 'Title': title, 'Filename': filename,
            'Status': status, 'Message': ''
        })
    return pd.DataFrame(parsed_data)

def init_task_manager():
    if not os.path.exists(DOWNLOAD_DIR): os.makedirs(DOWNLOAD_DIR)
    rebuild = False
    if os.path.exists(LOG_FILE):
        df = pd.read_csv(LOG_FILE)
        unknown_count = df['Title'].str.contains('Unknown Title', case=False, na=False).sum()
        if unknown_count > 5:
            print(f"[检测] 旧记录文件包含 {unknown_count} 个未知标题，判定为解析失败。正在强制重新解析 RIS...")
            rebuild = True
        else:
            print(f"[读取] 加载现有进度，共 {len(df)} 条记录。")
            df.loc[df['Status'] == 'Downloading', 'Status'] = 'Pending'
    else:
        rebuild = True
    
    if rebuild:
        df = parse_ris_robust(RIS_PATH)
        df.to_csv(LOG_FILE, index=False)
        print(f"[构建] 新的统计文件已创建，共 {len(df)} 条。")
    return df

def get_pdf_direct_link(session, url):
    try:
        resp = session.get(url, headers=HEADERS, timeout=10, verify=False, allow_redirects=True)
        if resp.status_code != 200: return None
        soup = BeautifulSoup(resp.content, 'html.parser')
        target = soup.find('embed', attrs={'type': 'application/pdf'}) or \
                 soup.find('iframe', attrs={'src': re.compile(r'\.pdf')})
        if target and target.get('src'):
            raw_url = target.get('src')
            if raw_url.startswith('//'): return 'https:' + raw_url
            if raw_url.startswith('/'): return '/'.join(url.split('/')[:3]) + raw_url
            return raw_url
        btn = soup.find('button', onclick=True)
        if btn and 'location.href' in btn['onclick']:
            match = re.search(r"href='(.*?)'", btn['onclick'])
            if match:
                raw_url = match.group(1)
                if raw_url.startswith('//'): return 'https:' + raw_url
                return raw_url
    except Exception: pass
    return None

def attempt_download_single_mirror(url_template, doi, save_path):
    mirror_url = url_template.replace("{doi}", doi)
    session = requests.Session()
    session.mount('https', requests.adapters.HTTPAdapter(max_retries=1))
    try:
        pdf_url = get_pdf_direct_link(session, mirror_url)
        if not pdf_url: return False, "Page parsed but no PDF found"
        r = session.get(pdf_url, headers=HEADERS, stream=True, timeout=20, verify=False)
        ct = r.headers.get('Content-Type', '').lower()
        if 'html' in ct or len(r.content) < 1000: return False, "Not a PDF file"
        if r.status_code == 200:
            with open(save_path, 'wb') as f:
                for chunk in r.iter_content(chunk_size=8192): f.write(chunk)
            return True, mirror_url
    except Exception as e: return False, str(e)
    return False, "Unknown Error"

def parallel_download_handler(index, row, df):
    doi = row['DOI']
    filename = row['Filename']
    save_path = os.path.join(DOWNLOAD_DIR, filename)
    print(f"\n--> [{index + 1}/{len(df)}] 开始并行下载：{doi}")
    print(f" 目标文件：{filename}")
    df.at[index, 'Status'] = 'Downloading'
    success = False
    winning_mirror = ""
    
    with ThreadPoolExecutor(max_workers=8) as executor:
        future_to_url = {
            executor.submit(attempt_download_single_mirror, url, doi, save_path): url 
            for url in SCIHUB_MIRRORS_TEMPLATE
        }
        for future in as_completed(future_to_url):
            url = future_to_url[future]
            try:
                is_success, msg = future.result()
                if is_success:
                    success = True
                    winning_mirror = msg
                    print(f" [√] 成功！来源镜像：{winning_mirror.split('/')[2]}")
                    executor.shutdown(wait=False, cancel_futures=True)
                    break
            except Exception: continue
    
    if success:
        df.at[index, 'Status'] = 'Downloaded'
        df.at[index, 'Message'] = f"From {winning_mirror}"
    else:
        df.at[index, 'Status'] = 'Failed'
        df.at[index, 'Message'] = "All mirrors failed or timed out"
        print(f" [X] 所有镜像均失败：{doi}")
    
    df.to_csv(LOG_FILE, index=False)

def main():
    print("=== WoS 极速下载器 (并行版) ===")
    df = init_task_manager()
    tasks = df[df['Status'] == 'Pending']
    total = len(tasks)
    print(f"\n=== 待处理队列：{total} 个文献 ===")
    if total == 0:
        print("没有由于下载的任务。检查是否需要重置 'Failed' 状态。")
        return
    
    for index, row in tasks.iterrows():
        file_path = os.path.join(DOWNLOAD_DIR, row['Filename'])
        if os.path.exists(file_path) and os.path.getsize(file_path) > 2000:
            print(f"[{index + 1}] 文件已存在，跳过：{row['Filename']}")
            df.at[index, 'Status'] = 'Downloaded'
            df.at[index, 'Message'] = 'File exists'
            df.to_csv(LOG_FILE, index=False)
            continue
        parallel_download_handler(index, row, df)
        time.sleep(1)

if __name__ == "__main__":
    main()

运行前请修改 RIS_PATH 和 DOWNLOAD_DIR 路径。脚本会自动记录下载状态，避免重复下载。

4. 整理文献

如果需要将分散的 PDF 集中到一个目录以便后续分析（如 LLM 阅读），可以使用以下脚本。

# -*- coding = utf-8 -*-
import os
import shutil
from pathlib import Path

def collect_zotero_pdfs(src_dir, dest_dir):
    src_path = Path(src_dir)
    dest_path = Path(dest_dir)
    if not dest_path.exists():
        dest_path.mkdir(parents=True, exist_ok=True)
    print(f"已创建输出目录：{dest_path}")
    print("正在扫描并复制文件，请稍候...")
    count = 0
    for pdf_file in src_path.rglob('*.pdf'):
        try:
            target_file = dest_path / pdf_file.name
            if target_file.exists():
                stem = pdf_file.stem
                suffix = pdf_file.suffix
                counter = 1
                while target_file.exists():
                    target_file = dest_path / f"{stem}_{counter}{suffix}"
                    counter += 1
            shutil.copy2(pdf_file, target_file)
            count += 1
            if count % 10 == 0:
                print(f"已处理 {count} 个文件...")
        except Exception as e:
            print(f"处理文件 {pdf_file.name} 时出错：{e}")
    print("-" * 30)
    print(f"任务完成！共成功复制 {count} 个 PDF 文件到：")
    print(dest_path)

if __name__ == "__main__":
    input_directory = r"你的 zotero 的数据保存路径\storage"
    output_directory = r"输出目录"
    collect_zotero_pdfs(input_directory, output_directory)

四、AI 插件深度分析

Zotero 支持多种 AI 插件来辅助阅读。以对话插件为例，其他如翻译插件可自行安装。

1. 安装：从社区或 GitHub 下载插件包并安装。
2. 配置：在插件设置中选择模型（如 Gemini、DeepSeek 等），填入 API Key。
   • 若使用 Gemini，API ID 可能需要手动填写（如 gemini-3-flash-preview）。
   • Token 限制等参数可根据需求调整。
3. 使用：选中一篇文献，点击插件图标，选择功能（如总结、问答）即可。

建议：不同模型的 API 成本差异较大，建议先测试小样本再批量使用。DeepSeek 等国内模型门槛较低，适合日常查阅。

注：本文档仅供技术交流，使用第三方下载服务时请注意遵守相关法律法规及知识产权规定。