“”"

import requests
from bs4 import BeautifulSoup
import os
import time
import random
import logging
from urllib.parse import urlencode
import re
from pathlib import Path

class LiteratureDownloader:
def init(self, doi_file=‘doi_list.txt’, download_dir=‘./downloads/’,
timeout=30, max_retries=3):
“”"

Args:
        doi_file: 包含DOI碼的文本文件路徑
        download_dir: 下載目錄
        timeout: 請求超時時間
        max_retries: 最大重試次數
    """
    self.doi_file = doi_file
    self.download_dir = Path(download_dir)
    self.timeout = timeout
    self.max_retries = max_retries
    
    # 創建下載目錄
    self.download_dir.mkdir(exist_ok=True)
    
    # 設置日誌
    self._setup_logging()
    
    # 設置請求頭
    self.headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1',
    }
    
    # Sci-Hub域名列表（可自動切換）
    self.scihub_domains = [
        'sci-hub.se',
        'sci-hub.st',
        'sci-hub.ru',
        'sci-hub.ee'
    ]
    
    self.current_domain_index = 0
    
def _setup_logging(self):
    """設置日誌配置"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(self.download_dir / 'download.log', encoding='utf-8'),
            logging.StreamHandler()
        ]
    )
    self.logger = logging.getLogger(__name__)

def get_current_domain(self):
    """獲取當前可用的Sci-Hub域名"""
    return self.scihub_domains[self.current_domain_index]

def switch_domain(self):
    """切換到下一個域名"""
    self.current_domain_index = (self.current_domain_index + 1) % len(self.scihub_domains)
    self.logger.info(f"切換到域名: {self.get_current_domain()}")

def read_doi_list(self):
    """
    讀取DOI列表文件
    
    Returns:
        list: DOI碼列表
    """
    try:
        with open(self.doi_file, 'r', encoding='utf-8') as f:
            doi_list = [line.strip() for line in f if line.strip()]
        
        self.logger.info(f"成功讀取 {len(doi_list)} 個DOI碼")
        return doi_list
        
    except FileNotFoundError:
        self.logger.error(f"DOI文件不存在: {self.doi_file}")
        return []
    except Exception as e:
        self.logger.error(f"讀取DOI文件失敗: {e}")
        return []

def construct_scihub_url(self, doi):
    """構造Sci-Hub請求URL"""
    base_url = f"https://{self.get_current_domain()}/{doi}"
    return base_url

def extract_download_url(self, html_content):
    """
    從HTML內容中提取下載鏈接
    
    Args:
        html_content: 網頁HTML內容
        
    Returns:
        str: 下載鏈接或None
    """
    try:
        soup = BeautifulSoup(html_content, 'html.parser')
        
        # 方法1: 通過iframe的src屬性獲取
        iframe = soup.find('iframe')
        if iframe and iframe.get('src'):
            download_url = iframe.get('src')
            if download_url.startswith('//'):
                download_url = 'https:' + download_url
            return download_url
        
        # 方法2: 通過按鈕的onclick屬性獲取
        button = soup.find('button', onclick=re.compile(r"location.href"))
        if button:
            onclick = button.get('onclick', '')
            match = re.search(r"location\.href='([^']+)'", onclick)
            if match:
                return match.group(1)
        
        # 方法3: 通過鏈接的href屬性獲取
        pdf_link = soup.find('a', href=re.compile(r'\.pdf'))
        if pdf_link:
            download_url = pdf_link.get('href')
            if download_url.startswith('//'):
                download_url = 'https:' + download_url
            return download_url
        
        return None
        
    except Exception as e:
        self.logger.error(f"解析下載鏈接失敗: {e}")
        return None

