Emit CID aliases from parser

Squash merge PR #50 after review feedback fixes and passing checks.

Author: michaelmwu

src/gmaps_scraper/models.py

@@ -82,6 +82,7 @@ class Place:
     maps_url: str
     cid: str | None = None
     google_id: str | None = None
+    cid_aliases: list[str] = field(default_factory=list)
     is_favorite: bool = False
     added_by: ListOwner | None = None
 
@@ -100,6 +101,8 @@ def to_dict(self) -> dict[str, object]:
             del result["note"]
         if self.cid is not None:
             result["cid"] = self.cid
+        if self.cid_aliases:
+            result["cid_aliases"] = self.cid_aliases
         if self.google_id is not None:
             result["google_id"] = self.google_id
         if self.added_by is not None:

src/gmaps_scraper/parser.py

@@ -71,6 +71,12 @@ class _Candidate:
     signal_score: int
 
 
+@dataclass(frozen=True, slots=True)
+class _PlaceDedupeKeys:
+    primary: set[str]
+    aliases: set[str]
+
+
 class ParseError(RuntimeError):
     """Raised when a saved list cannot be parsed from the supplied artifacts."""
 
@@ -388,7 +394,8 @@ def _parse_list_owner(node: JSONValue | None) -> ListOwner | None:
 
 def _extract_places(node: JSONValue) -> list[Place]:
     places: list[Place] = []
-    seen: set[str] = set()
+    primary_key_indexes: dict[str, int] = {}
+    alias_key_indexes: dict[str, int] = {}
 
     for current, ancestors in _walk_json(node):
         if not _is_coordinate_tuple(current):
@@ -405,6 +412,7 @@ def _extract_places(node: JSONValue) -> list[Place]:
             metadata_node = _find_place_metadata(ancestors)
         address = _extract_address(metadata_node)
         cid = _find_cid(metadata_node)
+        cid_aliases = _find_cid_aliases(metadata_node)
         google_id = _find_google_id(metadata_node)
         name = _find_place_name(ancestors, address=address, place_record=place_record)
         note = _find_place_note(place_record, name=name, address=address)
@@ -422,23 +430,95 @@ def _extract_places(node: JSONValue) -> list[Place]:
                 lng=lng,
             ),
             cid=cid,
+            cid_aliases=cid_aliases,
             google_id=google_id,
             is_favorite=is_favorite,
             added_by=_find_place_added_by(place_record),
         )
