Skip invalid ownerless JSON rows (#1540)

scripts/claim_ownerless.py

@@ -13,6 +13,18 @@ import json
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
+
+def claim_json_entries(entries, owner):
+    count = 0
+    for entry in entries:
+        if not isinstance(entry, dict):
+            continue
+        if not entry.get("owner"):
+            entry["owner"] = owner
+            count += 1
+    return count
+
+
 def main():
     if len(sys.argv) < 2:
         print("Usage: python scripts/claim_ownerless.py <username>")
@@ -31,11 +43,7 @@ def main():
             continue
         with open(path, "r", encoding="utf-8") as f:
             entries = json.load(f)
-        count = 0
-        for e in entries:
-            if not e.get("owner"):
-                e["owner"] = owner
-                count += 1
+        count = claim_json_entries(entries, owner)
         if count:
             with open(path, "w", encoding="utf-8") as f:
                 json.dump(entries, f, ensure_ascii=False, indent=2)

tests/test_claim_ownerless_json.py

@@ -0,0 +1,18 @@
+from scripts.claim_ownerless import claim_json_entries
+
+
+def test_claim_json_entries_skips_invalid_rows():
+    rows = [
+        {"id": "a"},
+        "bad-row",
+        None,
+        {"id": "b", "owner": "already"},
+    ]
+
+    assert claim_json_entries(rows, "admin") == 1
+    assert rows == [
+        {"id": "a", "owner": "admin"},
+        "bad-row",
+        None,
+        {"id": "b", "owner": "already"},
+    ]