fix: select popular subview across phrase batches

src/wordstat/collector.py

@@ -48,18 +48,44 @@
 SEARCH_SELECTOR = ".wordstat__search-button"
 DOWNLOAD_SELECTOR = "button.save-button"
 DOWNLOAD_CSV_MENU_ITEM_SELECTOR = "a[download]:has(button.save-csv-button)"
+TABLE_VIEW_SELECTOR = "label[for='table']"
 GRANULARITY_SELECTOR = ".wordstat__content-type_select > button"
 DATE_RANGE_SELECTOR = ".range-datepicker__selected-dates > button"
 REGION_BUTTON_SELECTOR = ".settings__selected button"
 TABLE_ROW_SELECTOR = ".table__wrapper tbody tr"
 # Tab selectors live here with the rest of the DOM knowledge; the markup is
 # inconsistent enough (id- vs for-based) that it is worth having in one place.
 VIEW_SELECTORS = {
-    WordstatView.TOP_POPULAR: "label[for='table']",
+    WordstatView.TOP_POPULAR: "label:has(#popular)",
     WordstatView.TOP_RELATED: "label:has(#associations)",
     WordstatView.DYNAMICS: "label[for='graph']",
     WordstatView.REGIONS: "label[for='map']",
 }
+
+_QUOTED_PHRASE_MARKERS = (("«", "»"), ('"', '"'), ("“", "”"), ("„", "“"))
+
+
+def _assert_export_phrase(dataset: CsvDataset, phrase: str, view: WordstatView) -> None:
+    """Reject a CSV whose metadata identifies a different search phrase.
+
+    Wordstat's localized metadata wording is not a contract, but the query
+    itself is embedded as a quoted value in the report header.  Check that
+    stable identity rather than matching the surrounding localized prose.
+    Two-column synthetic/legacy exports have no metadata field and remain
+    accepted for backwards compatibility; live Wordstat exports have three
+    or more columns.
+    """
+    if len(dataset.headers) < 3:
+        return
+    if any(
+        f"{opening}{phrase}{closing}" in header
+        for header in dataset.headers
+        for opening, closing in _QUOTED_PHRASE_MARKERS
+    ):
+        return
+    raise InterfaceChangedError(
+        f"Wordstat {view.value} CSV metadata does not identify the requested phrase {phrase!r}"
+    )
 # The live interface is Russian; map explicitly rather than depending on the
 # host locale. Shared by every place that needs a Russian month name (the
 # calendar popups and the applied-period wait below), so there is exactly
@@ -598,6 +624,7 @@ async def _collect_one(
                     # legitimately empty after that retry, so this is kept
                     # separate from the fail-closed predicate below.
                     dataset = parse_wordstat_csv(source, view)
+                    _assert_export_phrase(dataset, phrase, view)
                     if _should_retry_empty_export(view, dataset) or _is_untrustworthy_empty_export(view, dataset):
                         retry = await self._retry_empty_export(
                             page,
@@ -609,6 +636,7 @@ async def _collect_one(
                             escaped_download_warnings,
                         )
                         source, dataset = retry.source, retry.dataset
+                        _assert_export_phrase(dataset, phrase, view)
                     if _is_untrustworthy_empty_export(view, dataset):
                         raise InterfaceChangedError(
                             f"Wordstat returned an empty {view.value} CSV after a retry, but the page had "
@@ -1100,6 +1128,14 @@ async def _set_phrase(self, page, phrase: str) -> None:
             f"""() => new URL(location.href).searchParams.get('words') === {expected_phrase}
                 && Boolean(document.querySelector({download_selector}))""",
         )
+        # A phrase switch preserves the previously selected top-level view.
+        # Return to the table view before selecting a top-level subview; on
+        # the map, the popular/related radio controls are not in the DOM.
+        await self._click(page, TABLE_VIEW_SELECTOR)
+        await self._wait_for(
+            page,
+            f"() => document.querySelectorAll({json.dumps(VIEW_SELECTORS[WordstatView.TOP_POPULAR])}).length === 1",
+        )
 
     async def _set_region(self, page, region: str) -> None:
         await self._click(page, REGION_BUTTON_SELECTOR)

tests/test_collector_batch.py

@@ -485,6 +485,117 @@ async def fake_download(self, page, session, dl_path):
     assert parquet.read_table(result.run_directory / top_export.file).num_rows == top_export.row_count
 
 
+def test_collect_one_rejects_csv_for_a_different_phrase_before_writing(monkeypatch, tmp_path):
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    async def fake_download(self, page, session, dl_path):
+        source = dl_path / "export.csv"
+        source.write_text(
+            "Запросы со словами;Число запросов;Топ частотных запросов «другая фраза», Россия\n",
+            encoding="utf-8",
+        )
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, keep_raw=True, settling_seconds=0, empty_export_retry_seconds=0)
+    with pytest.raises(InterfaceChangedError, match="does not identify"):
+        asyncio.run(
+            collector._collect_one(
+                _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+            )
+        )
+
+    run_directories = list((tmp_path / "runs").iterdir())
+    assert len(run_directories) == 1
+    assert not list(run_directories[0].glob("*.parquet"))
+    assert list(run_directories[0].glob("*.csv"))
+
+
+def test_collect_one_rejects_different_phrase_after_empty_export_retry(monkeypatch, tmp_path):
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+    download_count = 0
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    async def fake_download(self, page, session, dl_path):
+        nonlocal download_count
+        download_count += 1
+        source = dl_path / f"export-{download_count}.csv"
+        phrase = "тест" if download_count == 1 else "другая фраза"
+        source.write_text(
+            f"Запросы со словами;Число запросов;Топ частотных запросов «{phrase}», Россия\n",
+            encoding="utf-8",
+        )
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, keep_raw=True, settling_seconds=0, empty_export_retry_seconds=0)
+    with pytest.raises(InterfaceChangedError, match="does not identify"):
+        asyncio.run(
+            collector._collect_one(
+                _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+            )
+        )
+
+    assert download_count == 2
+    run_directories = list((tmp_path / "runs").iterdir())
+    assert len(run_directories) == 1
+    assert not list(run_directories[0].glob("*.parquet"))
+    assert list(run_directories[0].glob("*.csv"))
+
+
+def test_collect_one_rejects_different_phrase_on_nonempty_dynamics_export(monkeypatch, tmp_path):
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+    download_count = 0
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    async def fake_download(self, page, session, dl_path):
+        nonlocal download_count
+        download_count += 1
+        source = dl_path / f"export-{download_count}.csv"
+        if download_count < 3:
+            _write_view_csv(source, "тест")
+        else:
+            source.write_text(
+                "Период;Число запросов;Доля от всех запросов, %;"
+                "Динамика частотности запросов «другая фраза», по месяцам\n"
+                "январь 2024;100;1;100\n",
+                encoding="utf-8",
+            )
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, keep_raw=True, settling_seconds=0, empty_export_retry_seconds=0)
+    result = asyncio.run(
+        collector._collect_one(
+            _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+        )
+    )
+
+    assert download_count == 3
+    assert "InterfaceChangedError" in result.view_errors[WordstatView.DYNAMICS]
+    assert not list(result.run_directory.glob("dynamics*.parquet"))
+    assert list(result.run_directory.glob("export-3.csv"))
+
+
 def test_collect_one_accepts_persistent_empty_top_exports_after_retry(monkeypatch, tmp_path):
     """Retrying top exports must not restore PR #25's fail-closed regression."""
 