-        dedupe_key = (
-            google_id
-            or (f"{cid}:{lat:.6f}:{lng:.6f}" if cid is not None else None)
-            or f"{place.name}:{lat:.6f}:{lng:.6f}"
+        dedupe_keys = _place_dedupe_keys(place)
+        duplicate_index = _duplicate_place_index(
+            dedupe_keys,
+            primary_key_indexes=primary_key_indexes,
+            alias_key_indexes=alias_key_indexes,
         )
-        if dedupe_key in seen:
+        if duplicate_index is not None:
+            if _place_dedupe_quality(place) > _place_dedupe_quality(places[duplicate_index]):
+                places[duplicate_index] = place
+                _record_place_dedupe_keys(
+                    duplicate_index,
+                    dedupe_keys,
+                    primary_key_indexes=primary_key_indexes,
+                    alias_key_indexes=alias_key_indexes,
+                )
             continue
-        seen.add(dedupe_key)
         places.append(place)
+        _record_place_dedupe_keys(
+            len(places) - 1,
+            dedupe_keys,
+            primary_key_indexes=primary_key_indexes,
+            alias_key_indexes=alias_key_indexes,
+        )
 
     return places
 
 
+def _place_dedupe_keys(place: Place) -> _PlaceDedupeKeys:
+    primary: set[str] = set()
+    aliases: set[str] = set()
+    if place.google_id:
+        primary.add(f"gid:{place.google_id}")
+    if place.cid is not None:
+        cid_suffix = f"{place.lat:.6f}:{place.lng:.6f}"
+        primary.add(f"cid:{place.cid}:{cid_suffix}")
+        aliases.update(
+            f"cid:{cid_alias}:{cid_suffix}"
+            for cid_alias in place.cid_aliases
+        )
+    if not primary:
+        primary.add(f"name:{place.name}:{place.lat:.6f}:{place.lng:.6f}")
+    return _PlaceDedupeKeys(primary=primary, aliases=aliases)
+
+
+def _duplicate_place_index(
+    keys: _PlaceDedupeKeys,
+    *,
+    primary_key_indexes: dict[str, int],
+    alias_key_indexes: dict[str, int],
+) -> int | None:
+    for key in keys.primary:
+        if key in primary_key_indexes:
+            return primary_key_indexes[key]
+    for key in keys.aliases:
+        if key in primary_key_indexes:
+            return primary_key_indexes[key]
+    for key in keys.primary:
+        if key in alias_key_indexes:
+            return alias_key_indexes[key]
+    return None
+
+
+def _record_place_dedupe_keys(
+    index: int,
+    keys: _PlaceDedupeKeys,
+    *,
+    primary_key_indexes: dict[str, int],
+    alias_key_indexes: dict[str, int],
+) -> None:
+    for key in keys.primary:
+        primary_key_indexes[key] = index
+    for key in keys.aliases:
+        alias_key_indexes[key] = index
+
+
+def _place_dedupe_quality(place: Place) -> tuple[bool, bool, bool, bool]:
+    return (
+        place.google_id is not None,
+        bool(place.cid_aliases),
+        place.cid is not None,
+        place.address is not None,
+    )
+
+
 def _find_place_metadata(ancestors: Sequence[JSONValue]) -> list[JSONValue] | None:
     place_record = _find_place_record(ancestors)
     metadata_node = _place_metadata_from_record(place_record)
@@ -597,6 +677,19 @@ def _find_cid(node: list[JSONValue] | None) -> str | None:
     return None
 
 
+def _find_cid_aliases(node: list[JSONValue] | None) -> list[str]:
+    if node is None:
+        return []
+    structured_value = _safe_index(node, 6)
+    structured_cid = _find_cid_in_structured_slot(structured_value)
+    if structured_cid is None:
+        return []
+    return _find_cid_aliases_in_structured_slot(
+        structured_value,
+        selected_cid=structured_cid,
+    )
+
+
 def _find_google_id(node: list[JSONValue] | None) -> str | None:
     if node is None:
         return None
@@ -660,6 +753,27 @@ def _find_cid_in_structured_slot(value: JSONValue | None) -> str | None:
     return _normalize_cid_token(numeric_texts[0])
 
 
+def _find_cid_aliases_in_structured_slot(
+    value: JSONValue | None,
+    *,
+    selected_cid: str,
+) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    owner = _parse_list_owner(value)
+    if owner is not None and (owner.photo_url is not None or owner.profile_id is not None):
+        return []
+    aliases: list[str] = []
+    for item in value:
+        text = _clean_text(item)
+        if text is None or _LONG_INTEGER_PATTERN.fullmatch(text) is None:
+            continue
+        alias = _normalize_cid_token(text)
+        if alias is not None and alias != selected_cid and alias not in aliases:
+            aliases.append(alias)
+    return aliases
+
+
 def _find_cid_in_fallback_value(value: JSONValue | None) -> str | None:
     if isinstance(value, int | str):
         return _normalize_fallback_cid_token(value)

tests/test_parser.py

@@ -255,6 +255,7 @@ def test_uses_fingerprint_not_positive_s2_cell_from_slot_6(self) -> None:
         parsed = parse_saved_list_artifacts(_LIST_URL, runtime_state=runtime_state)
 
         self.assertEqual(parsed.places[0].cid, _NORTHWIND_CID)
+        self.assertEqual(parsed.places[0].cid_aliases, [_NORTHWIND_CELL_ID])
         self.assertNotEqual(parsed.places[0].cid, _NORTHWIND_CELL_ID)
 
     def test_normalizes_negative_slot_6_fingerprint(self) -> None:
@@ -332,6 +333,64 @@ def test_dedupes_places_with_same_cid(self) -> None:
         self.assertEqual(parsed.places[0].cid, _NORTHWIND_CID)
         self.assertEqual(parsed.places[1].cid, _HARBOR_CID)
 
+    def test_dedupes_places_with_cid_alias(self) -> None:
+        runtime_state = copy.deepcopy(["noise", _LIST_NODE])
+        original_metadata = runtime_state[1][8][0][1]
+        assert isinstance(original_metadata, list)
+        original_metadata[7] = None
+        duplicate_place = copy.deepcopy(runtime_state[1][8][0])
+        duplicate_metadata = duplicate_place[1]
+        assert isinstance(duplicate_metadata, list)
+        duplicate_metadata[6] = [_NORTHWIND_CELL_ID]
+        duplicate_metadata[7] = None
+        runtime_state[1][8].append(duplicate_place)
+
+        parsed = parse_saved_list_artifacts(_LIST_URL, runtime_state=runtime_state)
+
+        self.assertEqual(len(parsed.places), 2)
+        self.assertEqual(parsed.places[0].cid, _NORTHWIND_CID)
+        self.assertEqual(parsed.places[1].cid, _HARBOR_CID)
+
+    def test_prefers_canonical_place_when_cid_alias_duplicate_appears_first(self) -> None:
+        runtime_state = copy.deepcopy(["noise", _LIST_NODE])
+        duplicate_place = copy.deepcopy(runtime_state[1][8][0])
+        duplicate_metadata = duplicate_place[1]
+        assert isinstance(duplicate_metadata, list)
+        duplicate_metadata[6] = [_NORTHWIND_CELL_ID]
+        duplicate_metadata[7] = None
+        runtime_state[1][8].insert(0, duplicate_place)
+
+        parsed = parse_saved_list_artifacts(_LIST_URL, runtime_state=runtime_state)
+
+        self.assertEqual(len(parsed.places), 2)
+        self.assertEqual(parsed.places[0].cid, _NORTHWIND_CID)
+        self.assertEqual(parsed.places[0].cid_aliases, [_NORTHWIND_CELL_ID])
+        self.assertEqual(parsed.places[0].google_id, "/g/11northwind")
+        self.assertEqual(parsed.places[1].cid, _HARBOR_CID)
+
+    def test_keeps_distinct_places_with_same_cid_alias_and_coordinates(self) -> None:
+        runtime_state = copy.deepcopy(["noise", _LIST_NODE])
+        first_place = runtime_state[1][8][0]
+        second_place = runtime_state[1][8][1]
+        assert isinstance(first_place, list)
+        assert isinstance(second_place, list)
+        first_metadata = first_place[1]
+        second_metadata = second_place[1]
+        assert isinstance(first_metadata, list)
+        assert isinstance(second_metadata, list)
+        first_metadata[7] = None
+        second_metadata[5] = [None, None, 35.6501307, 139.6868459]
+        second_metadata[6] = [_NORTHWIND_CELL_ID, _HARBOR_CID]
+        second_metadata[7] = None
+
+        parsed = parse_saved_list_artifacts(_LIST_URL, runtime_state=runtime_state)
+
+        self.assertEqual(len(parsed.places), 2)
+        self.assertEqual(parsed.places[0].cid, _NORTHWIND_CID)
+        self.assertEqual(parsed.places[1].cid, _HARBOR_CID)
+        self.assertEqual(parsed.places[0].cid_aliases, [_NORTHWIND_CELL_ID])
+        self.assertEqual(parsed.places[1].cid_aliases, [_NORTHWIND_CELL_ID])
+
     def test_keeps_distinct_places_that_share_a_cid(self) -> None:
         runtime_state = copy.deepcopy(["noise", _LIST_NODE])
         second_place = runtime_state[1][8][1]