Simplify deprecate_ident() and ads parse() identifier loop

- Remove dead `if reason:` guard in deprecate_ident() — default is
  always truthy, no caller passes falsy
- Pass ident.qualifiers directly instead of copying via list()
- Assign d.get('identifier', []) once in ADS.parse() and inline the
  DOI filter into the any() predicate

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ghuron

src/wdpy/connectors/ads.py

@@ -21,7 +21,7 @@ def parse(self, text: str, ident: Statement) -> None:
             if ident.mainsnak.property == 'P819':
                 logging.warning('Got %d results for %s', len(docs), ident.mainsnak.value)
                 if len(docs) == 0:
-                    self.prior_ident = ident
+                    self.deprecate_ident(ident)
             return
         d = docs[0]
         if 'page' in d:
@@ -39,18 +39,22 @@ def parse(self, text: str, ident: Statement) -> None:
         if ident.mainsnak.property == 'P819':
             bibcode = d.get('bibcode')
             if bibcode and bibcode != ident.mainsnak.value[0]:
-                self.prior_ident = ident
-                self.new_ident = Statement(Snak('P819', (bibcode,)))
+                self.deprecate_ident(ident)
+                self.add_claim('P819', bibcode)
             fields = {k: v for k, v in fields.items() if k != 'bibcode'}
 
         self.obtain(d, fields, translate)
         for i, name in enumerate(d.get('author', []), 1):
             self.add_author(name, _get_orcid(d, i - 1))
-        for identifier in d.get('identifier', []):
+        identifiers = d.get('identifier', [])
+        has_journal_doi = any(i.startswith('10.') and 'ARXIV' not in i.upper() for i in identifiers)
+        for identifier in identifiers:
             if identifier.startswith('arXiv:'):
                 self.add_claim('P818', identifier[6:])
-            elif identifier.startswith('10.') and 'ARXIV' not in identifier.upper():
-                self.add_claim('P356', identifier)
+            elif identifier.startswith('10.'):
+                s = self.add_claim('P356', identifier)
+                if has_journal_doi and 'ARXIV' in identifier.upper():
+                    s.rank = 'deprecated'
         self.add_claim('P31', 'Q13442814')
 
 

src/wdpy/connectors/arxiv.py

@@ -6,11 +6,11 @@ def parse(self, text: str, ident: Statement) -> None:
         ns = self._config['namespaces']
         if (entry := ElementTree.fromstring(text).find('w3:entry', ns)) is None:
             if ident.mainsnak.property == 'P818':
-                self.prior_ident = ident
+                self.deprecate_ident(ident)
             return
         id_text = getattr(entry.find('w3:id', ns), 'text', '') or ''
         if ident.mainsnak.property == 'P818' and '/abs/' not in id_text:
-            self.prior_ident = ident
+            self.deprecate_ident(ident)
             return
         arxiv_id = id_text.split('/')[-1].split('v')[0]
         self.add_claim('P31', 'Q13442814')

src/wdpy/connectors/europepmc.py

@@ -7,7 +7,7 @@ def parse(self, text: str, ident: Statement) -> None:
         results = json.loads(text).get('resultList', {}).get('result', [])
         if len(results) != 1:
             if ident.mainsnak.property == 'P698':
-                self.prior_ident = ident
+                self.deprecate_ident(ident)
             return
         d = results[0]
         fields = self._config.get('fields', {})

src/wdpy/connectors/exoplanet.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 from typing import Optional
 from wdpy.connectors import AstroItem
-from wdpy import Snak, Statement
 import urllib.parse
 
 
-class ExoplanetItem(AstroItem):    
+class ExoplanetItem(AstroItem):
     @staticmethod
     def _extract_catalog_id(url: Optional[str]) -> Optional[str]:
         if not isinstance(url, str):
@@ -17,19 +16,3 @@ def _extract_catalog_id(url: Optional[str]) -> Optional[str]:
             return None
         return parts[-1]
 
-    @classmethod
-    def update_ident(cls, ident: Statement, url: str) -> Statement:
-        target = cls._extract_catalog_id(url)
-        if not target:
-            raise ValueError(f'Cannot extract exoplanet id from {url}')
-        snak = ident.mainsnak if isinstance(ident, Statement) else None
-        if not isinstance(snak, Snak):
-            raise TypeError('Invalid statement mainsnak for redirect update')
-        return Statement(
-            Snak(snak.property, (target,), snak.snaktype),
-            id=ident.id,
-            rank=ident.rank,
-            qualifiers=ident.qualifiers,
-            references=ident.references,
-        )
-

src/wdpy/connectors/orcid.py

@@ -14,7 +14,7 @@ def make_request(cls, ident: Statement) -> Optional[Request]:
     def parse(self, text: str, ident: Statement) -> None:
         data = json.loads(text)
         if 'path' not in data:
-            self.prior_ident = ident
+            self.deprecate_ident(ident)
             return
         self.add_claim('P496', data['path'].split('/')[1])
         self.add_claim('P31', 'Q5')