tests/test_collector_view.py

@@ -7,12 +7,75 @@
 
 import pytest
 
-from wordstat.collector import WordstatCollector, _is_untrustworthy_empty_export
+from wordstat.collector import (
+    SEARCH_SELECTOR,
+    TABLE_VIEW_SELECTOR,
+    VIEW_SELECTORS,
+    WordstatCollector,
+    _assert_export_phrase,
+    _is_untrustworthy_empty_export,
+)
 from wordstat.errors import InterfaceChangedError
 from wordstat.models import CsvDataset, WordstatView
 from wordstat.periods import Granularity
 
 
+def test_view_selectors_target_each_control_not_the_table_parent():
+    assert VIEW_SELECTORS[WordstatView.TOP_POPULAR] == "label:has(#popular)"
+    assert VIEW_SELECTORS[WordstatView.TOP_RELATED] == "label:has(#associations)"
+    assert "#table" not in VIEW_SELECTORS[WordstatView.TOP_POPULAR]
+
+
+def test_set_phrase_returns_to_table_before_subview_selection(monkeypatch, tmp_path):
+    clicks = []
+    waits = []
+
+    class Element:
+        async def fill(self, phrase):
+            pass
+
+    class Page:
+        async def get_elements_by_css_selector(self, selector):
+            return [Element()]
+
+        async def evaluate(self, expression):
+            return '{"value":"подарки","searchDisabled":false}'
+
+    async def click(self, page, selector):
+        clicks.append(selector)
+
+    async def wait(self, page, expression, seconds=None, required=True):
+        waits.append(expression)
+
+    monkeypatch.setattr(WordstatCollector, "_click", click)
+    monkeypatch.setattr(WordstatCollector, "_wait_for", wait)
+
+    asyncio.run(WordstatCollector("cdp", tmp_path)._set_phrase(Page(), "подарки"))
+
+    assert clicks == [SEARCH_SELECTOR, TABLE_VIEW_SELECTOR]
+    assert any(VIEW_SELECTORS[WordstatView.TOP_POPULAR] in expression for expression in waits)
+
+
+@pytest.mark.parametrize(
+    ("phrase", "header", "expected"),
+    [
+        ("подарки", "Топ частотных запросов «подарки», Россия", True),
+        ("подарки", "Запросы, похожие на «подарки», Россия", True),
+        ("подарки", "Запросы, похожие на «новогодние подарки», Россия", False),
+        ("подарки", "Top queries \"подарки\", Russia", True),
+        ("подарки", "Top queries подарки, Russia", False),
+    ],
+)
+def test_assert_export_phrase_uses_exact_quoted_phrase(phrase, header, expected):
+    dataset = _dataset(WordstatView.TOP_POPULAR, [])
+    dataset = dataset.model_copy(update={"headers": ["query", "count", header]})
+    if expected:
+        _assert_export_phrase(dataset, phrase, WordstatView.TOP_POPULAR)
+    else:
+        with pytest.raises(InterfaceChangedError, match="does not identify"):
+            _assert_export_phrase(dataset, phrase, WordstatView.TOP_POPULAR)
+
+
 def test_select_view_retries_once_when_active_marker_does_not_change(monkeypatch, tmp_path):
     clicks = []
     waits = []