Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 44 additions & 11 deletions app/verify/wikipedia_smartphone_backfill.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,35 @@
("google", "Google_Pixel_Tablet", "Google Pixel Tablet"),
)

WATCH_CROSSREF_PAGES: tuple[tuple[str, str, str], ...] = (
("apple", "Apple_Watch", "Apple Watch"),
("samsung", "Samsung_Galaxy_Watch_series", "Samsung Galaxy Watch series"),
("samsung", "Samsung_Gear", "Samsung Gear"),
("google", "Pixel_Watch", "Pixel Watch"),
("fitbit", "List_of_Fitbit_products", "List of Fitbit products"),
("garmin", "Garmin_Forerunner", "Garmin Forerunner"),
("huawei", "Huawei_Watch", "Huawei Watch"),
("pebble", "Pebble_(watch)", "Pebble (watch)"),
)

PDA_CROSSREF_PAGES: tuple[tuple[str, str, str], ...] = (
("palm", "Palm_(companion)", "Palm (companion)"),
("palm", "Palm_Treo", "Palm Treo"),
("hp", "HP_iPAQ", "HP iPAQ"),
("htc", "List_of_HTC_devices", "List of HTC devices"),
("blackberry", "BlackBerry", "BlackBerry"),
("sony", "CLIÉ", "CLIÉ"),
("casio", "Casio_Cassiopeia", "Casio Cassiopeia"),
("dell", "Dell_Axim", "Dell Axim"),
)

CATEGORY_CROSSREF_PAGES = {
"smartphone": CROSSREF_PAGES,
"tablet": TABLET_CROSSREF_PAGES,
"watch": WATCH_CROSSREF_PAGES,
"pda": PDA_CROSSREF_PAGES,
}

_BRAND_TOKENS = (
"samsung",
"apple",
Expand All @@ -121,6 +150,13 @@
"meizu",
"infinix",
"tecno",
"palm",
"garmin",
"fitbit",
"pebble",
"hp",
"dell",
"casio",
)