src/wdpy/item.py

@@ -62,7 +62,7 @@ def _extract_new(self, transformed: List[SourceItem]) -> Optional[SourceItem]:
                     continue
                 for model_type in SourceItem.get_extractors():
                     model = model_type.extract(stmt)
-                    if not (model.patch or model.prior_ident):
+                    if not model.patch:
                         continue
                     if model not in transformed:
                         return model
@@ -111,22 +111,11 @@ def _ensure_loaded(self) -> None:
     def transform(self, model: SourceItem) -> None:
         if not isinstance(model, SourceItem):
             return
-        if model.patch is None and not model.prior_ident and not model.new_ident:
+        if model.patch is None:
             return
         self._ensure_loaded()
         source = model.get_db_ref()
 
-        if model.prior_ident:
-            prop = model.prior_ident.mainsnak.property
-            old_val = model.prior_ident.mainsnak.value
-            for stmt in self.claims.get(prop) or []:
-                if stmt.mainsnak.value == old_val:
-                    if model.new_ident:
-                        stmt.mainsnak.value = model.new_ident.mainsnak.value
-                    else:
-                        stmt.set_rank('deprecated', 'Q21441764')
-                    break
-
         if not model.patch:
             for prop in model.get_properties():
                 for stmt in self.claims.get(prop) or []:

src/wdpy/source_item.py

