Connect existing articles to author (#51)

src/pyorcidator/helper.py

@@ -6,7 +6,7 @@
 import json
 import logging
 import re
-from typing import List, Mapping, Optional, Tuple, Union
+from typing import Dict, List, Mapping, Optional, Tuple, Union
 
 import requests
 from wdcuration import add_key
@@ -114,6 +114,14 @@ def get_orcid_quickstatements(orcid: str) -> List[Line]:
         )
     )
 
+    papers_data = data["activities-summary"]["works"]["group"]
+    papers_entries = get_paper_dois(papers_data)
+    lines.extend(
+        process_paper_entries(
+            orcid=orcid, researcher_qid=researcher_qid, paper_dois=papers_entries, property_id="P50"
+        )
+    )
+
     external_ids = get_external_ids(data)
     for key, value in external_ids.items():
         predicate = EXTERNAL_ID_PROPERTIES.get(key)
@@ -355,12 +363,52 @@ def process_affiliation_entries(
     return rv
 
 
-def get_paper_dois(group_of_works_from_orcid):
-    """ """
+def get_paper_dois(group_of_works_from_orcid: List[Dict]) -> List[str]:
+    """Return list of DOIs (strings) from an ORCID API result"""
     dois = []
     for work in group_of_works_from_orcid:
         for external_id in work["external-ids"]["external-id"]:
             if external_id["external-id-type"] == "doi":
                 dois.append(external_id["external-id-value"])
     logger.info("got paper DOIs: %s", dois)
     return dois
+
+
+def get_paper_qids(papers_dois: List[str]) -> List[str]:
+    """Get QIDs for list of DOIs"""
+
+    doi_values = " ".join(f"'{doi.upper()}'" for doi in papers_dois)
+
+    query = f"""\
+        SELECT ?item
+        WHERE
+        {{
+            VALUES ?doi {{{doi_values}}}
+            ?itemURL wdt:P356 ?doi .
+            BIND(REPLACE(STR(?itemURL), "http://www.wikidata.org/entity/", "") AS ?item)
+        }}
+    """
+
+    bindings = query_wikidata(query)
+    if len(bindings) > 0:
+        return [binding["item"]["value"] for binding in bindings]
+
+
+def process_paper_entries(
+    orcid: str, researcher_qid: str, paper_dois: List[str], property_id: str
+) -> List[EntityLine]:
+    """From a list of paper DOIs create statements for linking them to author"""
+
+    paper_qids = get_paper_qids(paper_dois)
+
+    paper_statements = []
+
+    qualifiers = [_get_orcid_qualifier(orcid)]
+    for paper in paper_qids:
+        entry = EntityLine(
+            subject=paper, predicate=property_id, target=researcher_qid, qualifiers=qualifiers
+        )
+
+        paper_statements.append(entry)
+
+    return paper_statements

tests/test_helper.py

@@ -10,6 +10,7 @@
     get_paper_dois,
     lookup_id,
     process_keyword_entries,
+    process_paper_entries,
     render_orcid_qs,
 )
 from pyorcidator.quickstatements import prepare_date
@@ -39,6 +40,24 @@ def test_get_paper_dois(sample_orcid_data):
     assert "10.3233/jad-201397" in test_dois
 
 
+def test_process_paper_entries(sample_orcid_data):
+
+    test_papers = sample_orcid_data["activities-summary"]["works"]["group"]
+    test_dois = get_paper_dois(test_papers)
+
+    entries = process_paper_entries(
+        orcid="0000-0003-4423-4370",
+        researcher_qid="Q47475003",
+        paper_dois=test_dois,
+        property_id="P50",
+    )
+
+    entries_qids = [entry.subject for entry in entries]
+
+    assert len(entries) == 31
+    assert entries_qids[1:3] == ["Q63709723", "Q82511885"]
+
+
 def test_get_org_list(sample_orcid_data):
     employment_data = sample_orcid_data["activities-summary"]["employments"]["employment-summary"]