Stage 2: Identification and Standardization Module.
| 27 | |
| 28 | |
| 29 | class IdentifierModule: |
| 30 | """Stage 2: Identification and Standardization Module.""" |
| 31 | |
| 32 | def __init__(self, use_google_scholar: bool = False): |
| 33 | """Initialize the identifier module. |
| 34 | |
| 35 | Args: |
| 36 | use_google_scholar: Enable Google Scholar lookups when |
| 37 | ``True``. Defaults to ``False``. |
| 38 | """ |
| 39 | self.logger = logging.getLogger(__name__) |
| 40 | self.crossref_base_url = "https://api.crossref.org/works" |
| 41 | self.use_google_scholar = use_google_scholar |
| 42 | self.github_api_base = "https://api.github.com" |
| 43 | self.zenodo_api_base = "https://zenodo.org/api/records" |
| 44 | self.base_search_url = "https://api.base-search.net/cgi-bin/BaseHttpSearchInterface.fcgi" |
| 45 | self.openaire_api_base = "https://api.openaire.eu/search/publications" |
| 46 | self.semantic_scholar_base = "https://api.semanticscholar.org/graph/v1" |
| 47 | self.pubmed_base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" |
| 48 | self.datacite_base = "https://api.datacite.org" |
| 49 | |
| 50 | |
| 51 | def identify(self, raw_entries: List[RawEntry], |
| 52 | interactive_callback: Callable[[List[Dict]], int]) -> List[IdentifiedEntry]: |
| 53 | """Identify and standardize entries, finding a DOI for each. |
| 54 | |
| 55 | Args: |
| 56 | raw_entries: List of raw entries produced by |
| 57 | :meth:`ParserModule.parse`. |
| 58 | interactive_callback: A callable that receives a list of |
| 59 | candidate dictionaries and returns the index of the |
| 60 | selected candidate. |
| 61 | |
| 62 | Returns: |
| 63 | A list of :class:`IdentifiedEntry` dictionaries. |
| 64 | """ |
| 65 | self.logger.info(f"Starting to identify {len(raw_entries)} entries") |
| 66 | identified_entries = [] |
| 67 | |
| 68 | for entry in raw_entries: |
| 69 | identified_entry = self._identify_single_entry(entry, interactive_callback) |
| 70 | identified_entries.append(identified_entry) |
| 71 | |
| 72 | successful_count = sum(1 for e in identified_entries if e['status'] == 'identified') |
| 73 | self.logger.info(f"Identification completed: {successful_count}/{len(identified_entries)} entries successfully identified") |
| 74 | |
| 75 | return identified_entries |
| 76 | |
| 77 | def _identify_single_entry(self, raw_entry: RawEntry, |
| 78 | interactive_callback: Callable[[List[Dict]], int]) -> IdentifiedEntry: |
| 79 | """Identify a single entry""" |
| 80 | identified_entry: IdentifiedEntry = { |
| 81 | 'id': raw_entry['id'], |
| 82 | 'raw_text': raw_entry['raw_text'], |
| 83 | 'doi': None, |
| 84 | 'arxiv_id': None, |
| 85 | 'url': None, |
| 86 | 'metadata': {}, |
no outgoing calls