# Header rules for parsing smartphone tables in list articles.
Expand Down Expand Up @@ -1182,10 +1218,10 @@ def backfill(
cache_path: Path | None = None,
category: str = "smartphone",
) -> RunResult:
if category not in {"smartphone", "tablet"}:
if category not in CATEGORY_CROSSREF_PAGES:
raise ValueError(f"unsupported category: {category}")
writing = apply and not dry_run
html_cache = data_root / "data" / "_verify" / "cache" / "wikipedia_html"
html_cache = None if dry_run else data_root / "data" / "_verify" / "cache" / "wikipedia_html"
polite = PoliteWiki(sleep_s=sleep_s, cache_dir=html_cache) if fetch_page is None else None
fetch = fetch_page or (polite.fetch if polite is not None else None)
assert fetch is not None
Expand All @@ -1198,9 +1234,7 @@ def backfill(
parsed_pages: set[str] = set()

# Pre-parse list pages
target_pages = pages if pages is not None else (
TABLET_CROSSREF_PAGES if category == "tablet" else CROSSREF_PAGES
)
target_pages = pages if pages is not None else CATEGORY_CROSSREF_PAGES[category]
for _mfg, page, _title in target_pages:
status, final, html = fetch(page)
_alive, reason = classify(f"https://en.wikipedia.org/wiki/{page}", status, final or None)
Expand Down Expand Up @@ -1259,7 +1293,7 @@ def consider_fallback(base: str, record_brand: str = "") -> None:
if records is None:
phone_dir, repo_root = smartphone_scan_root(data_root, category)
chosen = sample_diverse_records(phone_dir, repo_root, limit, exclude_paths=cached_paths)
if category == "tablet" and limit is not None and len(chosen) < limit:
if category != "smartphone" and limit is not None and len(chosen) < limit:
remaining = sample_diverse_records(
phone_dir,
repo_root,
Expand Down Expand Up @@ -1325,12 +1359,11 @@ def maybe_write(rel: str, decision: object, url: object) -> None:
consider_fallback(phone.base, record_brand=rec_b)
hits = matching_rows(name, fetcher.rows, record_brand=rec_b)

if category == "tablet":
if category != "smartphone":
# A shared substring or a brand-omitted article is insufficient for
# the tablet batch: the article heading must name this exact model.
# these category batches: the article heading must name this exact model.
hits = [
row for row in hits
if normalize_heading(phone.base) == normalize_heading(row.model)
row for row in hits if normalize_heading(phone.base) == normalize_heading(row.model)
]

live = _liveness_for(hits[0].url if hits else None, page_liveness)
Expand Down Expand Up @@ -1415,7 +1448,7 @@ def render_summary(
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--data-root", type=Path, default=Path("."), help="TechAPI repository root")
parser.add_argument("--category", choices=("smartphone", "tablet"), default="smartphone")
parser.add_argument("--category", choices=tuple(CATEGORY_CROSSREF_PAGES), default="smartphone")
parser.add_argument("--limit", type=int, default=300, help="Max records to process")
parser.add_argument(
"--sleep", type=float, default=MIN_SLEEP_S, help="Sleep between Wikipedia calls"
Expand Down
74 changes: 74 additions & 0 deletions tests/verify/test_wikipedia_smartphone_backfill.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,13 @@
import json
from pathlib import Path

import pytest

from app.verify import wikipedia_smartphone_backfill as wiki_backfill
from app.verify.crossref import _heading_matches
from app.verify.wikipedia_smartphone_backfill import (
PDA_CROSSREF_PAGES,
WATCH_CROSSREF_PAGES,
WikiRow,
backfill,
decide,
Expand Down Expand Up @@ -241,3 +246,72 @@ def test_apply_writes_only_confirmed_records(tmp_path: Path) -> None:
"https://en.wikipedia.org/wiki/List_of_Samsung_Galaxy_smartphones#Galaxy_S_series"
in updated["source_urls"]
)


@pytest.mark.parametrize(
("category", "pages", "brand", "model"),
[
("watch", WATCH_CROSSREF_PAGES, "apple", "Apple Watch Series 6"),
("pda", PDA_CROSSREF_PAGES, "dell", "Dell Axim X5"),
],
)
def test_category_pages_and_exact_heading(
tmp_path: Path,
category: str,
pages: tuple[tuple[str, str, str], ...],
brand: str,
model: str,
) -> None:
page = next(page for page_brand, page, _ in pages if page_brand == brand)
html = f"""<table class="wikitable"><tr><th>Model</th><th>Released</th>
<th>RAM</th><th>Battery</th></tr><tr><td>{model}</td><td>2020</td>
<td>8 GB</td><td>4000 mAh</td></tr></table>"""
fetched: list[str] = []

def fetch(candidate: str) -> tuple[int, str, str]:
fetched.append(candidate)
return 200, f"https://en.wikipedia.org/wiki/{candidate}", html if candidate == page else ""

record = _sample_rec(name=model, brand=brand)
result = backfill(
tmp_path,
category=category,
records=[(f"data/{category}/{brand}/model.json", record)],
fetch_page=fetch,
search_fn=lambda _name: [],
cache_path=tmp_path / "cache.jsonl",
)
assert fetched == [item[1] for item in pages]
assert result.counts()["confirm"] == 1
assert result.written == 0

near_match = {**record, "name": f"{model} Pro"}
rejected = backfill(
tmp_path,
category=category,
records=[(f"data/{category}/{brand}/near.json", near_match)],
pages=[(brand, page, model)],
fetch_page=fetch,
search_fn=lambda _name: [],
)
assert rejected.counts()["confirm"] == 0


@pytest.mark.parametrize("category", ["watch", "pda"])
def test_category_cli_uses_category_cache(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, category: str
) -> None:
captured: dict[str, object] = {}

def fake_backfill(data_root: Path, **kwargs: object) -> wiki_backfill.RunResult:
captured.update(kwargs)
return wiki_backfill.RunResult()

monkeypatch.setattr(wiki_backfill, "backfill", fake_backfill)
assert wiki_backfill.main(["--data-root", str(tmp_path), "--category", category]) == 0
assert captured["category"] == category
assert (
captured["cache_path"]
== tmp_path / "data" / "_verify" / "state" / f"wikipedia_{category}_cache.jsonl"
)
assert captured["apply"] is False
Loading