huixiangdou/service/findarticles.py

# yyj
import requests
import xml.etree.ElementTree as ET
import os 
from tqdm import tqdm
import json
import shutil
from loguru import logger
from lxml import etree
import requests
from bs4 import BeautifulSoup
import os

def download_pdfs(path, doi_list): #fox dalao contribution https://github.com/BigWhiteFox
    # 确保下载目录存在
    if not os.path.exists(path):
        os.makedirs(path)
    if isinstance(doi_list, str):
        doi_list = [doi_list]
    href_list = []

    for doi in doi_list:
        url = f"https://sci-hub.se/{doi}"
        response = requests.get(url)

        # 检查请求是否成功
        if response.status_code == 200:
            print(f"成功请求：{url}")
        else:
            print(f"请求失败：{url}，状态码：{response.status_code}")
            continue  # 如果请求失败，跳过本次循环

        soup = BeautifulSoup(response.text, 'html.parser')
        buttons = soup.find_all('button', onclick=True)

        for button in buttons:
            onclick = button.get('onclick')
            if onclick:
                pdf_url = onclick.split("'")[1]
                href_list.append((pdf_url, doi))
                print("pdf_url:", pdf_url)
        print("href_list:", href_list)

    # 遍历href_list中的每个URL
    for href, doi in href_list:
        pdf_url = f"https:{href}"
        try:
            response = requests.get(pdf_url, stream=True)
            if response.status_code == 200:
                filename = doi.replace("/", "_") + ".pdf"
                file_path = os.path.join(path, filename)
                with open(file_path, 'wb') as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        f.write(chunk)
                print(f"File downloaded and saved as: {file_path}")
            else:
                print(f"Download failed, Status Code: {response.status_code}, URL: {pdf_url}")
        except requests.RequestException as e:
            print(f"Failed to download due to an exception: {e}")


