feat: self-citations

Author: jannisborn

paperscraper/citations/entity/paper.py

@@ -84,15 +84,9 @@ def get_result(self) -> Optional[PaperResult]:
         Returns: PaperResult if available.
         """
         if not hasattr(self, "self_ref"):
-            logger.warning(
-                f"Can't get result since no referencing result for {self.input} exists. Run `.self_references` first."
-            )
-            return
-        elif not hasattr(self, "self_cite"):
-            logger.warning(
-                f"Can't get result since no citation result for {self.input} exists. Run `.self_citations` first."
-            )
-            return
+            self.self_references()
+        if not hasattr(self, "self_cite"):
+            self.self_citations()
         return PaperResult(
             title=self.title,
             **{

paperscraper/citations/entity/researcher.py

@@ -1,20 +1,19 @@
 import asyncio
 import os
 from typing import Any, List, Literal, Optional, Tuple
-from time import sleep
 
 from semanticscholar import SemanticScholar
 
 from ..orcid import orcid_to_author_name
-from ..self_citations import CitationResult
+from ..self_citations import CitationResult, self_citations_paper
 from ..self_references import ReferenceResult, self_references_paper
 from ..utils import author_name_to_ssaid, get_papers_for_author
 from .core import Entity, EntityResult
 
 
 class ResearcherResult(EntityResult):
     name: str
-    ssid: int
+    ssaid: int
     orcid: Optional[str] = None
 
     def _ordered_items(self) -> List[Tuple[str, Any]]:
@@ -27,7 +26,7 @@ def _ordered_items(self) -> List[Tuple[str, Any]]:
             ("num_citations", self.num_citations),
             ("self_references", self.self_references),
             ("self_citations", self.self_citations),
-            ("ssid", self.ssid),
+            ("ssaid", self.ssaid),
             ("orcid", self.orcid),
         ]
 
@@ -46,8 +45,9 @@ def __str__(self) -> str:
 
 class Researcher(Entity):
     name: str
-    ssid: int
+    ssaid: int
     orcid: Optional[str] = None
+    ssids: List[int] = []
 
     def __init__(self, input: str, mode: ModeType = "infer"):
         """
@@ -78,22 +78,31 @@ def __init__(self, input: str, mode: ModeType = "infer"):
                 mode = "name"
         if mode == "ssaid":
             self.name = sch.get_author(input)._name
-            self.ssid = input
+            self.ssaid = input
         elif mode == "orcid":
             orcid_name = orcid_to_author_name(input)
             self.orcid = input
-            self.ssid, self.name = author_name_to_ssaid(orcid_name)
+            self.ssaid, self.name = author_name_to_ssaid(orcid_name)
         elif mode == "name":
-            name = input
-            self.ssid, self.name = author_name_to_ssaid(input)
+            self.name = input
+            self.ssaid, self.name = author_name_to_ssaid(input)
+
+        self.result = ResearcherResult(
+            name=self.name,
+            ssaid=int(self.ssaid),
+            orcid=self.orcid,
+            num_citations=-1,
+            num_references=-1,
+        )
 
     async def _self_references_async(
         self, verbose: bool = False
     ) -> List[ReferenceResult]:
         """Async version of self_references."""
-        if self.ssid == '-1':
+        if self.ssaid == "-1":
             return []
