Add select_outdated and rank_by_recency to Statement

Port delete_all_but_one and deprecate_all_but_one from legacy Claim,
with improved naming and unit tests.

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

Author: ghuron

src/wdpy/statement.py

@@ -174,4 +174,85 @@ def _display_name(s: Statement) -> str:
 
     def save(self, summary: str) -> Optional[Dict[str, Any]]:
         """Persist statement via wbsetclaim."""
-        return api_write('wbsetclaim', claim=self.json(), summary=summary)
+        return api_write('wbsetclaim', claim=self.json(), summary=summary)
+
+    @staticmethod
+    def select_outdated(statements: List[Statement],
+                        group_by: Optional[str] = None) -> List[Statement]:
+        """Return statements to delete, keeping only the most recently sourced per group.
+
+        Statements are grouped by the value of the group_by qualifier (or treated as
+        one group when group_by is None).  Within each group the keeper is the
+        novalue/somevalue statement if one exists, otherwise the statement with the
+        highest reference publication date.  Statements lacking the group_by qualifier
+        are always returned for deletion.
+        """
+        _MISSING = object()
+
+        def _group(s: Statement):
+            if not group_by:
+                return None
+            for q in (s.qualifiers or []):
+                if q.property == group_by and q.value:
+                    return q.value[0]
+            return _MISSING
+
+        # First pass: find the best ref date per group (None = novalue/somevalue wins).
+        best: Dict[Any, Optional[str]] = {}
+        for s in statements:
+            if (g := _group(s)) is _MISSING:
+                continue
+            if s.mainsnak.snaktype != 'value':
+                best[g] = None
+            elif g not in best:
+                best[g] = (s.references and s.references.publication_date) or '00000000'
+            elif best[g] is not None:
+                d = (s.references and s.references.publication_date) or '00000000'
+                if d > best[g]:
+                    best[g] = d
+
+        # Second pass: keep exactly one per group, queue the rest for deletion.
+        kept: set = set()
+        to_delete: List[Statement] = []
+        for s in statements:
+            if (g := _group(s)) is _MISSING:
+                to_delete.append(s)
+                continue
+            if best.get(g) is None:          # novalue group
+                if s.mainsnak.snaktype == 'value' or g in kept:
+                    to_delete.append(s)
+                else:
+                    kept.add(g)
+            elif g in kept or ((s.references and s.references.publication_date) or '00000000') < best[g]:
+                to_delete.append(s)
+            else:
+                kept.add(g)
+        return to_delete
+
+    @staticmethod
+    def rank_by_recency(statements: List[Statement]) -> None:
+        """Assign deprecated rank to all but the most recently sourced statement.
+
+        Statements carrying a P2241 (reason for deprecation) qualifier are left
+        unchanged.  If any statement already has preferred rank, nothing is modified.
+        The one statement with the highest reference publication date keeps normal
+        rank; all others receive deprecated rank.
+        """
+        for s in statements:
+            if s.rank == 'preferred':
+                return
+
+        best_date = '00000000'
+        for s in statements:
+            if not any(q.property == 'P2241' for q in (s.qualifiers or [])):
+                if (d := (s.references and s.references.publication_date) or '00000000') > best_date:
+                    best_date = d
+
+        remaining = 1
+        for s in statements:
+            if not any(q.property == 'P2241' for q in (s.qualifiers or [])):
+                if remaining > 0 and ((s.references and s.references.publication_date) or '00000000') == best_date:
+                    s.rank = 'normal'
+                    remaining -= 1
+                else:
+                    s.rank = 'deprecated'

tests/test_statement.py

@@ -1,6 +1,7 @@
 import json
 from unittest import TestCase, mock
 from wdpy import Snak, Statement, References
+from wdpy.references import _PUB_DATES, _REDIRECTS
 
 
 # ── helpers ──────────────────────────────────────────────────────────────────
@@ -292,3 +293,178 @@ def test_mixed_ordinals_only_duplicates_deleted(self):
         b2 = p50('Q99', '2')
         result = self.dedup(a1, a1_dup, b2)
         self.assertEqual(result, [a1_dup])
