src/crossref_client.py - common_agents_notes - IvanMysin/Topics

src/crossref_client.py
134 строки4 KB
ivan
Add pubmed load and files for agents
16 фев 2026, 15:49
16 фев 2026, 15:49cb411ba
"""
Crossref API client for retrieving article metadata.

Provides functions to query Crossref works endpoint for:
- Journal information
- Authors
- Volume, issue, pages
- PMID (when available)
"""

import requests
import time
from typing import Dict, Optional, List, Tuple
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class CrossrefClient:
    """Client for querying Crossref API"""

    BASE_URL = "https://api.crossref.org/v1/works"

    def __init__(self, mailto: str = "demo@crossref.org", timeout: int = 30):
        self.mailto = mailto
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': f'Topics-clustering/1.0 (mailto:{mailto})'
        })

    def query_by_doi(self, doi: str) -> Optional[Dict]:
        """Query article metadata by DOI"""
        if not doi:
            return None

        try:
            params = {
                'rows': 1,
                'select': 'DOI,title,author,container-title,volume,issue,page,ISSN,published-print,published-online,member,reference,type,URL',
                'mailto': self.mailto
            }

            url = f"{self.BASE_URL}/{doi}"
            response = self.session.get(url, params=params, timeout=self.timeout)

            if response.status_code == 200:
                data = response.json()
                if 'message' in data:
                    return self._parse_crossref_item(data['message'])
        except Exception as e:
            logger.warning(f"Error querying Crossref for DOI {doi}: {e}")

        return None

    def query_by_title(self, title: str) -> Optional[Dict]:
        """Query article metadata by title"""
        if not title:
            return None

        try:
            params = {
                'query.title': title,
                'rows': 1,
                'select': 'DOI,title,author,container-title,volume,issue,page,ISSN,published-print,published-online,member,reference,type,URL',
                'mailto': self.mailto
            }

            response = self.session.get(self.BASE_URL, params=params, timeout=self.timeout)

            if response.status_code == 200:
                data = response.json()
                if 'message' in data and 'items' in data['message'] and len(data['message']['items']) > 0:
                    return self._parse_crossref_item(data['message']['items'][0])
        except Exception as e:
            logger.warning(f"Error querying Crossref for title {title}: {e}")

        return None

    def batch_query(self, dois: List[str], delay: float = 1.0) -> Dict[str, Optional[Dict]]:
        """Query multiple DOIs in batch"""
        results = {}

        for doi in dois:
            results[doi] = self.query_by_doi(doi)
            time.sleep(delay)

        return results

    def _parse_crossref_item(self, item: Dict) -> Dict:
        """Parse Crossref API response item"""
        result = {
            'doi': item.get('DOI', ''),
            'title': '',
            'journal': '',
            'authors': '',
            'volume': '',
            'issue': '',
            'pages': '',
            'year': ''
        }

        title = item.get('title', [])
        if title:
            result['title'] = title[0] if isinstance(title, list) else title

        container_title = item.get('container-title', [])
        if container_title:
            result['journal'] = container_title[0] if isinstance(container_title, list) else container_title

        authors_list = item.get('author', [])
        if authors_list:
            authors_names = []
            for author in authors_list:
                given = author.get('given', '')
                family = author.get('family', '')
                if given and family:
                    authors_names.append(f"{family} {given[0]}.")
                elif family:
                    authors_names.append(family)
            result['authors'] = ', '.join(authors_names[:10])

        result['volume'] = item.get('volume', '')
        result['issue'] = item.get('issue', '')
        result['pages'] = item.get('page', '')

        published = item.get('published-print', item.get('published-online', {}))
        if published and 'date-parts' in published:
            date_parts = published['date-parts'][0]
            if date_parts:
                result['year'] = str(date_parts[0])

        return result