-        self.ssids = await get_papers_for_author(self.ssid)
+        if self.ssids == []:
+            self.ssids = await get_papers_for_author(self.ssaid)
 
         results: List[ReferenceResult] = await self_references_paper(
             self.ssids, verbose=verbose
@@ -122,36 +131,91 @@ def self_references(self, verbose: bool = False) -> ResearcherResult:
         reference_results = asyncio.run(self._self_references_async(verbose=verbose))
 
         individual_self_references = {
-            getattr(result, "title"): getattr(result, "self_references").get(self.name, 0.0)
+            getattr(result, "title"): getattr(result, "self_references").get(
+                self.name, 0.0
+            )
             for result in reference_results
         }
-        reference_ratio = sum(individual_self_references.values()) / max(1, len(
-            individual_self_references
-        ))
-        return ResearcherResult(
-            name=self.name,
-            ssid=int(self.ssid),
-            orcid=self.orcid,
-            num_references=sum(r.num_references for r in reference_results),
-            num_citations=-1,
-            self_references=dict(
-                sorted(
-                    individual_self_references.items(), key=lambda x: x[1], reverse=True
-                )
-            ),
-            self_citations={},
-            self_reference_ratio=round(reference_ratio, 3),
-            self_citation_ratio=-1.0,
+        reference_ratio = sum(individual_self_references.values()) / max(
+            1, len(individual_self_references)
         )
 
-    def self_citations(self) -> ResearcherResult:
+        self.result = self.result.model_copy(
+            update={
+                "num_references": sum(r.num_references for r in reference_results),
+                "self_references": dict(
+                    sorted(
+                        individual_self_references.items(),
+                        key=lambda x: x[1],
+                        reverse=True,
+                    )
+                ),
+                "self_reference_ratio": round(reference_ratio, 3),
+            }
+        )
+
+        return self.result
+
+    async def _self_citations_async(
+        self, verbose: bool = False
+    ) -> List[CitationResult]:
+        """Async version of self_citations."""
+        if self.ssaid == "-1":
+            return []
+        if self.ssids == []:
+            self.ssids = await get_papers_for_author(self.ssaid)
+
+        results: List[CitationResult] = await self_citations_paper(
+            self.ssids, verbose=verbose
+        )
+        # Remove papers with zero references or that are erratum/corrigendum
+        results = [
+            r
+            for r in results
+            if r.num_citations > 0
+            and "erratum" not in r.title.lower()
+            and "corrigendum" not in r.title.lower()
+        ]
+
+        return results
+
+    def self_citations(self, verbose: bool = False) -> ResearcherResult:
         """
         Sifts through all papers of a researcher and finds how often they are self-cited.
         """
-        ...
+        citation_results = asyncio.run(self._self_citations_async(verbose=verbose))
+        individual_self_citations = {
+            getattr(result, "title"): getattr(result, "self_citations").get(
+                self.name, 0.0
+            )
+            for result in citation_results
+        }
+        citation_ratio = sum(individual_self_citations.values()) / max(
+            1, len(individual_self_citations)
+        )
+
+        self.result = self.result.model_copy(
+            update={
+                "num_citations": sum(r.num_citations for r in citation_results),
+                "self_citations": dict(
+                    sorted(
+                        individual_self_citations.items(),
+                        key=lambda x: x[1],
+                        reverse=True,
+                    )
+                ),
+                "self_citation_ratio": round(citation_ratio, 3),
+            }
+        )
+
+        return self.result
 
     def get_result(self) -> ResearcherResult:
         """
         Provides the result of the analysis.
         """
-        ...
+        if not hasattr(self, "self_ref"):
+            self.self_references()
+        if not hasattr(self, "self_cite"):
+            self.self_citations()
+        return self.result

paperscraper/citations/self_citations.py

@@ -24,6 +24,7 @@ class CitationResult(BaseModel):
     citation_score: float
 
 
+@retry_with_exponential_backoff(max_retries=14, base_delay=1.0)
 async def _fetch_citation_data(
     client: httpx.AsyncClient, suffix: str
 ) -> Dict[str, Any]:
@@ -120,7 +121,7 @@ async def self_citations_paper(
     if verbose:
         for res in results:
             logger.info(
-                f'Self-citations in "{res.ssid}": N={res.num_citations}, Score={res.citation_score}%'
+                f'Self-citations in "{res.title}": N={res.num_citations}, Score={res.citation_score}%'
             )
             for author, pct in res.self_citations.items():
                 logger.info(f"  {author}: {pct}%")

paperscraper/citations/tests/test_self_citations.py

@@ -1,9 +1,11 @@
 import logging
 import time
+from typing import Dict
 
 import pytest
 
 from paperscraper.citations import self_citations_paper
+from paperscraper.citations.entity import Researcher
 from paperscraper.citations.self_citations import CitationResult
 
 logging.disable(logging.INFO)
@@ -69,3 +71,77 @@ def test_multiple_dois(self, dois):
 
         for a, s in zip(result, sync_result):
             assert a == s, f"{a} vs {s}"
+
+    def test_researcher(self):
+        """
+        Tests calculation of self-references for all papers of an author.
+        """
+        ssaid = "2328976118"
+        researcher = Researcher(ssaid)
+        result = researcher.self_citations(verbose=True)
+        assert result.ssaid == int(ssaid)
+        assert isinstance(result.name, str)
+        assert result.name == "Kacper Wyrwal"
+        assert isinstance(result.num_references, int)
+        assert result.num_references == -1
+        assert isinstance(result.num_citations, int)
+        assert result.num_citations > 0
+        assert isinstance(result.self_citations, Dict)
+        for title, ratio in result.self_citations.items():
+            assert isinstance(title, str)
+            assert isinstance(ratio, float)
+            assert ratio >= 0 and ratio <= 100
+
+        assert result.self_citation_ratio >= 0 and result.self_citation_ratio <= 100
+        print(result)
+
+    def test_researcher_from_orcid(self):
+        """
+        Tests calculation of self-references for all papers of an author.
+        """
+        orcid = "0000-0003-4221-6988"
+        researcher = Researcher(orcid)
+        result = researcher.self_citations(verbose=True)
+        assert result.orcid == orcid
+        assert isinstance(result.name, str)
+        assert result.name == "Juan M. Galeazzi"
+        assert isinstance(result.num_references, int)
+        assert result.num_references == -1
+        assert isinstance(result.num_citations, int)
+        assert result.num_citations > 0
+        assert isinstance(result.self_references, Dict)
+        for title, ratio in result.self_citations.items():
+            assert isinstance(title, str)
+            assert isinstance(ratio, float)
+            assert ratio >= 0 and ratio <= 100
+
+        assert result.self_citation_ratio >= 0 and result.self_citation_ratio <= 100
+        print(result)
+
+    def test_whole_researcher(self):
+        ssaid = "2104445902"
+        researcher = Researcher(ssaid)
+        result = researcher.get_result()
+        assert result.ssaid == int(ssaid)
+        assert isinstance(result.name, str)
+        assert result.name == "Aleksandros Sobczyk"
+        assert isinstance(result.num_references, int)
+        assert result.num_references > 0
+        assert isinstance(result.num_citations, int)
+        assert result.num_citations > 0
+        assert isinstance(result.self_citations, Dict)
+        assert isinstance(result.self_references, Dict)
+        assert len(result.self_citations) > 5
+        assert len(result.self_references) >= 3
+        for title, ratio in result.self_citations.items():
+            assert isinstance(title, str)
+            assert isinstance(ratio, float)
+            assert ratio >= 0 and ratio <= 100
+        for title, ratio in result.self_references.items():
+            assert isinstance(title, str)
+            assert isinstance(ratio, float)
+            assert ratio >= 0 and ratio <= 100
+
+        assert result.self_citation_ratio >= 0 and result.self_citation_ratio <= 100
+        assert result.self_reference_ratio >= 0 and result.self_reference_ratio <= 100
+        print(result)

paperscraper/citations/tests/test_self_references.py

@@ -84,7 +84,7 @@ def test_researcher(self):
         ssaid = "2326988211"
         researcher = Researcher(ssaid)
         result = researcher.self_references(verbose=True)
-        assert result.ssid == int(ssaid)
+        assert result.ssaid == int(ssaid)
         assert isinstance(result.name, str)
         assert result.name == "Patrick Soga"
         assert isinstance(result.num_references, int)
@@ -121,4 +121,4 @@ def test_researcher_from_orcid(self):
             assert ratio >= 0 and ratio <= 100
 
         assert result.self_reference_ratio >= 0 and result.self_reference_ratio <= 100
-        print(result)
+        print(result)