def download_pdf(self, download_url, filename, doi):
    """
    下載PDF文件
    
    Args:
        download_url: 下載鏈接
        filename: 保存的文件名
        doi: DOI碼（用於日誌）
        
    Returns:
        bool: 是否下載成功
    """
    try:
        response = requests.get(
            download_url, 
            headers=self.headers, 
            timeout=self.timeout,
            stream=True
        )
        
        if response.status_code == 200:
            filepath = self.download_dir / filename
            
            with open(filepath, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
            
            # 檢查文件大小，避免下載到錯誤頁面
            file_size = filepath.stat().st_size
            if file_size < 1024:  # 小於1KB可能是錯誤頁面
                filepath.unlink()
                self.logger.warning(f"文件大小異常，可能下載失敗: {doi}")
                return False
            
            self.logger.info(f"成功下載: {filename} ({file_size} bytes)")
            return True
        else:
            self.logger.warning(f"下載請求失敗: {response.status_code}")
            return False
            
    except Exception as e:
        self.logger.error(f"下載過程出錯: {e}")
        return False

def sanitize_filename(self, doi):
    """
    根據DOI生成安全的文件名
    
    Args:
        doi: DOI碼
        
    Returns:
        str: 安全的文件名
    """
    # 替換文件名中的非法字符
    safe_name = re.sub(r'[\\/*?:"<>|]', "_", doi)
    return f"{safe_name}.pdf"

def process_single_doi(self, doi, retry_count=0):
    """
    處理單個DOI碼的下載
    
    Args:
        doi: DOI碼
        retry_count: 當前重試次數
        
    Returns:
        bool: 是否成功下載
    """
    if retry_count >= self.max_retries:
        self.logger.error(f"達到最大重試次數，跳過: {doi}")
        return False
    
    try:
        self.logger.info(f"處理DOI: {doi}")
        
        # 構造Sci-Hub URL
        scihub_url = self.construct_scihub_url(doi)
        
        # 發送請求
        response = requests.get(
            scihub_url, 
            headers=self.headers, 
            timeout=self.timeout
        )
        
        if response.status_code != 200:
            self.logger.warning(f"請求失敗: {response.status_code}")
            if response.status_code in [403, 404]:
                self.switch_domain()
                return self.process_single_doi(doi, retry_count + 1)
            return False
        
        # 提取下載鏈接
        download_url = self.extract_download_url(response.text)
        
        if not download_url:
            self.logger.warning(f"無法提取下載鏈接: {doi}")
            # 嘗試切換域名重試
            self.switch_domain()
            return self.process_single_doi(doi, retry_count + 1)
        
        # 生成文件名並下載
        filename = self.sanitize_filename(doi)
        success = self.download_pdf(download_url, filename, doi)
        
        if not success and retry_count < self.max_retries - 1:
            # 下載失敗，切換域名重試
            self.switch_domain()
            time.sleep(2)  # 等待一段時間再重試
            return self.process_single_doi(doi, retry_count + 1)
        
        return success
        
    except requests.RequestException as e:
        self.logger.error(f"網絡請求異常: {e}")
        if retry_count < self.max_retries - 1:
            self.switch_domain()
            time.sleep(2)
            return self.process_single_doi(doi, retry_count + 1)
        return False
    except Exception as e:
        self.logger.error(f"處理DOI時發生未知錯誤: {e}")
        return False

def run(self):
    """運行批量下載"""
    self.logger.info("=== 開始批量下載文獻 ===")
    
    # 讀取DOI列表
    doi_list = self.read_doi_list()
    if not doi_list:
        self.logger.error("沒有可用的DOI碼，程序退出")
        return
    
    # 統計信息
    success_count = 0
    failed_dois = []
    
    # 逐個處理DOI
    for i, doi in enumerate(doi_list, 1):
        self.logger.info(f"進度: {i}/{len(doi_list)}")
        
        success = self.process_single_doi(doi)
        
        if success:
            success_count += 1
        else:
            failed_dois.append(doi)
        
        # 隨機延遲，避免請求過於頻繁
        time.sleep(random.uniform(1, 3))
    
    # 輸出統計結果
    self.logger.info("=== 下載完成 ===")
    self.logger.info(f"成功下載: {success_count}/{len(doi_list)}")
    
    if failed_dois:
        self.logger.info("失敗的DOI碼:")
        for doi in failed_dois:
            self.logger.info(f"  {doi}")
        
        # 保存失敗列表到文件
        failed_file = self.download_dir / 'failed_dois.txt'
        with open(failed_file, 'w', encoding='utf-8') as f:
            for doi in failed_dois:
                f.write(doi + '\n')
        self.logger.info(f"失敗列表已保存至: {failed_file}")

def main():
“”“主函數”“”
# 使用示例
downloader = LiteratureDownloader(
doi_file=‘doi_list.txt’, # DOI文件路徑
download_dir=‘./articles/’, # 下載目錄
timeout=30, # 超時時間
max_retries=3 # 最大重試次數
)

downloader.run()
AI寫代碼
1
if name == “main”:
# 免責聲明
print(“=” * 60)
print(“文獻下載工具 - 僅用於學術研究”)
print(“請尊重版權，僅下載您有合法訪問權限的文獻”)
print(“請勿用於商業用途或大規模批量下載”)
print(“=” * 60)

main()