class ArticleRetrieval:
    def __init__(self,
                    keywords: list = [],
                    pmids: list = [],
                    repo_dir = 'repodir',
                    retmax = 500):
        if keywords is [] and pmids is []:
            raise ValueError("Either keywords or pmids must be provided.")
        
        self.keywords = keywords
        self.pmids = pmids
        self.repo_dir = repo_dir
        self.retmax = retmax
        self.pmc_ids = []


    def esummary_pmc(self):
        base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi?"
        params = {
            "db": "pubmed",
            "id": ','.join(self.pmids),
            # "retmax": self.retmax
        }
        response = requests.get(base_url, params=params)
        root = ET.fromstring(response.content)
        results = []
        for docsum in root.findall('DocSum'):
            pmcid = None
            doi = None
            id_value = docsum.find('Id').text
            for item in docsum.findall('.//Item[@Name="doi"]'):
                doi = item.text
                break
            for item in docsum.findall('.//Item[@Name="pmc"]'):
                pmcid = item.text
                break

            results.append((id_value, pmcid, doi))
        
        logger.info(f"total {len(results)} articles:")
        logger.info(f"found {len([r for r in results if r[1] is not None])} articles with PMC ID.")
        logger.info(f"found {len([r for r in results if r[2] is not None])} articles with DOI.")
        logger.info(f"found {len([r for r in results if r[1] is None and r[2] is None])} articles without PMC ID and DOI.")
                
        self.esummary = results
        self.pmc_ids = [r[1] for r in results if r[1] is not None]
        self.scihub_doi = [r[2] for r in results if r[1] is None and r[2] is not None]
        self.failed_pmids = [r[0] for r in results if r[1] is None and r[2] is None]
       
    ## 通过Pubmed数据库检索文章
    def esearch_pmc(self):

        base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi"
        params = {
            "db": "pubmed",
            "term": '+'.join(self.keywords),
            "retmax": self.retmax
        }
        response = requests.get(base_url, params=params)
        root = ET.fromstring(response.content)
        idlist = root.find('.//IdList') 
        try:
            pmids = [id_element.text for id_element in idlist.findall('.//Id')]
        except:
            pmids = []

        print(f"Found {len(pmids)} articles for keywords {self.keywords}.")
        self.search_pmid = pmids
        self.pmids.extend(pmids)
        
    # 解析XML文件
    def _get_all_text(self, element):
        """递归获取XML元素及其所有子元素的文本内容。确保element不为None."""
        if element is None:
            return ""
        
        text = element.text or ""
        for child in element:
            text += self._get_all_text(child)
            if child is not None and child.tail:
                text += child.tail
        return text

    ## 清洗XML文件
    def _clean_xml(self,txt):
        parser = etree.XMLParser(recover=True)
        root = ET.fromstring(txt,parser=parser)
        txt = self._get_all_text(root)
        txt = txt.split('REFERENCES')[0]  # 截取参考文献之前的文本
        text = '\n\n'.join([t.strip() for t in txt.split('\n') if len(t.strip())>250])
        return text

    ## 通过PMC数据库获取全文
    def fetch_full_text(self):
        if not os.path.exists(self.repo_dir):
            os.makedirs(self.repo_dir)
        print(f"Saving articles to {self.repo_dir}.")
        self.pmc_success = 0
        self.scihub_success = 0
        self.failed_download = []
        downloaded = os.listdir(self.repo_dir)
        for id in tqdm(self.pmc_ids, desc="Fetching full texts", unit="article"):
            # check if file already downloaded
            if f"{id}.txt" in downloaded:
                print(f"File already downloaded: {id}")
                self.pmc_success += 1
                continue
            base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi"
            params = {
                "db": "pmc",
                "id": id,
                "rettype": "xml",
                "retmode": "text"
            }
            response = requests.get(base_url, params=params)
            full_text = self._clean_xml(response.text)
            if full_text.strip() == '':
                self.failed_download.append(id)
                continue
            else:
                logger.info(full_text[:200])
                with open(os.path.join(self.repo_dir,f'{id}.txt'), 'w') as f:
                    f.write(full_text)
                self.pmc_success += 1
        for doi in tqdm(self.scihub_doi, desc="Fetching full texts", unit="article"):
            # check if file already downloaded
            if f"{doi.replace('/','_')}.pdf" in downloaded: 
                print(f"File already downloaded: {doi}")
                self.scihub_success += 1
                continue

            if download_pdfs(path=self.repo_dir,doi_list = doi):
                self.scihub_success += 1
            else:
                self.failed_download.append(doi)

    def save_config(self):
        config = {
            'repo_dir': self.repo_dir,
            'keywords': self.keywords,
            'retmax': self.retmax,
            "search_pmids": self.search_pmid,
            'import_pmids': [id for id in self.pmids if id not in self.search_pmid],
            'failed_pmids': self.failed_pmids,
            'result': [
                {
                    'pmid': r[0],
                    'pmcid': r[1],
                    'doi': r[2]
                } for r in self.esummary
            ],
            "pmc_success_d": self.pmc_success,
            "scihub_success_d": self.scihub_success,
            "failed_download": self.failed_download,
            
        }
        with open(os.path.join(self.repo_dir, 'info.json'), 'w') as f:
            json.dump(config, f, indent=4, ensure_ascii=False)

    def initiallize(self):
        if self.keywords !=[]:
            print(self.keywords)
            self.esearch_pmc() # get pmids from pubmed database using keywords

        self.esummary_pmc() # get pmc ids from pubmed database using pmids
        self.fetch_full_text() # get full text from pmc database using pmc ids
        self.save_config() # save config file

if __name__ == '__main__':
    if os.path.exists('repodir'):
        shutil.rmtree('repodir')
    
    strings = """
34536239
7760895
36109602
24766875"""
    string = [k.strip() for k in strings.split('\n')]

    pmids = [k for k in string if k.isdigit()]
    print(pmids)
    keys = [k for k in string if not k.isdigit() and k != '']
    print(keys)
    articelfinder = ArticleRetrieval(keywords = keys,pmids = pmids,
                                     repo_dir = 'repodir',retmax = 5)
    articelfinder.initiallize()