Emit CID aliases from parser

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

@@ -405,6 +405,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,18 +423,24 @@ 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}"
-        )
-        if dedupe_key in seen:
+        dedupe_keys: set[str] = set()
+        if google_id:
+            dedupe_keys.add(f"gid:{google_id}")
+        if cid is not None:
+            dedupe_keys.update({
+                f"cid:{cid_value}:{lat:.6f}:{lng:.6f}"
+                for cid_value in [cid, *cid_aliases]
+            })
+        if not dedupe_keys:
+            dedupe_keys = {f"name:{place.name}:{lat:.6f}:{lng:.6f}"}
+        if seen.intersection(dedupe_keys):
             continue
-        seen.add(dedupe_key)
+        seen.update(dedupe_keys)
         places.append(place)
 
     return places
@@ -597,6 +604,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 +680,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,21 @@ 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])
+        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_keeps_distinct_places_that_share_a_cid(self) -> None:
         runtime_state = copy.deepcopy(["noise", _LIST_NODE])
         second_place = runtime_state[1][8][1]