Function fetch_all

scripts/fetch_nifdc_standards.py:164–222

Paginate the full standards list (general + professional domains). NIFDC ``qxqwk.do`` lists only 228 general-domain standards by default. To fetch ALL 2100+ standards (including the professional/specialized domain), we use the ``istiaojian`` query parameter -- any non-empty value trigge

(max_pages: Optional[int] = None)

def _stable_id(number: str) -> str:
    base = number.lower().strip().replace(" ", "-").replace("/", "-")
    return f"nmpa-{re.sub(r'[^a-z0-9.-]+', '-', base)}"


def fetch_all(max_pages: Optional[int] = None) -> tuple[list[dict], dict]:
    """Paginate the full standards list (general + professional domains).

    NIFDC ``qxqwk.do`` lists only 228 general-domain standards by default. To
    fetch ALL 2100+ standards (including the professional/specialized domain),
    we use the ``istiaojian`` query parameter -- any non-empty value triggers
    the full-catalog pagination. Each page returns up to ~300 rows (from
    multiple ``<table>`` elements). We paginate with ``index=N`` in the URL.

    Stops when two consecutive pages yield zero new (de-duped) rows.
    """
    session = _make_session()
    # Warm up session cookies
    _request_with_retry(session, "GET", LIST_URL)
    all_rows: list[dict] = []
    seen_keys: set[str] = set()
    page = 1
    consecutive_empty = 0
    while True:
        if max_pages and page > max_pages:
            break
        url = f"{LIST_ACTION_URL}&istiaojian=all&index={page}"
        r = _request_with_retry(session, "GET", url)
        if r is None:
            print(f"[error] page {page} failed all retries; stopping.")
            break
        if r.status_code != 200:
            print(f"[error] page {page} HTTP {r.status_code}; stopping.")
            break
        rows, _info = _parse_listing_page(r.text)
        new_count = 0
        for row in rows:
            key = row["number"]
            if key in seen_keys:
                continue
            seen_keys.add(key)
            row["status"] = _normalize_status(row.get("status_zh", ""))
            row["id"] = _stable_id(row["number"])
            row["source_url"] = LIST_URL
            all_rows.append(row)
            new_count += 1
        print(f"[info] page {page}: parsed {len(rows)} (new {new_count}; total {len(all_rows)})", flush=True)
        if new_count == 0:
            consecutive_empty += 1
            if consecutive_empty >= 2:
                break
        else:
            consecutive_empty = 0
        if not rows:
            break
        page += 1
        time.sleep(SLEEP_BETWEEN)
    return all_rows, {