+
+
+# ── helpers for date-based tests ──────────────────────────────────────────────
+
+def _dated_refs(date: int) -> References:
+    """References with a single P248 snak whose publication date is `date`."""
+    qid = f'Q{date}'
+    _PUB_DATES[qid] = date
+    return References([{'snaks': {'P248': [Snak('P248', (qid,))]}}])
+
+
+class SelectOutdated(TestCase):
+    GROUP = 'P1545'
+
+    def setUp(self):
+        _PUB_DATES.clear()
+        _REDIRECTS.clear()
+
+    def _s(self, val='Q1', date=None):
+        refs = _dated_refs(date) if date is not None else None
+        return Statement(Snak('P31', (val,)), references=refs)
+
+    def _sg(self, val='Q1', group='1', date=None):
+        s = self._s(val, date)
+        s.qualifiers = [Snak(self.GROUP, (group,))]
+        return s
+
+    def _nv(self, group=None):
+        quals = [Snak(self.GROUP, (group,))] if group else None
+        return Statement(Snak('P31', None, 'novalue'), qualifiers=quals)
+
+    # ── no group_by ───────────────────────────────────────────────────────────
+
+    def test_empty(self):
+        self.assertEqual(Statement.select_outdated([]), [])
+
+    def test_single(self):
+        self.assertEqual(Statement.select_outdated([self._s()]), [])
+
+    def test_older_deleted(self):
+        old = self._s('Q1', 20200101)
+        new = self._s('Q2', 20220101)
+        self.assertEqual(Statement.select_outdated([new, old]), [old])
+
+    def test_older_first_still_deleted(self):
+        old = self._s('Q1', 20200101)
+        new = self._s('Q2', 20220101)
+        self.assertEqual(Statement.select_outdated([old, new]), [old])
+
+    def test_no_refs_deleted_when_other_is_dated(self):
+        no_date = self._s('Q1')
+        dated = self._s('Q2', 20220101)
+        self.assertEqual(Statement.select_outdated([no_date, dated]), [no_date])
+
+    def test_equal_dates_first_kept_second_deleted(self):
+        a = self._s('Q1', 20220101)
+        b = self._s('Q2', 20220101)
+        self.assertEqual(Statement.select_outdated([a, b]), [b])
+
+    def test_novalue_beats_value(self):
+        val = self._s('Q1', 20220101)
+        nv = self._nv()
+        result = Statement.select_outdated([val, nv])
+        self.assertIn(val, result)
+        self.assertNotIn(nv, result)
+
+    def test_two_novalue_second_deleted(self):
+        nv1, nv2 = self._nv(), self._nv()
+        self.assertEqual(Statement.select_outdated([nv1, nv2]), [nv2])
+
+    # ── with group_by ─────────────────────────────────────────────────────────
+
+    def test_missing_qualifier_always_deleted(self):
+        s = self._s()   # no P1545
+        self.assertEqual(Statement.select_outdated([s], group_by=self.GROUP), [s])
+
+    def test_groups_managed_independently(self):
+        a_old = self._sg('Q1', '1', 20200101)
+        a_new = self._sg('Q2', '1', 20220101)
+        b_old = self._sg('Q3', '2', 20190101)
+        b_new = self._sg('Q4', '2', 20210101)
+        result = Statement.select_outdated([a_old, a_new, b_old, b_new],
+                                           group_by=self.GROUP)
+        self.assertIn(a_old, result)
+        self.assertIn(b_old, result)
+        self.assertNotIn(a_new, result)
+        self.assertNotIn(b_new, result)
+
+    def test_unqualified_deleted_alongside_group_outdated(self):
+        no_qual = self._s()
+        old = self._sg('Q1', '1', 20200101)
+        new = self._sg('Q2', '1', 20220101)
+        result = Statement.select_outdated([no_qual, old, new], group_by=self.GROUP)
+        self.assertIn(no_qual, result)
+        self.assertIn(old, result)
+        self.assertNotIn(new, result)
+
+    def test_novalue_in_group_beats_value(self):
+        val = self._sg('Q1', '1', 20220101)
+        nv = self._nv(group='1')
+        result = Statement.select_outdated([val, nv], group_by=self.GROUP)
+        self.assertIn(val, result)
+        self.assertNotIn(nv, result)
+
+
+class RankByRecency(TestCase):
+
+    def setUp(self):
+        _PUB_DATES.clear()
+        _REDIRECTS.clear()
+
+    def _s(self, date=None, *, rank=None, p2241=False):
+        refs = _dated_refs(date) if date is not None else None
+        quals = [Snak('P2241', ('Q1',))] if p2241 else None
+        return Statement(Snak('P31', ('Q1',)), rank=rank, qualifiers=quals, references=refs)
+
+    # ── no-op cases ───────────────────────────────────────────────────────────
+
+    def test_empty(self):
+        Statement.rank_by_recency([])   # must not raise
+
+    def test_preferred_halts_all_changes(self):
+        preferred = self._s(20220101, rank='preferred')
+        old = self._s(20200101)
+        Statement.rank_by_recency([old, preferred])
+        self.assertIsNone(old.rank)
+
+    def test_p2241_statement_not_modified(self):
+        s = self._s(p2241=True)
+        Statement.rank_by_recency([s])
+        self.assertIsNone(s.rank)
+
+    # ── rank assignment ───────────────────────────────────────────────────────
+
+    def test_single_no_date_becomes_normal(self):
+        s = self._s()
+        Statement.rank_by_recency([s])
+        self.assertEqual(s.rank, 'normal')
+
+    def test_newer_normal_older_deprecated(self):
+        new = self._s(20220101)
+        old = self._s(20200101)
+        Statement.rank_by_recency([new, old])
+        self.assertEqual(new.rank, 'normal')
+        self.assertEqual(old.rank, 'deprecated')
+
+    def test_order_independent(self):
+        new = self._s(20220101)
+        old = self._s(20200101)
+        Statement.rank_by_recency([old, new])
+        self.assertEqual(new.rank, 'normal')
+        self.assertEqual(old.rank, 'deprecated')
+
+    def test_no_refs_deprecated_when_other_is_dated(self):
+        dated = self._s(20220101)
+        undated = self._s()
+        Statement.rank_by_recency([dated, undated])
+        self.assertEqual(dated.rank, 'normal')
+        self.assertEqual(undated.rank, 'deprecated')
+
+    def test_equal_dates_first_normal_second_deprecated(self):
+        a = self._s(20220101)
+        b = self._s(20220101)
+        Statement.rank_by_recency([a, b])
+        self.assertEqual(a.rank, 'normal')
+        self.assertEqual(b.rank, 'deprecated')
+
+    def test_p2241_excluded_others_ranked(self):
+        with_reason = self._s(20230101, p2241=True)
+        new = self._s(20220101)
+        old = self._s(20200101)
+        Statement.rank_by_recency([with_reason, new, old])
+        self.assertIsNone(with_reason.rank)
+        self.assertEqual(new.rank, 'normal')
+        self.assertEqual(old.rank, 'deprecated')