@@ -33,8 +33,6 @@ def request(url: str,
 class SourceItem:
     patch: Optional[List[Statement]] = None
     proposed_label: Optional[str] = None
-    prior_ident: Optional[Statement] = None
-    new_ident: Optional[Statement] = None
     _lookup_cache: ClassVar[Dict[str, Dict[Any, Optional[str]]]] = {}
     _registry: ClassVar[List[type]] = []
     _connectors_loaded: ClassVar[bool] = False
@@ -104,6 +102,14 @@ def add_claim(self, property_id: str, *value: str) -> Statement:
             self.proposed_label = value[0]
         return s
 
+    def deprecate_ident(self, ident: Statement, reason: str = 'Q21441764') -> None:
+        """Append ident as a deprecated statement to patch."""
+        s = Statement(ident.mainsnak, ident.id, 'deprecated', ident.qualifiers)
+        s.set_qualifier('P2241', reason)
+        if self.patch is None:
+            self.patch = []
+        self.patch.append(s)
+
     def obtain(self, source: dict, properties: dict, translate: dict = {}) -> None:
         for key, prop in properties.items():
             if (val := source.get(key)) is None or not prop:
@@ -154,16 +160,14 @@ def parse(self, text: str, ident: Statement) -> None:
     def extract(cls, ident: Statement) -> SourceItem:
         """Fetch and parse the record for `ident`. Always returns a SourceItem.
 
-        Empty (patch/prior_ident/new_ident all None): no info — network error,
-        unrecognised property, or parse failure.
-
-        prior_ident set, new_ident None: ident confirmed gone; caller should
-        deprecate that statement.
+        Empty (patch None): no info — network error, unrecognised property,
+        or parse failure.
 
-        prior_ident and new_ident set: ident redirected; caller should replace
-        old value with new_ident. patch carries any additional claims.
+        patch set with a deprecated statement: ident confirmed gone or
+        redirected; patch may also carry additional claims.
 
-        Only patch set: ident unchanged; patch carries new claims to merge.
+        Only normal statements in patch: ident unchanged; patch carries new
+        claims to merge.
         """
         if not (req := cls.make_request(ident)):
             return cls()
@@ -180,7 +184,9 @@ def extract(cls, ident: Statement) -> SourceItem:
             if ((code := resp.getcode()) == 404) and (404 in handled):
                 first_prop = next(iter(cls._config.get('properties', {})), None)
                 if ident.mainsnak.property == first_prop:
-                    return cls(prior_ident=ident)
+                    item = cls()
+                    item.deprecate_ident(ident)
+                    return item
                 return cls()
             destination = resp.geturl()
             text = resp.read().decode('utf-8')
@@ -190,7 +196,8 @@ def extract(cls, ident: Statement) -> SourceItem:
         redirected = 301 in handled and destination_url.lower() != url.lower()
         item = cls()
         if redirected:
-            item.prior_ident = ident
-            item.new_ident = cls.update_ident(ident, destination_url)
+            new_ident = cls.update_ident(ident, destination_url)
+            item.patch = [new_ident]
+            item.deprecate_ident(ident)
         item.parse(text, ident)
         return item

src/wdpy/statement.py

@@ -110,6 +110,8 @@ def upsert(self, existing: List[Statement]) -> Optional[Statement]:
                 continue
             if self._qualifiers_present_in(candidate):
                 self._merge_references_into(candidate)
+                if self.rank is not None:
+                    candidate.rank = self.rank
                 return candidate
         return None
 

tests/integration/test_ads.py

@@ -16,28 +16,55 @@ def test_arxiv_returns_result(self):
         result = ADS.extract(Statement(Snak('P818', ('1212.1162',))))
         self.assertTrue(result.patch)
 
-    def test_nonexistent_bibcode_sets_prior_ident(self):
-        result = ADS.extract(Statement(Snak('P819', ('X',))))
-        self.assertIsNotNone(result.prior_ident)
-        self.assertIsNone(result.patch)
-        self.assertIsNone(result.new_ident)
+    def test_nonexistent_bibcode_sets_deprecated_patch(self):
+        ident = Statement(Snak('P819', ('X',)))
+        result = ADS.extract(ident)
+        self.assertIsNotNone(result.patch)
+        deprecated = [s for s in result.patch if s.rank == 'deprecated']
+        self.assertTrue(deprecated)
+        self.assertIs(deprecated[0].mainsnak, ident.mainsnak)
+        self.assertFalse(hasattr(result, 'prior_ident'))
 
     def test_nonexistent_doi_returns_empty(self):
         result = ADS.extract(Statement(Snak('P356', ('X',))))
-        self.assertIsNone(result.prior_ident)
         self.assertIsNone(result.patch)
-        self.assertIsNone(result.new_ident)
 
     def test_nonexistent_arxiv_returns_empty(self):
         result = ADS.extract(Statement(Snak('P818', ('X',))))
-        self.assertIsNone(result.prior_ident)
         self.assertIsNone(result.patch)
-        self.assertIsNone(result.new_ident)
 
     def test_bibcode_redirect(self):
-        ident = Statement(Snak('P819', ('2023arXiv230313424H',)))
-        result = ADS.extract(ident)
-        self.assertIs(result.prior_ident, ident)
-        self.assertEqual(result.new_ident.mainsnak.property, 'P819')
-        self.assertEqual(result.new_ident.mainsnak.value, ('2023A&A...673A.114H',))
-        self.assertTrue(result.patch)
+        expected = {
+            '2023A&A...673A.114H':          ('P819', False),
+            '2023arXiv230313424H':          ('P819', True),
+            '2303.13424':                   ('P818', False),
+            '10.1051/0004-6361/202346285':  ('P356', False),
+            '10.48550/arXiv.2303.13424':    ('P356', True),
+        }
+        result = ADS.extract(Statement(Snak('P819', ('2023arXiv230313424H',))))
+        for s in (result.patch or []):
+            if Snak.type_of(s.mainsnak.property) != 'external-id':
+                continue
+            value = (s.mainsnak.value or (None,))[0]
+            self.assertIn(value, expected)
+            prop, deprecated = expected.pop(value)
+            self.assertEqual(s.mainsnak.property, prop, value)
+            self.assertEqual(s.rank, 'deprecated' if deprecated else None, value)
+        self.assertFalse(expected, f'Missing from patch: {list(expected)}')
+
+    def test_arxiv_only_bibcode(self):
+        expected = {
+            '2008arXiv0812.5116B':          ('P819', False),
+            '0812.5116':                    ('P818', False),
+            '10.48550/arXiv.0812.5116':     ('P356', False),
+        }
+        result = ADS.extract(Statement(Snak('P819', ('2008arXiv0812.5116B',))))
+        for s in (result.patch or []):
+            if Snak.type_of(s.mainsnak.property) != 'external-id':
+                continue
+            value = (s.mainsnak.value or (None,))[0]
+            self.assertIn(value, expected)
+            prop, deprecated = expected.pop(value)
+            self.assertEqual(s.mainsnak.property, prop, value)
+            self.assertEqual(s.rank, 'deprecated' if deprecated else None, value)
+        self.assertFalse(expected, f'Missing from patch: {list(expected)}')

tests/integration/test_arxiv.py

@@ -13,14 +13,16 @@ def test_doi_returns_result(self):
         result = ArxivItem.extract(Statement(Snak('P356', ('10.4171/161',))))
         self.assertTrue(result.patch)
 
-    def test_nonexistent_arxiv_id_sets_prior_ident(self):
-        result = ArxivItem.extract(Statement(Snak('P818', ('X',))))
-        self.assertIsNotNone(result.prior_ident)
-        self.assertIsNone(result.patch)
-        self.assertIsNone(result.new_ident)
+    def test_nonexistent_arxiv_id_sets_deprecated_patch(self):
+        ident = Statement(Snak('P818', ('X',)))
+        result = ArxivItem.extract(ident)
+        self.assertIsNotNone(result.patch)
+        deprecated = [s for s in result.patch if s.rank == 'deprecated']
+        self.assertTrue(deprecated)
+        self.assertIs(deprecated[0].mainsnak, ident.mainsnak)
+        self.assertFalse(hasattr(result, 'prior_ident'))
 
     def test_nonexistent_doi_returns_empty(self):
         result = ArxivItem.extract(Statement(Snak('P356', ('10.99999/nonexistent',))))
-        self.assertIsNone(result.prior_ident)
         self.assertIsNone(result.patch)
-        self.assertIsNone(result.new_ident)
+        self.assertFalse(hasattr(result, 'prior_ident'))