#!/usr/bin/env python
"""Polite Wikidata harvester for the Company Atlas seed registry (docs/SEEDS.md).

Stages (each resumable; every SPARQL result is cached under data/seed/wikidata/sparql/<sha256>.json):
  candidates  company-class × sitelink-band queries (sitelinks ≥ 3), every listed company (P414 per exchange / P249), companies with
              ≥ 500 employees (P1128) or a revenue (P2139) per class, per-country and per-industry boost queries → data/seed/candidates.json
  dissolved   P576 / P582 check over every candidate → data/seed/dissolved.json (excluded by `select`)
  select      normalise websites, drop generic hosts / duplicate domains, diversify (listed companies first, US ≤ 35 %, others ≤ 12 %,
              country minimums) → data/seed/selected.json
  details     batched detail queries (labels, legal names, industries, HQ, coordinates, tickers, LEI, CIK, parent, logo, employees)
              → data/seed/details.json
  assemble    importance + tiers + industry mapping + parent/domain conflicts → registry/companies/wikidata-<region>.ndjson + README.md
  all         the four stages in sequence (default)

Politeness: User-Agent CompanyAtlasBot/0.1, one query at a time, ≥ 2 s between queries, 60 s server timeout, retries with backoff.
"""
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import logging
import math
import re
import sys
import time
from collections import Counter, defaultdict
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib.parse import quote, unquote, urlparse

import httpx

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from companyatlas.registry.industries import load_industries, map_industry, top_level_of, top_level_slugs
from companyatlas.urls import registrable_domain

log = logging.getLogger("seed_wikidata")

SPARQL_URL = "https://query.wikidata.org/sparql"
USER_AGENT = "CompanyAtlasBot/0.1 (contact@spboucher.ai)"
MIN_INTERVAL_S = 2.0
TIMEOUT_S = 60.0
MAX_TRIES = 6
DATA_DIR = ROOT / "data" / "seed"
CACHE_DIR = DATA_DIR / "wikidata" / "sparql"
REGISTRY_DIR = ROOT / "registry"
OUT_DIR = REGISTRY_DIR / "companies"
COUNTRIES_CSV = REGISTRY_DIR / "countries.csv"

# Wikidata classes queried with wdt:P31 (no P279* — the subclass tree of "business" is too broad to page). Banded classes are large.
# Labels are the English Wikidata labels (checked 2026-09-13: the 2026-09-12 list had six wrong labels, and three QIDs that are not
# company classes at all — Q708676 "charitable organization", Q1668024 "service on Internet", Q19967801 "online service" — were dropped).
CLASSES: dict[str, str] = {
    "Q4830453": "business", "Q6881511": "enterprise", "Q891723": "public company", "Q783794": "company", "Q1589009": "privately held company",
    "Q167037": "corporation", "Q18388277": "technology company", "Q1058914": "software company", "Q22687": "bank", "Q46970": "airline",
    "Q786820": "automobile manufacturer", "Q6500733": "printing company", "Q19644607": "pharmaceutical company", "Q210167": "video game developer",
    "Q1137109": "video game publisher", "Q1762059": "film production company", "Q18127": "record label", "Q2085381": "publishing house",
    "Q613142": "law firm", "Q740752": "transport company", "Q2401749": "telecommunications company", "Q131734": "brewery",
    "Q507619": "retail chain", "Q936518": "aerospace manufacturer", "Q206361": "concern", "Q778575": "conglomerate", "Q249556": "railway company",
    "Q2005696": "commercial vehicle manufacturer", "Q190928": "shipyard", "Q1320047": "book publisher",
    # Added 2026-09-13 for the 30 k harvest: the classes most used by listed / large companies outside the generic ones above.
    "Q134161": "joint-stock company", "Q1480166": "kabushiki gaisha", "Q60997538": "kōkai gaisha", "Q219577": "holding company",
    "Q650241": "financial institution", "Q730038": "credit institution", "Q2143354": "insurance company", "Q697852": "real estate investment trust",
    "Q658255": "subsidiary company", "Q270791": "state-owned enterprise", "Q161726": "multinational corporation", "Q18534542": "restaurant chain",
    "Q1631129": "hotel chain", "Q64027599": "gas station chain",
}
BANDED_CLASSES = {"Q4830453", "Q6881511", "Q891723", "Q46970"}
# Country boosts run only when the class stage yielded fewer than BOOST_MARGIN × minimum candidates for that country.
BOOST_MARGIN = 2.0
SITELINK_BANDS: list[tuple[int, int | None]] = [(80, None), (50, 80), (35, 50), (25, 35), (18, 25), (12, 18), (8, 12), (5, 8), (3, 5)]
MIN_SITELINKS = 5                 # single-query classes were harvested at ≥ 5 first (2026-09-12); the (3, 5) band is added for every class
LOW_BAND: tuple[int, int] = (3, 5)
SELECT_MIN_SITELINKS = 3          # class-band candidates need ≥ 3 sitelinks to be selected …
FALLBACK_MIN_SITELINKS = 2        # … unless the target is still short, then ≥ 2 (industry / country boosts reach down to 2)
EMPLOYEES_MIN = 500               # P1128 band: companies with at least this many employees, any sitelink count
BOOST_MIN_SITELINKS = 2
BOOST_LIMIT = 2500
INDUSTRY_BOOST_LIMIT = 600
EXCHANGE_GROUP_MAX = 1500         # listed-company queries group small exchanges until ≈ this many P414 statements
SPLIT_DEPTH_MAX = 2               # a query that still times out is split by the last (then the last two) digits of the QID
CLASS_BATCH = 300                 # QIDs per P31 class query in the details stage

# Non-company exclusion (assemble, 2026-09-13): an item carrying one of these P31 classes is dropped unless it has a current stock-exchange
# listing or a ticker (listed sports clubs such as Manchester United or Juventus stay). Companies = business / enterprise / company /
# public company / corporation / state-owned enterprise / cooperative / bank / insurer / airline / manufacturer / retailer … (CLASSES).
NON_COMPANY_CLASSES: dict[str, str] = {
    # museums, libraries, archives
    "Q33506": "museum", "Q207694": "art museum", "Q2087181": "historic house museum", "Q17431399": "national museum", "Q7075": "library",
    "Q28564": "public library", "Q26271642": "library network", "Q166118": "archives",
    # education
    "Q3918": "university", "Q3914": "school", "Q9826": "high school", "Q159334": "secondary school", "Q189004": "college",
    "Q2385804": "educational institution", "Q875538": "public university", "Q902104": "private university", "Q1336920": "community college",
    "Q269770": "boarding school", "Q2418495": "independent school", "Q615150": "land-grant university", "Q62078547": "public research university",
    "Q23002039": "public educational institution of the United States", "Q23002054": "private not-for-profit educational institution",
    "Q38723": "higher education institution", "Q1371037": "institute of technology", "Q3354859": "collegiate university",
    # government
    "Q327333": "government agency", "Q192350": "ministry", "Q732717": "law enforcement agency", "Q2659904": "government organization",
    "Q7188": "government", "Q4287745": "medical organization",
    # nonprofit sector
    "Q79913": "non-governmental organization", "Q708676": "charitable organization", "Q157031": "foundation", "Q163740": "nonprofit organization",
    "Q18325436": "501(c)(3) organization", "Q48204": "voluntary association", "Q15911314": "association", "Q829080": "professional association",
    "Q955824": "learned society", "Q155271": "think tank", "Q431603": "advocacy group", "Q1666019": "pressure group",
    "Q5774403": "historical society", "Q1021488": "community foundation", "Q1785733": "environmental organization",
    "Q1899015": "conservation organization", "Q336473": "aid agency", "Q4438121": "sports organization", "Q1530022": "religious organization",
    "Q94670589": "Christian organization", "Q7278": "political party", "Q178790": "labor union", "Q16917": "hospital",
    # sports clubs and teams (kept only when listed)
    "Q476028": "association football club", "Q847017": "sports club", "Q12973014": "sports team", "Q13393265": "basketball team",
    "Q14752149": "amateur football club", "Q18558301": "college sports team",
    # channels
    "Q17558136": "YouTube channel",
}
# Description rule: matches this (and none of the company words below) with no ticker, exchange, employee count or revenue → dropped.
NON_COMPANY_DESC_RE = re.compile(r"\b(museum|university|school|ministry|agency|charity|foundation|association|club|church|channel)\b", re.IGNORECASE)
COMPANY_DESC_RE = re.compile(
    r"\b(compan(?:y|ies)|corporation|manufacturer|enterprise|firm|conglomerate|retailer|bank|insurer|insurance|developer|publisher|publishing|producer|"
    r"operator|provider|chain|holding|startup|studio|label|brand|business|airline|carrier|maker|supplier|distributor|wholesaler|contractor|"
    r"consultancy|consulting|(?:advertising|travel|news|talent|marketing|staffing|recruitment|employment|real[- ]estate|estate|creative|design|"
    r"digital|literary|model(?:ing|ling)?|public relations|PR|media|shipping|rating|photo|press|ad|insurance|shipping)\s+agenc(?:y|ies))\b", re.IGNORECASE)

# Diversification (spec: "seed diversified companies across US, Canada, Europe, UK, Japan, South Korea, India, Australia, LatAm, ME, Africa, SEA")
US_CAP_SHARE = 0.35
OTHER_CAP_SHARE = 0.12
UNKNOWN_COUNTRY_SHARE = 0.03
# Tiers by quantile of importance: top 1 % → tier 1 (≈ 300 at 30 k), next 6.67 % → tier 2 (≈ 2,000), next 26.67 % → tier 3 (≈ 8,000), rest 4.
TIER_SHARES: tuple[float, float, float] = (0.01, 0.0667, 0.2667)
COUNTRY_MINIMUMS: dict[str, int] = {
    **dict.fromkeys(["CA", "GB", "DE", "FR", "JP", "KR", "IN", "AU"], 150),
    **dict.fromkeys(["BR", "MX", "AE", "SA", "ZA", "NG", "SG", "ID", "NL", "SE", "CH", "ES", "IT", "CN", "TW", "HK"], 60),
}
# Share-of-target caps for narrow classes that Wikidata over-represents among high-sitelink items (airlines, record labels, publishers…).
CLASS_CAPS: dict[str, float] = {"Q46970": 0.05, "Q18127": 0.03, "Q210167": 0.04, "Q1137109": 0.02, "Q2085381": 0.03, "Q1320047": 0.02,
                                "Q1762059": 0.03, "Q190928": 0.02, "Q131734": 0.02, "Q249556": 0.03}
INDUSTRY_MINIMUM = 60
INDUSTRY_RESERVE = 150      # extra candidates per top-level industry fetched in the detail stage to satisfy INDUSTRY_MINIMUM
# English labels of Wikidata items commonly used as P452 (industry) values, per top-level industry, for the industry-boost queries.
INDUSTRY_BOOST_LABELS: dict[str, list[str]] = {
    "technology": ["information technology", "electronics industry", "consumer electronics", "computer hardware", "electronics"],
    "financial-services": ["financial services", "insurance", "banking", "asset management", "investment banking", "payment system"],
    "real-estate": ["real estate", "real estate development", "property management", "real estate industry"],
    "construction": ["construction", "construction industry", "civil engineering", "building construction", "engineering"],
    "retail": ["retail", "retailing", "supermarket", "department store", "wholesale", "grocery store"],
    "consumer-goods": ["consumer goods", "cosmetics industry", "toy industry", "furniture industry", "fast-moving consumer goods",
                       "household goods", "luxury goods", "personal care", "home appliance"],
    "energy": ["energy industry", "energy", "electric power industry", "nuclear power", "oil industry", "petroleum industry",
               "renewable energy", "solar energy", "wind power"],
    "utilities": ["public utility", "electric utility", "water industry", "waste management", "electricity generation",
                  "electric power distribution", "water supply"],
    "mining": ["mining", "mining industry", "steel industry", "metallurgy", "gold mining", "steelmaking", "metal industry"],
    "chemicals": ["chemical industry", "chemicals", "petrochemical industry", "specialty chemicals", "plastics industry", "fertilizer"],
    "materials": ["glass industry", "paper industry", "packaging industry", "pulp and paper industry", "forestry", "building material",
                  "cement industry", "wood industry"],
    "manufacturing": ["manufacturing", "mechanical engineering", "industrial machinery", "machine industry", "electrical engineering",
                      "machine tool", "industrial engineering", "heavy industry", "shipbuilding"],
    "automotive": ["automotive industry", "automobile", "automotive", "motor vehicle", "auto parts"],
    "aerospace-defense": ["aerospace industry", "arms industry", "defense industry", "aerospace", "aviation industry", "space industry",
                          "defence industry"],
    "transportation": ["transport", "rail transport", "public transport", "logistics", "transportation", "freight transport", "shipping",
                       "maritime transport", "railway"],
    "telecommunications": ["telecommunications industry", "telecommunications", "telecommunication", "mobile telephony",
                           "internet service provider"],
    "media": ["mass media", "publishing", "advertising", "broadcasting", "entertainment industry", "film industry", "video game industry",
              "music industry", "media industry", "newspaper"],
    "healthcare": ["health care industry", "health care", "medical technology", "hospital", "pharmaceutical industry", "biotechnology",
                   "medical device", "healthcare industry", "medical equipment"],
    "hospitality": ["hospitality industry", "hotel industry", "hotel", "restaurant", "hospitality", "catering", "restaurant chain"],
    "travel": ["tourism", "travel agency", "travel industry", "travel", "tourism industry", "online travel agency", "tour operator"],
    "education": ["education", "educational technology", "higher education", "e-learning", "education industry", "educational services"],
    "professional-services": ["professional services", "consulting", "management consulting", "accounting", "outsourcing", "staffing",
                              "business services", "legal services", "information technology consulting", "human resources"],
    "agriculture": ["agriculture", "agribusiness", "agricultural industry", "forestry", "fishing industry", "food industry", "farming",
                    "aquaculture", "agricultural machinery"],
}

# Hosts that are never a company's own website (social profiles, blogs, stores, code forges, encyclopaedias, site builders …).
GENERIC_DOMAINS = {
    "facebook.com", "fb.com", "linkedin.com", "twitter.com", "x.com", "instagram.com", "youtube.com", "youtu.be", "wikipedia.org",
    "wikimedia.org", "wikidata.org", "blogspot.com", "blogspot.co.uk", "wordpress.com", "tumblr.com", "medium.com", "github.com", "gitlab.com",
    "sourceforge.net", "archive.org", "tiktok.com", "vk.com", "weibo.com", "t.me", "telegram.me", "bit.ly", "wix.com", "wixsite.com",
    "weebly.com", "squarespace.com", "webnode.com", "jimdo.com", "jimdosite.com", "godaddysites.com", "carrd.co", "notion.site", "substack.com",
    "patreon.com", "itch.io", "steampowered.com", "bandcamp.com", "soundcloud.com", "imdb.com", "myspace.com", "flickr.com", "pinterest.com",
    "twitch.tv", "discord.gg", "discord.com", "bilibili.com", "tistory.com", "ameblo.jp", "fc2.com", "livejournal.com", "geocities.com",
    "angelfire.com", "tripod.com", "netlify.app", "vercel.app", "herokuapp.com", "github.io", "pages.dev", "web.app", "firebaseapp.com",
    "glitch.me", "strikingly.com", "yolasite.com", "webs.com", "linktr.ee", "about.me", "crunchbase.com", "bloomberg.com", "sec.gov",
    "yelp.com", "tripadvisor.com", "foursquare.com", "goo.gl", "ow.ly", "tinyurl.com", "wa.me", "whatsapp.com", "line.me", "kakao.com",
    "spotify.com", "deezer.com", "vimeo.com", "dailymotion.com", "behance.net", "dribbble.com", "etsy.com", "ebay.com", "aliexpress.com",
    "taobao.com", "tmall.com", "rakuten.co.jp", "shopee.com", "mercadolibre.com", "google.co.uk", "googleusercontent.com", "gstatic.com",
    "webflow.io", "mystrikingly.com", "site123.me", "simplesite.com", "ucoz.ru", "narod.ru", "hatenablog.com", "note.com", "wixstatic.com",
    "shopify.com", "myshopify.com", "bigcartel.com", "storenvy.com", "yahoo.co.jp", "yahoo.com", "aol.com", "cargo.site", "format.com",
    "blogger.com", "mixi.jp", "naver.me", "cafe24.com", "modoo.at", "over-blog.com", "canalblog.com", "skyrock.com", "free.fr", "orange.fr",
    "wanadoo.fr", "pagesperso-orange.fr", "t-online.de", "web.de", "gmx.de", "chello.at", "bplaced.net", "beepworld.de", "npage.de",
    "altervista.org", "xoom.it", "libero.it", "interfree.it", "terra.com.br", "uol.com.br", "ig.com.br", "sapo.pt", "webcindario.com",
    "iespana.es", "galeon.com", "hpage.com", "wordpress.org", "js.org", "readthedocs.io", "gitbook.io", "gumroad.com", "ko-fi.com",
    "onlyfans.com", "reddit.com", "quora.com", "scribd.com", "issuu.com", "slideshare.net", "docs.google.com", "drive.google.com",
    "sites.google.com", "play.google.com", "apps.apple.com", "itunes.apple.com", "amazon.com", "amazon.co.uk", "amazon.de", "amazon.co.jp",
    "amazon.fr", "amazon.ca", "amazon.in", "amazon.com.br", "amzn.to", "microsoft.com", "apple.com", "google.com", "naver.com", "daum.net",
    "qq.com", "163.com", "sina.com.cn", "baidu.com", "sohu.com", "douyin.com", "kuaishou.com", "zhihu.com", "xiaohongshu.com",
}
# Registrable domains that are themselves seed companies: only the bare/www host counts as that company's site (not sub-brands/store pages).
PLATFORM_ROOTS = {"google.com": "www.google.com", "apple.com": "www.apple.com", "amazon.com": "www.amazon.com", "microsoft.com": "www.microsoft.com",
                  "naver.com": "www.naver.com", "yahoo.com": "www.yahoo.com", "qq.com": "www.qq.com", "baidu.com": "www.baidu.com",
                  "163.com": "www.163.com", "sohu.com": "www.sohu.com", "sina.com.cn": "www.sina.com.cn", "daum.net": "www.daum.net",
                  "kakao.com": "www.kakao.com", "shopify.com": "www.shopify.com", "spotify.com": "www.spotify.com", "reddit.com": "www.reddit.com",
                  "ebay.com": "www.ebay.com", "etsy.com": "www.etsy.com", "yelp.com": "www.yelp.com", "tripadvisor.com": "www.tripadvisor.com",
                  "linkedin.com": "www.linkedin.com", "facebook.com": "www.facebook.com", "instagram.com": "www.instagram.com",
                  "youtube.com": "www.youtube.com", "twitter.com": "twitter.com", "x.com": "x.com", "tiktok.com": "www.tiktok.com",
                  "github.com": "github.com", "gitlab.com": "gitlab.com", "medium.com": "medium.com", "substack.com": "substack.com",
                  "patreon.com": "www.patreon.com", "twitch.tv": "www.twitch.tv", "pinterest.com": "www.pinterest.com", "vimeo.com": "vimeo.com",
                  "soundcloud.com": "soundcloud.com", "bandcamp.com": "bandcamp.com", "imdb.com": "www.imdb.com", "crunchbase.com": "www.crunchbase.com",
                  "bloomberg.com": "www.bloomberg.com", "wix.com": "www.wix.com", "squarespace.com": "www.squarespace.com", "weebly.com": "www.weebly.com",
                  "godaddy.com": "www.godaddy.com", "wordpress.com": "wordpress.com", "tumblr.com": "www.tumblr.com", "flickr.com": "www.flickr.com",
                  "quora.com": "www.quora.com", "scribd.com": "www.scribd.com", "issuu.com": "issuu.com", "discord.com": "discord.com",
                  "telegram.org": "telegram.org", "whatsapp.com": "www.whatsapp.com", "line.me": "line.me", "bilibili.com": "www.bilibili.com",
                  "weibo.com": "weibo.com", "vk.com": "vk.com", "zhihu.com": "www.zhihu.com", "aliexpress.com": "www.aliexpress.com",
                  "taobao.com": "www.taobao.com", "tmall.com": "www.tmall.com", "rakuten.co.jp": "www.rakuten.co.jp", "shopee.com": "shopee.com",
                  "mercadolibre.com": "www.mercadolibre.com", "archive.org": "archive.org", "sourceforge.net": "sourceforge.net", "itch.io": "itch.io",
                  "steampowered.com": "store.steampowered.com", "deezer.com": "www.deezer.com", "dailymotion.com": "www.dailymotion.com",
                  "behance.net": "www.behance.net", "dribbble.com": "dribbble.com", "notion.so": "www.notion.so", "gumroad.com": "gumroad.com",
                  "onlyfans.com": "onlyfans.com", "yahoo.co.jp": "www.yahoo.co.jp", "aol.com": "www.aol.com", "free.fr": "www.free.fr",
                  "orange.fr": "www.orange.fr", "t-online.de": "www.t-online.de", "web.de": "web.de", "gmx.de": "www.gmx.de", "uol.com.br": "www.uol.com.br",
                  "terra.com.br": "www.terra.com.br", "sapo.pt": "www.sapo.pt", "libero.it": "www.libero.it", "douyin.com": "www.douyin.com",
                  "kuaishou.com": "www.kuaishou.com", "xiaohongshu.com": "www.xiaohongshu.com", "note.com": "note.com", "cafe24.com": "www.cafe24.com",
                  "hatenablog.com": "hatenablog.com", "mixi.jp": "mixi.jp", "myspace.com": "myspace.com", "livejournal.com": "www.livejournal.com",
                  "webflow.com": "webflow.com", "carrd.co": "carrd.co", "linktr.ee": "linktr.ee", "about.me": "about.me", "netlify.com": "www.netlify.com",
                  "vercel.com": "vercel.com", "heroku.com": "www.heroku.com", "glitch.com": "glitch.com", "readthedocs.org": "readthedocs.org",
                  "gitbook.com": "www.gitbook.com", "ko-fi.com": "ko-fi.com", "foursquare.com": "foursquare.com", "slideshare.net": "www.slideshare.net"}


# ------------------------------------------------------------------------------------------------------------ SPARQL client
class Sparql:
    def __init__(self, *, refresh: bool = False) -> None:
        self.client = httpx.Client(headers={"User-Agent": USER_AGENT, "Accept": "application/sparql-results+json"}, timeout=TIMEOUT_S + 15,
                                   follow_redirects=True)
        self.last_call = 0.0
        self.refresh = refresh
        self.queries = 0
        self.cached = 0
        CACHE_DIR.mkdir(parents=True, exist_ok=True)

    def query(self, sparql: str, *, label: str = "") -> list[dict[str, str]]:
        key = hashlib.sha256(sparql.encode("utf-8")).hexdigest()
        path = CACHE_DIR / f"{key}.json"
        if path.exists() and not self.refresh:
            self.cached += 1
            return json.loads(path.read_text(encoding="utf-8"))["bindings"]
        delay = 5.0
        last_err = ""
        gateway_timeouts = 0
        for attempt in range(1, MAX_TRIES + 1):
            wait = MIN_INTERVAL_S - (time.monotonic() - self.last_call)
            if wait > 0:
                time.sleep(wait)
            t0 = time.monotonic()
            try:
                r = self.client.get(SPARQL_URL, params={"query": sparql, "format": "json"})
                self.last_call = time.monotonic()
                self.queries += 1
                if r.status_code == 200:
                    try:
                        # strict=False: a handful of Wikidata literals contain raw control characters.
                        payload = json.loads(r.text, strict=False)
                    except json.JSONDecodeError:
                        # The endpoint streams results and, on a server-side timeout, appends a Java stack trace to a *200* body
                        # (which the gateway may even cache). Treat a truncated body as a timeout so callers can split or skip.
                        if "TimeoutException" in r.text or "SPARQL-QUERY" in r.text:
                            raise TimeoutError(f"server timeout (truncated body) {label}") from None
                        last_err = "truncated/invalid JSON body"
                        gateway_timeouts += 1
                        if gateway_timeouts >= 2:
                            raise TimeoutError(f"truncated body twice {label}") from None
                        continue
                    rows = [{k: v["value"] for k, v in b.items()} for b in payload["results"]["bindings"]]
                    path.write_text(json.dumps({"label": label, "fetched_at": datetime.now(UTC).isoformat(), "query": sparql, "bindings": rows},
                                               ensure_ascii=False), encoding="utf-8")
                    log.info("sparql ok %s rows=%d %.1fs", label, len(rows), self.last_call - t0)
                    return rows
                last_err = f"HTTP {r.status_code}: {r.text[:160]!r}"
                if r.status_code == 504:
                    # Gateway timeout = the query ran past the 60 s server limit. One retry (load varies), then let the caller split it.
                    gateway_timeouts += 1
                    if gateway_timeouts >= 2:
                        raise TimeoutError(f"gateway timeout {label}")
                if r.status_code == 429:
                    retry_after = r.headers.get("Retry-After")
                    delay = max(delay, float(retry_after)) if retry_after and retry_after.isdigit() else max(delay, 30.0)
                if r.status_code == 400:
                    raise RuntimeError(f"bad query {label}: {r.text[:500]}")
                if r.status_code == 500 and "TimeoutException" in r.text:
                    raise TimeoutError(f"server timeout {label}")
            except (httpx.TimeoutException, httpx.TransportError) as e:
                self.last_call = time.monotonic()
                last_err = f"{type(e).__name__}: {e}"
                if isinstance(e, httpx.RemoteProtocolError):
                    # "incomplete chunked read": the server cut the stream at its time limit — same as a timeout, fail fast.
                    gateway_timeouts += 1
                    if gateway_timeouts >= 2:
                        raise TimeoutError(f"stream cut by server {label}") from None
            log.warning("sparql retry %d/%d %s (%s) sleeping %.0fs", attempt, MAX_TRIES, label, last_err, delay)
            time.sleep(delay)
            delay = min(delay * 2, 120.0)
        raise RuntimeError(f"sparql failed {label}: {last_err}")


def values_clause(var: str, qids: list[str]) -> str:
    return f"VALUES ?{var} {{ {' '.join('wd:' + q for q in qids)} }}"


def class_values() -> str:
    return values_clause("cls", list(CLASSES))


# Truthy website (best rank, deprecated excluded). Dissolution is fetched as an OPTIONAL and filtered client-side: `FILTER NOT EXISTS`
# sub-selects over tens of thousands of bindings are what pushed the big class scans past the 60 s server limit.
WEBSITE_BLOCK = """
  ?item wdt:P856 ?web .
  OPTIONAL { ?item wdt:P576 ?dissolved }
  OPTIONAL { ?item wdt:P17 ?c . ?c wdt:P297 ?iso }
"""


def q_class_band(cls: str, lo: int, hi: int | None, extra: str = "") -> str:
    band = f"?sl >= {lo}" + (f" && ?sl < {hi}" if hi else "")
    return f"""SELECT ?item ?sl ?web ?iso WHERE {{
  ?item wdt:P31 wd:{cls} ; wikibase:sitelinks ?sl .
  FILTER({band})
  {WEBSITE_BLOCK}{extra}
}}"""


def suffix_filter(suffix: str) -> str:
    """Split filter for queries that time out as a whole: keep the items whose QID ends with `suffix` (standard SPARQL, no modulo)."""
    return f'  FILTER(STRENDS(STR(?item), "{suffix}"))\n' if suffix else ""


def q_exchanges() -> str:
    """Every stock exchange used as a P414 value with its statement count — drives the per-exchange listed-company scans."""
    return """SELECT ?ex ?exLabel (COUNT(?item) AS ?n) WHERE {
  ?item wdt:P414 ?ex .
  SERVICE wikibase:label { bd:serviceParam wikibase:language "en". }
} GROUP BY ?ex ?exLabel ORDER BY DESC(?n)"""


def q_listed(exchange_qids: list[str], extra: str = "") -> str:
    """Companies listed on the given exchanges (any P31 class): the listing's end time (P582) is fetched so delisted-only items are
    treated as not listed client-side (they remain ordinary candidates through their other bands)."""
    return f"""SELECT ?item ?sl ?web ?iso ?ex ?end WHERE {{
  {values_clause("ex", exchange_qids)}
  ?item p:P414 ?st . ?st ps:P414 ?ex .
  OPTIONAL {{ ?st pq:P582 ?end }}
  ?item wikibase:sitelinks ?sl .
  {WEBSITE_BLOCK}{extra}
}}"""


def q_ticker_only(extra: str = "") -> str:
    """Items with a ticker symbol as a main statement (P249 is normally a qualifier of P414; a few dozen items carry it directly)."""
    return f"""SELECT ?item ?sl ?web ?iso ?ticker WHERE {{
  ?item wdt:P249 ?ticker ; wikibase:sitelinks ?sl .
  {WEBSITE_BLOCK}{extra}
}}"""


def q_class_measure(cls: str, prop: str, minimum: int | None, extra: str = "") -> str:
    """Companies of one class with a P1128 (employees ≥ minimum) or P2139 (revenue, any value) statement, any sitelink count. Class-
    restricted on purpose: 70 % of the items with a revenue on Wikidata are nonprofits, universities, hospitals or municipalities."""
    cond = f"  FILTER(?val >= {minimum})\n" if minimum is not None else ""
    return f"""SELECT ?item ?sl ?web ?iso ?val WHERE {{
  ?item wdt:P31 wd:{cls} ; wdt:{prop} ?val ; wikibase:sitelinks ?sl .
{cond}  {WEBSITE_BLOCK}{extra}
}}"""


def q_country_qids(isos: list[str]) -> str:
    vals = " ".join(json.dumps(x) for x in isos)
    return f"""SELECT ?iso ?country WHERE {{ VALUES ?iso {{ {vals} }} ?country wdt:P297 ?iso . FILTER NOT EXISTS {{ ?country wdt:P576 [] }} }}"""


def q_country_boost(iso: str, country_qid: str) -> str:
    """Country-first scan (P17 → website → class) with the optimizer pinned. Cheap for countries with few items in Wikidata; the
    caller only issues it for countries still short of their minimum after the class stage and tolerates a timeout."""
    return f"""SELECT ?item ?sl ?web ?iso WHERE {{
  hint:Query hint:optimizer "None" .
  ?item wdt:P17 wd:{country_qid} .
  ?item wdt:P856 ?web .
  ?item wdt:P31 ?cls .
  {class_values()}
  ?item wikibase:sitelinks ?sl . FILTER(?sl >= {BOOST_MIN_SITELINKS})
  OPTIONAL {{ ?item wdt:P576 ?dissolved }}
  BIND("{iso}" AS ?iso)
}}"""


def q_industry_boost(labels: list[str]) -> str:
    vals = " ".join(json.dumps(x) + "@en" for x in labels)
    return f"""SELECT ?item ?sl ?web ?iso ?indLabel WHERE {{
  hint:Query hint:optimizer "None" .
  VALUES ?indLabel {{ {vals} }}
  ?ind rdfs:label ?indLabel .
  ?item wdt:P452 ?ind .
  ?item wikibase:sitelinks ?sl .
  FILTER(?sl >= {BOOST_MIN_SITELINKS})
  ?item wdt:P31 ?cls .
  {class_values()}
  {WEBSITE_BLOCK}
}} ORDER BY DESC(?sl) LIMIT {INDUSTRY_BOOST_LIMIT}"""


def q_dissolved(qids: list[str]) -> str:
    """Only items that have a dissolution / abolition date (P576) or an end time (P582) as a main statement come back."""
    return f"""SELECT ?item ?dissolved WHERE {{ {values_clause("item", qids)} ?item wdt:P576|wdt:P582 ?dissolved . }}"""


def q_classes(qids: list[str]) -> str:
    """Every P31 class of the items — the non-company exclusion in `assemble` needs more than the class a candidate was matched on."""
    return f"""SELECT ?item ?cls WHERE {{ {values_clause("item", qids)} ?item wdt:P31 ?cls . }}"""


def q_labels(qids: list[str]) -> str:
    return f"""SELECT ?item ?itemLabel ?itemDescription ?itemAltLabel WHERE {{
  {values_clause("item", qids)}
  SERVICE wikibase:label {{ bd:serviceParam wikibase:language "en". }}
}}"""


# Plain (non-aggregated) detail queries, reduced client-side: a GROUP BY with a dozen SAMPLE() aggregates over nested OPTIONALs makes
# Blazegraph throw StackOverflowError.
def q_misc(qids: list[str]) -> str:
    return f"""SELECT ?item ?inception ?coord ?lei ?cik ?logo WHERE {{
  {values_clause("item", qids)}
  OPTIONAL {{ ?item wdt:P571 ?inception }}
  OPTIONAL {{ ?item wdt:P625 ?coord }}
  OPTIONAL {{ ?item wdt:P1278 ?lei }}
  OPTIONAL {{ ?item wdt:P5531 ?cik }}
  OPTIONAL {{ ?item wdt:P154 ?logo }}
}}"""


def q_hq_parent(qids: list[str]) -> str:
    return f"""SELECT ?item ?hq ?hqEnd ?hqLabel ?hqcoord ?hqRegionLabel ?hqiso ?parent ?parentEnd ?parentLabel WHERE {{
  {values_clause("item", qids)}
  OPTIONAL {{
    ?item p:P159 ?hqs . ?hqs ps:P159 ?hq .
    OPTIONAL {{ ?hqs pq:P582 ?hqEnd }}
    OPTIONAL {{ ?hq rdfs:label ?hqLabel FILTER(LANG(?hqLabel) = "en") }}
    OPTIONAL {{ ?hq wdt:P625 ?hqcoord }}
    OPTIONAL {{ ?hq wdt:P131 ?hqRegion . ?hqRegion rdfs:label ?hqRegionLabel FILTER(LANG(?hqRegionLabel) = "en") }}
    OPTIONAL {{ ?hq wdt:P17 ?hqc . ?hqc wdt:P297 ?hqiso }}
  }}
  OPTIONAL {{
    ?item p:P749 ?ps . ?ps ps:P749 ?parent .
    OPTIONAL {{ ?ps pq:P582 ?parentEnd }}
    OPTIONAL {{ ?parent rdfs:label ?parentLabel FILTER(LANG(?parentLabel) = "en") }}
  }}
}}"""


def q_names_industries(qids: list[str]) -> str:
    return f"""SELECT ?item ?legal ?indLabel WHERE {{
  {values_clause("item", qids)}
  OPTIONAL {{ ?item wdt:P1448 ?legal }}
  OPTIONAL {{ ?item wdt:P452 ?ind . ?ind rdfs:label ?indLabel FILTER(LANG(?indLabel) = "en") }}
}}"""


def q_tickers_employees(qids: list[str]) -> str:
    return f"""SELECT ?item ?ticker ?exchangeLabel ?ticker2 ?employees ?empDate WHERE {{
  {values_clause("item", qids)}
  OPTIONAL {{
    ?item p:P414 ?exs . ?exs ps:P414 ?exchange . FILTER NOT EXISTS {{ ?exs pq:P582 [] }}
    OPTIONAL {{ ?exs pq:P249 ?ticker }}
    OPTIONAL {{ ?exchange rdfs:label ?exchangeLabel FILTER(LANG(?exchangeLabel) = "en") }}
  }}
  OPTIONAL {{ ?item wdt:P249 ?ticker2 }}
  OPTIONAL {{ ?item p:P1128 ?es . ?es ps:P1128 ?employees . OPTIONAL {{ ?es pq:P585 ?empDate }} }}
}}"""


# ------------------------------------------------------------------------------------------------------------ helpers
def qid_of(uri: str) -> str:
    return uri.rsplit("/", 1)[-1]


def load_countries() -> dict[str, dict[str, str]]:
    with COUNTRIES_CSV.open(encoding="utf-8") as f:
        return {r["code"]: r for r in csv.DictReader(f)}


def normalise_website(url: str) -> tuple[str, str] | None:
    """→ (website `https://host`, registrable domain) or None when unusable/generic."""
    url = (url or "").strip()
    if not url:
        return None
    if "://" not in url:
        url = "https://" + url
    try:
        p = urlparse(url)
    except ValueError:
        return None
    host = (p.hostname or "").lower().strip().rstrip(".")
    if not host or "." not in host or re.fullmatch(r"[\d.]+", host) or ":" in host:
        return None
    if p.scheme not in ("http", "https"):
        return None
    try:
        host.encode("idna")
    except UnicodeError:
        return None
    dom = registrable_domain(host)
    if not dom or "." not in dom:
        return None
    path = (p.path or "/").rstrip("/")
    if dom in PLATFORM_ROOTS:
        canonical_host = PLATFORM_ROOTS[dom]
        if host not in (canonical_host, dom, "www." + dom) or path:
            return None
        return f"https://{canonical_host}", dom
    if dom in GENERIC_DOMAINS or host in GENERIC_DOMAINS:
        return None
    if host.endswith((".blogspot.com", ".wordpress.com", ".github.io", ".wixsite.com", ".weebly.com", ".tumblr.com")):
        return None
    return f"https://{host}", dom


def parse_point(value: str | None) -> tuple[float, float] | None:
    if not value or not value.startswith("Point("):
        return None
    try:
        lon, lat = value[6:-1].split()
        return round(float(lat), 5), round(float(lon), 5)
    except ValueError:
        return None


def parse_year(value: str | None) -> int | None:
    if not value:
        return None
    m = re.match(r"^(-?\d{1,4})-", value)
    if not m:
        return None
    y = int(m.group(1))
    return y if 1000 <= y <= datetime.now(UTC).year else None


def parse_int(value: str | None) -> int | None:
    try:
        return int(float(value)) if value not in (None, "") else None
    except ValueError:
        return None


def commons_url(filename: str | None) -> str | None:
    """P154 arrives as an already-encoded Special:FilePath URL (or a bare file name): decode, then encode once."""
    if not filename:
        return None
    name = unquote(filename.rsplit("/", 1)[-1]).replace(" ", "_")
    return f"https://commons.wikimedia.org/wiki/Special:FilePath/{quote(name)}"


# Home exchanges per country: preferred listing when a company has several tickers (Wikidata order is arbitrary).
HOME_EXCHANGES: dict[str, tuple[str, ...]] = {
    "US": ("NASDAQ", "Nasdaq", "New York Stock Exchange", "NYSE"), "GB": ("London Stock Exchange",), "JP": ("Tokyo Stock Exchange",),
    "DE": ("Frankfurt Stock Exchange", "Xetra", "Börse"), "FR": ("Euronext Paris", "Paris"), "CA": ("Toronto Stock Exchange", "TSX Venture"),
    "KR": ("Korea Exchange", "KOSDAQ"), "IN": ("National Stock Exchange of India", "Bombay Stock Exchange"), "AU": ("Australian Securities Exchange",),
    "CN": ("Shanghai Stock Exchange", "Shenzhen Stock Exchange"), "HK": ("Hong Kong Stock Exchange",), "TW": ("Taiwan Stock Exchange", "Taipei"),
    "CH": ("SIX Swiss Exchange",), "NL": ("Euronext Amsterdam",), "BE": ("Euronext Brussels",), "SE": ("Nasdaq Stockholm", "Stockholm"),
    "FI": ("Nasdaq Helsinki", "Helsinki"), "DK": ("Nasdaq Copenhagen", "Copenhagen"), "NO": ("Oslo",), "IT": ("Borsa Italiana", "Euronext Milan"),
    "ES": ("Bolsa de Madrid", "Madrid"), "PT": ("Euronext Lisbon",), "BR": ("B3", "São Paulo"), "MX": ("Mexican Stock Exchange", "Bolsa Mexicana"),
    "SG": ("Singapore Exchange",), "ID": ("Indonesia Stock Exchange",), "MY": ("Bursa Malaysia",), "TH": ("Stock Exchange of Thailand",),
    "ZA": ("Johannesburg Stock Exchange", "JSE"), "SA": ("Saudi Stock Exchange", "Tadawul"), "AE": ("Dubai Financial Market", "Abu Dhabi"),
    "IL": ("Tel Aviv Stock Exchange",), "RU": ("Moscow Exchange",), "PL": ("Warsaw Stock Exchange",), "AT": ("Vienna Stock Exchange", "Wiener Börse"),
    "IE": ("Euronext Dublin", "Irish Stock Exchange"), "NZ": ("New Zealand Exchange", "NZX"), "AR": ("Buenos Aires",), "CL": ("Santiago",),
    "TR": ("Borsa Istanbul", "Istanbul"), "EG": ("Egyptian Exchange",), "NG": ("Nigerian", "Nigeria"), "PH": ("Philippine Stock Exchange",),
    "VN": ("Ho Chi Minh", "Hanoi"), "PK": ("Pakistan Stock Exchange",), "GR": ("Athens Stock Exchange",), "CZ": ("Prague Stock Exchange",),
    "HU": ("Budapest Stock Exchange",), "LU": ("Luxembourg Stock Exchange",), "KZ": ("Kazakhstan",), "QA": ("Qatar Stock Exchange",),
    "KW": ("Boursa Kuwait", "Kuwait"), "CO": ("Colombia",), "PE": ("Lima",),
}


def pick_ticker(tickers: list[list[str | None]], country: str | None) -> tuple[str | None, str | None]:
    """(ticker, exchange): home exchange of the company's country first, then any listing with an alphabetic ticker, then the first."""
    if not tickers:
        return None, None
    home = HOME_EXCHANGES.get(country or "", ())
    for t, ex in tickers:
        if ex and any(h.lower() in ex.lower() for h in home):
            return t, ex
    for t, ex in tickers:
        if t and not t.isdigit():
            return t, ex
    return tickers[0][0], tickers[0][1]


DOMAIN_LIKE_RE = re.compile(r"^(https?://)?[\w-]+(\.[\w-]+)+(/.*)?$")
JUNK_LABEL_RE = re.compile(r"classification|category|list of|wikimedia", re.IGNORECASE)


def dump_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")


def load_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


STATS_PATH = DATA_DIR / "harvest-stats.json"


def update_stats(**sections: Any) -> dict[str, Any]:
    """Per-stage counters kept across runs (each stage overwrites its own section); rendered as the README harvest report."""
    stats = load_json(STATS_PATH) if STATS_PATH.exists() else {}
    stats.update(sections)
    dump_json(STATS_PATH, stats)
    return stats


# ------------------------------------------------------------------------------------------------------------ stage: candidates
def new_candidate() -> dict[str, Any]:
    return {"sitelinks": 0, "websites": [], "isos": [], "classes": [], "boost_countries": [], "boost_industries": [], "bands": [],
            "listed": False, "exchanges": [], "employees_hint": None, "has_revenue": False}


def stage_candidates(sp: Sparql, *, refresh: bool = False) -> dict[str, dict[str, Any]]:
    out_path = DATA_DIR / "candidates.json"
    cands: dict[str, dict[str, Any]] = {}
    dissolved: set[str] = set()
    band_rows: Counter[str] = Counter()          # rows returned per band family (for the harvest report)
    band_new: Counter[str] = Counter()           # candidates first seen in each band family
    exchange_labels: dict[str, str] = {}

    def add(row: dict[str, str], *, band: str, cls: str | None = None, boost_country: str | None = None,
            boost_industry: str | None = None) -> None:
        band_rows[band] += 1
        if row.get("dissolved"):
            dissolved.add(qid_of(row["item"]))
            return
        qid = qid_of(row["item"])
        if qid not in cands:
            band_new[band] += 1
        c = cands.setdefault(qid, new_candidate())
        c["sitelinks"] = max(c["sitelinks"], parse_int(row.get("sl")) or 0)
        if row.get("web") and row["web"] not in c["websites"]:
            c["websites"].append(row["web"])
        iso = (row.get("iso") or "").upper()
        if iso and iso not in c["isos"]:
            c["isos"].append(iso)
        if cls and cls not in c["classes"]:
            c["classes"].append(cls)
        if boost_country and boost_country not in c["boost_countries"]:
            c["boost_countries"].append(boost_country)
        if boost_industry and boost_industry not in c["boost_industries"]:
            c["boost_industries"].append(boost_industry)
        if band not in c["bands"]:
            c["bands"].append(band)
        if band == "listed" and row.get("ex") and not row.get("end"):
            c["listed"] = True
            ex_label = exchange_labels.get(qid_of(row["ex"]), qid_of(row["ex"]))
            if ex_label not in c["exchanges"]:
                c["exchanges"].append(ex_label)
        if band == "ticker" and row.get("ticker"):
            c["listed"] = True
        if band == "employees" and parse_int(row.get("val")):
            c["employees_hint"] = max(c["employees_hint"] or 0, parse_int(row["val"]) or 0)
        if band == "revenue":
            c["has_revenue"] = True

    def run_split(build: Any, label: str, handle: Any, suffix: str = "") -> None:
        """Run `build(extra)`; when the endpoint times out, split the query ten ways on the last QID digit (then the last two)."""
        try:
            rows = sp.query(build(suffix_filter(suffix)), label=label + (f" qid…{suffix}" if suffix else ""))
        except TimeoutError:
            if len(suffix) >= SPLIT_DEPTH_MAX:
                raise
            log.warning("splitting %s by QID suffix", label + (f" …{suffix}" if suffix else ""))
            for d in "0123456789":
                run_split(build, label, handle, d + suffix)
            return
        for r in rows:
            handle(r)

    def run_band(cls: str, lo: int, hi: int | None) -> None:
        label = f"class {cls} {CLASSES[cls]} sl[{lo},{hi or '∞'})"
        try:
            rows = sp.query(q_class_band(cls, lo, hi), label=label)
        except TimeoutError:
            if hi is None:
                hi = 400
            if hi - lo <= 1:
                log.warning("band %s too large even at width 1 — splitting by QID suffix", label)
                run_split(lambda extra: q_class_band(cls, lo, hi, extra), label, lambda r: add(r, band="class", cls=cls))
                return
            mid = (lo + hi) // 2
            log.warning("splitting band %s → [%d,%d) [%d,%d)", label, lo, mid, mid, hi)
            run_band(cls, lo, mid)
            run_band(cls, mid, hi)
            return
        for r in rows:
            add(r, band="class", cls=cls)

    # 1. Company classes × sitelink bands (≥ 5 as in the first harvest, then the (3, 5) band for every class).
    for cls in CLASSES:
        if cls in BANDED_CLASSES:
            for lo, hi in SITELINK_BANDS:
                run_band(cls, lo, hi)
        else:
            run_band(cls, MIN_SITELINKS, None)
            run_band(cls, *LOW_BAND)
        log.info("candidates so far: %d", len(cands))
    # 2. Every listed company: one scan per large exchange, small exchanges grouped (≈ EXCHANGE_GROUP_MAX statements per query).
    exchanges = sp.query(q_exchanges(), label="exchanges")
    exchange_labels.update({qid_of(r["ex"]): r.get("exLabel") or qid_of(r["ex"]) for r in exchanges})
    groups: list[list[str]] = []
    current: list[str] = []
    current_n = 0
    for r in exchanges:
        n = parse_int(r.get("n")) or 0
        if n >= EXCHANGE_GROUP_MAX:
            groups.append([qid_of(r["ex"])])
            continue
        if current and current_n + n > EXCHANGE_GROUP_MAX:
            groups.append(current)
            current, current_n = [], 0
        current.append(qid_of(r["ex"]))
        current_n += n
    if current:
        groups.append(current)
    for i, group in enumerate(groups, 1):
        names = ", ".join(exchange_labels.get(q, q) for q in group[:3]) + (f" +{len(group) - 3}" if len(group) > 3 else "")
        run_split(lambda extra, g=group: q_listed(g, extra), f"listed {i}/{len(groups)} {names}", lambda r: add(r, band="listed"))
    run_split(q_ticker_only, "ticker-only P249", lambda r: add(r, band="ticker"))
    log.info("candidates after listed companies: %d (%d listed)", len(cands), sum(1 for c in cands.values() if c["listed"]))
    # 3. Large companies by employees (P1128 ≥ EMPLOYEES_MIN) or with a revenue statement (P2139), per class, any sitelink count.
    for cls, cls_name in CLASSES.items():
        run_split(lambda extra, c=cls: q_class_measure(c, "P1128", EMPLOYEES_MIN, extra), f"employees≥{EMPLOYEES_MIN} {cls} {cls_name}",
                  lambda r, c=cls: add(r, band="employees", cls=c))
        run_split(lambda extra, c=cls: q_class_measure(c, "P2139", None, extra), f"revenue {cls} {cls_name}",
                  lambda r, c=cls: add(r, band="revenue", cls=c))
    log.info("candidates after employees/revenue bands: %d", len(cands))
    # 4. Boosts (country-first scans only where a minimum-coverage country is still short; industry label scans always).
    iso_to_qid = {r["iso"]: qid_of(r["country"]) for r in sp.query(q_country_qids(list(COUNTRY_MINIMUMS)), label="country qids")}
    per_country: Counter[str] = Counter(c["isos"][0] for c in cands.values() if c["isos"])
    boost_failures: list[str] = []
    for iso, minimum in COUNTRY_MINIMUMS.items():
        if per_country[iso] >= minimum * BOOST_MARGIN or iso not in iso_to_qid:
            log.info("country %s: %d candidates (min %d) — no boost", iso, per_country[iso], minimum)
            continue
        try:
            rows = sp.query(q_country_boost(iso, iso_to_qid[iso]), label=f"country boost {iso}")
        except (RuntimeError, TimeoutError) as e:
            log.warning("country boost %s failed (%s) — known gap, see README", iso, e)
            boost_failures.append(iso)
            continue
        for r in rows:
            add(r, band="country-boost", boost_country=iso)
    for slug, labels in INDUSTRY_BOOST_LABELS.items():
        for r in sp.query(q_industry_boost(labels), label=f"industry boost {slug}"):
            add(r, band="industry-boost", boost_industry=slug)
    for qid in dissolved:
        cands.pop(qid, None)
    dump_json(out_path, cands)
    update_stats(candidates={
        "total": len(cands), "dissolved_inline": len(dissolved), "listed": sum(1 for c in cands.values() if c["listed"]),
        "classes": len(CLASSES), "exchanges": len(exchanges), "exchange_queries": len(groups), "band_rows": dict(band_rows),
        "band_new": dict(band_new), "country_boost_failures": boost_failures,
        "live_queries": sp.queries, "cached_queries": sp.cached, "at": datetime.now(UTC).replace(microsecond=0).isoformat(),
    })
    log.info("candidates: %d (%d dissolved dropped; bands %s) → %s", len(cands), len(dissolved), dict(band_new), out_path)
    return cands


# ------------------------------------------------------------------------------------------------------------ stage: dissolved
def stage_dissolved(sp: Sparql, *, batch: int = 300) -> set[str]:
    """P576 (dissolved/abolished) check over every candidate — the candidate queries cannot afford a NOT EXISTS filter."""
    cands = load_json(DATA_DIR / "candidates.json")
    qids = sorted(cands, key=lambda q: int(q[1:]))
    dissolved: dict[str, str] = {}
    for i in range(0, len(qids), batch):
        chunk = qids[i:i + batch]
        for r in sp.query(q_dissolved(chunk), label=f"dissolved {i + len(chunk)}/{len(qids)}"):
            dissolved.setdefault(qid_of(r["item"]), r.get("dissolved") or "")
    dump_json(DATA_DIR / "dissolved.json", dissolved)
    update_stats(dissolved={"candidates": len(qids), "dissolved": len(dissolved), "queries": (len(qids) + batch - 1) // batch})
    log.info("dissolved: %d of %d candidates → %s", len(dissolved), len(qids), DATA_DIR / "dissolved.json")
    return set(dissolved)


# ------------------------------------------------------------------------------------------------------------ stage: select
def prefilter(cands: dict[str, dict[str, Any]], countries: dict[str, dict[str, str]]) -> tuple[list[dict[str, Any]], dict[str, int]]:
    """Normalise websites, pick one country, drop generic hosts and duplicate registrable domains (highest sitelinks wins)."""
    stats: Counter[str] = Counter()
    rows: list[dict[str, Any]] = []
    for qid, c in cands.items():
        picked = None
        for w in c["websites"]:
            picked = normalise_website(w)
            if picked:
                break
        if not picked:
            stats["dropped_generic_or_invalid_website"] += 1
            continue
        website, dom = picked
        isos = [i for i in c["isos"] if i in countries]
        country = isos[0] if isos else None
        rows.append({"wikidata_id": qid, "sitelinks": c["sitelinks"], "website": website, "canonical_domain": dom, "country": country,
                     "country_candidates": isos, "classes": c["classes"], "boost_industries": c["boost_industries"],
                     "bands": c.get("bands") or [], "listed": bool(c.get("listed")), "exchanges": c.get("exchanges") or [],
                     "employees_hint": c.get("employees_hint"), "has_revenue": bool(c.get("has_revenue"))})
    rows.sort(key=lambda r: (-r["sitelinks"], int(r["wikidata_id"][1:])))
    by_domain: dict[str, dict[str, Any]] = {}
    for r in rows:
        keeper = by_domain.get(r["canonical_domain"])
        if keeper is None:
            by_domain[r["canonical_domain"]] = r
        else:
            keeper.setdefault("domain_conflicts", []).append(r["wikidata_id"])
            stats["dropped_duplicate_domain"] += 1
    kept = list(by_domain.values())
    stats["kept"] = len(kept)
    return kept, dict(stats)


def eligible(r: dict[str, Any], min_sitelinks: int) -> bool:
    """Listed companies and large companies (employees / revenue bands) qualify at any sitelink count; class-band and boost
    candidates need `min_sitelinks`."""
    if r.get("listed") or r.get("employees_hint") or r.get("has_revenue"):
        return True
    return r["sitelinks"] >= min_sitelinks


def select_diversified(rows: list[dict[str, Any]], *, target: int, countries: dict[str, dict[str, str]]) -> tuple[list[dict[str, Any]], dict[str, int]]:
    """Diversified selection: country minimums first, then every listed company, then the rest in sitelink order (≥ 3, then ≥ 2 if the
    target is still short) — all under the country and class caps. Rows must be sorted by sitelinks desc."""
    caps: dict[str | None, int] = defaultdict(lambda: int(target * OTHER_CAP_SHARE))
    caps["US"] = int(target * US_CAP_SHARE)
    caps[None] = int(target * UNKNOWN_COUNTRY_SHARE)
    chosen: dict[str, dict[str, Any]] = {}
    per_country: Counter[str | None] = Counter()
    per_class: Counter[str] = Counter()
    passes: dict[str, int] = {}

    def take(r: dict[str, Any]) -> None:
        chosen[r["wikidata_id"]] = r
        per_country[r["country"]] += 1
        for c in r.get("classes", []):
            per_class[c] += 1

    def class_capped(r: dict[str, Any]) -> bool:
        return any(per_class[c] >= int(target * CLASS_CAPS[c]) for c in r.get("classes", []) if c in CLASS_CAPS)

    def sweep(name: str, pool: list[dict[str, Any]], min_sitelinks: int) -> None:
        before = len(chosen)
        for r in pool:
            if len(chosen) >= target:
                break
            if r["wikidata_id"] in chosen or not eligible(r, min_sitelinks):
                continue
            if per_country[r["country"]] >= caps[r["country"]] or class_capped(r):
                continue
            take(r)
        passes[name] = len(chosen) - before

    by_country: dict[str | None, list[dict[str, Any]]] = defaultdict(list)
    for r in rows:
        by_country[r["country"]].append(r)
    for iso, minimum in COUNTRY_MINIMUMS.items():
        for r in [x for x in by_country.get(iso, []) if eligible(x, FALLBACK_MIN_SITELINKS)][:minimum]:
            take(r)
    passes["country_minimums"] = len(chosen)
    sweep("listed", [r for r in rows if r.get("listed")], SELECT_MIN_SITELINKS)
    sweep(f"sitelinks_ge_{SELECT_MIN_SITELINKS}", rows, SELECT_MIN_SITELINKS)
    if len(chosen) < target:
        sweep(f"sitelinks_ge_{FALLBACK_MIN_SITELINKS}", rows, FALLBACK_MIN_SITELINKS)
    return list(chosen.values()), passes


def stage_select(*, target: int) -> list[dict[str, Any]]:
    countries = load_countries()
    cands = load_json(DATA_DIR / "candidates.json")
    dissolved_path = DATA_DIR / "dissolved.json"
    if dissolved_path.exists():
        dissolved = set(load_json(dissolved_path))
        cands = {q: c for q, c in cands.items() if q not in dissolved}
        log.info("select: %d dissolved candidates excluded", len(dissolved))
    else:
        log.warning("select: no dissolved.json — run the `dissolved` stage first to exclude defunct companies")
    kept, stats = prefilter(cands, countries)
    log.info("prefilter: %s", stats)
    selected, passes = select_diversified(kept, target=target, countries=countries)
    log.info("selection passes: %s", passes)
    stats["listed_after_prefilter"] = sum(1 for r in kept if r["listed"])
    stats["listed_selected"] = sum(1 for r in selected if r["listed"])
    stats["passes"] = passes
    stats["target"] = target
    # Pool for the industry guarantee: best boost candidates per industry, fetched in the details stage too (≤ INDUSTRY_RESERVE each).
    chosen_ids = {r["wikidata_id"] for r in selected}
    reserve: list[dict[str, Any]] = []
    per_ind: Counter[str] = Counter()
    for r in kept:
        if r["wikidata_id"] in chosen_ids:
            continue
        for slug in r["boost_industries"]:
            if per_ind[slug] < INDUSTRY_RESERVE:
                per_ind[slug] += 1
                reserve.append(r)
                break
    dump_json(DATA_DIR / "selected.json", {"selected": selected, "reserve": reserve, "stats": stats, "target": target})
    update_stats(select=stats)
    log.info("selected %d (+%d reserve for industry minimums) → %s", len(selected), len(reserve), DATA_DIR / "selected.json")
    return selected


# ------------------------------------------------------------------------------------------------------------ stage: details
def stage_details(sp: Sparql, *, batch: int = 100) -> dict[str, dict[str, Any]]:
    sel = load_json(DATA_DIR / "selected.json")
    qids = [r["wikidata_id"] for r in sel["selected"]] + [r["wikidata_id"] for r in sel["reserve"]]
    out_path = DATA_DIR / "details.json"
    details: dict[str, dict[str, Any]] = load_json(out_path) if out_path.exists() else {}
    todo = [q for q in qids if q not in details]
    log.info("details: %d to fetch (%d cached)", len(todo), len(qids) - len(todo))
    for i in range(0, len(todo), batch):
        chunk = todo[i:i + batch]
        tag = f"details {i + len(chunk)}/{len(todo)}"
        d: dict[str, dict[str, Any]] = {q: {"legal_names": [], "industry_labels": [], "tickers": [], "employees_obs": []} for q in chunk}
        for r in sp.query(q_labels(chunk), label=f"{tag} labels"):
            q = qid_of(r["item"])
            d[q]["label"] = r.get("itemLabel")
            d[q]["description"] = r.get("itemDescription")
            d[q]["alt_labels"] = [a.strip() for a in (r.get("itemAltLabel") or "").split(",") if a.strip()]
        for r in sp.query(q_misc(chunk), label=f"{tag} misc"):
            m = d[qid_of(r["item"])]
            for key in ("inception", "coord", "lei", "cik", "logo"):
                if r.get(key) and (not m.get(key) or (key == "inception" and r[key] < m[key])):
                    m[key] = r[key]
        grouped: dict[str, list[dict[str, str]]] = defaultdict(list)
        for r in sp.query(q_hq_parent(chunk), label=f"{tag} hq/parent"):
            grouped[qid_of(r["item"])].append(r)
        for q, rs in grouped.items():
            hqs = [r for r in rs if r.get("hq")]
            current = [r for r in hqs if not r.get("hqEnd")] or hqs
            if current:
                r = current[0]
                d[q].update({"hq": qid_of(r["hq"]), "hq_label": r.get("hqLabel"), "hq_coord": r.get("hqcoord"),
                             "hq_region": r.get("hqRegionLabel"), "hq_iso": r.get("hqiso")})
            parents = [r for r in rs if r.get("parent") and not r.get("parentEnd")]
            if parents:
                d[q].update({"parent": qid_of(parents[0]["parent"]), "parent_label": parents[0].get("parentLabel")})
        for r in sp.query(q_names_industries(chunk), label=f"{tag} names/industries"):
            q = qid_of(r["item"])
            if r.get("legal") and r["legal"] not in d[q]["legal_names"]:
                d[q]["legal_names"].append(r["legal"])
            if r.get("indLabel") and r["indLabel"] not in d[q]["industry_labels"]:
                d[q]["industry_labels"].append(r["indLabel"])
        for r in sp.query(q_tickers_employees(chunk), label=f"{tag} tickers/employees"):
            q = qid_of(r["item"])
            t = r.get("ticker") or r.get("ticker2")
            if t:
                pair = [t, r.get("exchangeLabel")]
                if pair not in d[q]["tickers"]:
                    d[q]["tickers"].append(pair)
            if r.get("employees"):
                obs = [r["employees"], r.get("empDate")]
                if obs not in d[q]["employees_obs"]:
                    d[q]["employees_obs"].append(obs)
        details.update(d)
        dump_json(out_path, details)
    # Full P31 class lists (300 QIDs per query) for every item that lacks them — feeds the non-company exclusion in `assemble`.
    todo_cls = [q for q in qids if q in details and "p31" not in details[q]]
    log.info("details: P31 classes to fetch for %d items", len(todo_cls))
    for i in range(0, len(todo_cls), CLASS_BATCH):
        chunk = todo_cls[i:i + CLASS_BATCH]
        for q in chunk:
            details[q]["p31"] = []
        for r in sp.query(q_classes(chunk), label=f"classes {i + len(chunk)}/{len(todo_cls)}"):
            q, c = qid_of(r["item"]), qid_of(r["cls"])
            if c not in details[q]["p31"]:
                details[q]["p31"].append(c)
        if (i // CLASS_BATCH) % 10 == 9:
            dump_json(out_path, details)
    dump_json(out_path, details)
    log.info("details: %d entries → %s", len(details), out_path)
    return details


# ------------------------------------------------------------------------------------------------------------ stage: assemble
def importance_score(sitelinks: int, employees: int | None, public: bool, ticker: str | None) -> float:
    s_sl = min(1.0, math.log1p(max(sitelinks, 0)) / math.log1p(300))
    s_emp = min(1.0, math.log10((employees or 0) + 1) / 6) if employees else 0.0
    score = 0.6 * s_sl + 0.2 * s_emp + 0.1 * (1 if public else 0) + 0.1 * (1 if ticker else 0)
    return round(min(1.0, max(0.02, score)), 4)


def tier_cutoffs(n: int) -> tuple[int, int, int]:
    """Cumulative rank cut-offs of tiers 1–3 for a universe of `n` companies (TIER_SHARES quantiles; ≈ 300 / 2,300 / 10,300 at 30 k)."""
    t1 = round(n * TIER_SHARES[0])
    t2 = t1 + round(n * TIER_SHARES[1])
    t3 = t2 + round(n * TIER_SHARES[2])
    return t1, t2, t3


def assign_tiers(rows: list[dict[str, Any]]) -> None:
    rows.sort(key=lambda r: (-r["importance"], -r["sitelinks"], r["wikidata_id"]))
    t1, t2, t3 = tier_cutoffs(len(rows))
    for i, r in enumerate(rows):
        r["tier"] = 1 if i < t1 else 2 if i < t2 else 3 if i < t3 else 4


def build_row(sel: dict[str, Any], d: dict[str, Any], harvested_at: str) -> dict[str, Any]:
    label = d.get("label") or ""
    if not label or re.fullmatch(r"Q\d+", label):
        label = (d.get("legal_names") or [sel["canonical_domain"]])[0]
    legal = next((x for x in d.get("legal_names", []) if re.search(r"[A-Za-z]", x)), None) or (d.get("legal_names") or [None])[0]
    aliases = [a for a in d.get("alt_labels", []) if a and a not in (label, legal) and len(a) <= 80 and not DOMAIN_LIKE_RE.match(a)][:8]
    emp = None
    obs = d.get("employees_obs") or []
    if obs:
        obs = sorted(obs, key=lambda o: (o[1] or ""), reverse=True)
        emp = parse_int(obs[0][0])
    if emp is None and sel.get("employees_hint"):
        emp = parse_int(str(sel["employees_hint"]))
    coord = parse_point(d.get("coord")) or parse_point(d.get("hq_coord"))
    country = sel["country"]
    if d.get("hq_iso") and d["hq_iso"] in (sel.get("country_candidates") or []):
        country = d["hq_iso"]
    ticker, exchange = pick_ticker(d.get("tickers") or [], country)
    if not exchange and sel.get("listed") and sel.get("exchanges"):
        exchange = sel["exchanges"][0]          # listing without a ticker qualifier on Wikidata: keep the exchange, ticker stays null
    public = "Q891723" in sel.get("classes", []) or bool(sel.get("listed")) or bool(ticker) or bool(exchange)
    industry_labels = [x for x in (d.get("industry_labels") or []) if not JUNK_LABEL_RE.search(x)]
    slugs = map_industry(industry_labels)
    if not slugs and sel.get("boost_industries"):
        slugs = [sel["boost_industries"][0]]
    if not slugs:
        slugs = map_industry([label, d.get("description") or ""] + [CLASSES.get(c, "") for c in sel.get("classes", [])], limit=2)
    return {
        "wikidata_id": sel["wikidata_id"], "display_name": label.strip(), "legal_name": legal, "aliases": aliases, "website": sel["website"],
        "canonical_domain": sel["canonical_domain"], "country": country, "hq_city": d.get("hq_label"), "hq_region": d.get("hq_region"),
        "lat": coord[0] if coord else None, "lon": coord[1] if coord else None, "industries": slugs, "industry_labels": industry_labels[:8],
        "founded_year": parse_year(d.get("inception")), "employees": emp, "public_company": public, "ticker": ticker, "exchange": exchange,
        "lei": d.get("lei"), "sec_cik": str(d["cik"]).lstrip("0") or None if d.get("cik") else None,
        "parent": {"wikidata_id": d["parent"], "name": d.get("parent_label")} if d.get("parent") else None,
        "logo_url": commons_url(d.get("logo")), "description": (d.get("description") or None), "sitelinks": sel["sitelinks"],
        "importance": importance_score(sel["sitelinks"], emp, public, ticker), "tier": 4, "source": "wikidata", "harvested_at": harvested_at,
        "domain_conflicts": sel.get("domain_conflicts") or [],
    }


def non_company_reason(sel: dict[str, Any], d: dict[str, Any], row: dict[str, Any]) -> str | None:
    """Why an item is not a company (None when it is one): a NON_COMPANY_CLASSES P31 class, or a museum/university/club/… description
    without any ticker, exchange, employee count or revenue. A current stock-exchange listing or a ticker always keeps the item."""
    if sel.get("listed") or row.get("ticker") or row.get("exchange"):
        return None
    bad = [c for c in (d.get("p31") or []) if c in NON_COMPANY_CLASSES]
    if bad:
        return f"non_company_class:{bad[0]}"
    desc = row.get("description") or ""
    if NON_COMPANY_DESC_RE.search(desc) and not COMPANY_DESC_RE.search(desc) and not row.get("employees") and not sel.get("has_revenue"):
        return "non_company_description"
    return None


def stage_assemble() -> list[dict[str, Any]]:
    countries = load_countries()
    sel = load_json(DATA_DIR / "selected.json")
    details = load_json(DATA_DIR / "details.json")
    harvested_at = datetime.now(UTC).replace(microsecond=0).isoformat()
    selected = [r for r in sel["selected"] if r["wikidata_id"] in details]
    reserve = [r for r in sel["reserve"] if r["wikidata_id"] in details]
    dropped: list[dict[str, Any]] = []
    non_company: Counter[str] = Counter()

    def build_companies(sels: list[dict[str, Any]]) -> list[dict[str, Any]]:
        out = []
        for s in sels:
            row = build_row(s, details[s["wikidata_id"]], harvested_at)
            reason = non_company_reason(s, details[s["wikidata_id"]], row)
            if reason:
                non_company[reason] += 1
                dropped.append({"wikidata_id": s["wikidata_id"], "reason": reason, "name": row["display_name"], "description": row["description"]})
                continue
            out.append(row)
        return out

    rows = build_companies(selected)
    selected_kept = len(rows)
    # Industry minimums: top up from the reserve where a top-level industry is below the floor.
    counts: Counter[str] = Counter()
    for r in rows:
        for top in {top_level_of(s) for s in r["industries"]}:
            counts[top] += 1
    chosen = {r["wikidata_id"] for r in rows}
    reserve_rows = build_companies(reserve)
    reserve_rows.sort(key=lambda r: -r["sitelinks"])
    for top in top_level_slugs():
        for rr in reserve_rows:
            if counts[top] >= INDUSTRY_MINIMUM:
                break
            if rr["wikidata_id"] in chosen or top not in {top_level_of(s) for s in rr["industries"]}:
                continue
            rows.append(rr)
            chosen.add(rr["wikidata_id"])
            for t in {top_level_of(s) for s in rr["industries"]}:
                counts[t] += 1
    # Parent/child sharing a domain: parent wins (regardless of sitelinks); duplicate domains (should not happen after prefilter) drop the lower.
    by_domain: dict[str, dict[str, Any]] = {}
    for r in sorted(rows, key=lambda r: -r["sitelinks"]):
        by_domain.setdefault(r["canonical_domain"], r)
    ids = {r["wikidata_id"] for r in by_domain.values()}
    for r in list(by_domain.values()):
        p = r.get("parent")
        if p and p["wikidata_id"] in ids and p["wikidata_id"] != r["wikidata_id"]:
            parent_row = next((x for x in by_domain.values() if x["wikidata_id"] == p["wikidata_id"]), None)
            if parent_row and parent_row["canonical_domain"] == r["canonical_domain"]:
                dropped.append({"wikidata_id": r["wikidata_id"], "reason": "shares_parent_domain", "parent": p["wikidata_id"]})
    for r in rows:
        if r["wikidata_id"] not in ids:
            dropped.append({"wikidata_id": r["wikidata_id"], "reason": "duplicate_domain", "domain": r["canonical_domain"]})
    drop_ids = {x["wikidata_id"] for x in dropped}
    final = [r for r in by_domain.values() if r["wikidata_id"] not in drop_ids]
    for r in final:
        r["notes"] = [{"related_domain_conflict": q} for q in r.pop("domain_conflicts", [])] or []
        if not r["notes"]:
            r.pop("notes")
    assign_tiers(final)
    # Write per region.
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    for old in OUT_DIR.glob("wikidata-*.ndjson"):
        old.unlink()
    by_region: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for r in final:
        region = (countries.get(r["country"] or "", {}).get("region") or "other").lower().replace(" ", "-")
        by_region[region].append(r)
    for region, items in sorted(by_region.items()):
        items.sort(key=lambda r: (r["tier"], -r["importance"], r["wikidata_id"]))
        with (OUT_DIR / f"wikidata-{region}.ndjson").open("w", encoding="utf-8") as f:
            for r in items:
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
    dump_json(DATA_DIR / "dropped.json", dropped)
    by_bad_class = Counter(NON_COMPANY_CLASSES.get(k.split(":", 1)[1], k) for k in non_company.elements() if k.startswith("non_company_class:"))
    stats = update_stats(assemble={
        "companies": len(final), "selected_with_details": len(selected), "industry_top_up": len(rows) - selected_kept,
        "non_company_excluded": sum(non_company.values()), "non_company_by_class": sum(v for k, v in non_company.items() if k != "non_company_description"),
        "non_company_by_description": non_company.get("non_company_description", 0), "non_company_top_classes": dict(by_bad_class.most_common(12)),
        "dropped_shares_parent_domain": sum(1 for x in dropped if x["reason"] == "shares_parent_domain"),
        "dropped_duplicate_domain": sum(1 for x in dropped if x["reason"] == "duplicate_domain"),
        "domain_conflict_notes": sum(len(r.get("notes") or []) for r in final), "tier_cutoffs": tier_cutoffs(len(final)),
        "harvested_at": harvested_at,
    })
    write_readme(final, countries, stats)
    log.info("assembled %d companies into %d region files (%d dropped) → %s", len(final), len(by_region), len(dropped), OUT_DIR)
    return final


def write_readme(rows: list[dict[str, Any]], countries: dict[str, dict[str, str]], stats: dict[str, Any] | None = None) -> None:
    names = {i.slug: i.name for i in load_industries()}
    by_country = Counter(r["country"] or "??" for r in rows)
    by_tier = Counter(r["tier"] for r in rows)
    by_top: Counter[str] = Counter()
    by_ind: Counter[str] = Counter()
    for r in rows:
        for s in r["industries"]:
            by_ind[s] += 1
        for t in {top_level_of(s) for s in r["industries"]}:
            by_top[t] += 1
    n = len(rows)
    no_ind = sum(1 for r in rows if not r["industries"])
    public = sum(1 for r in rows if r["public_company"])
    with_ticker = sum(1 for r in rows if r["ticker"])
    with_exchange = sum(1 for r in rows if r["exchange"])
    with_employees = sum(1 for r in rows if r["employees"])
    region = Counter((countries.get(r["country"] or "", {}).get("region") or "other") for r in rows)
    by_exchange = Counter(r["exchange"] for r in rows if r["exchange"])
    generated = rows[0]["harvested_at"] if rows else ""
    intro = (f"Generated by `scripts/seed_wikidata.py` on {generated}. **{n} companies**, {public} public ({100 * public / n:.1f} %), "
             f"{with_exchange} with a stock exchange, {with_ticker} with a ticker, {with_employees} with an employee count, {no_ind} without an "
             f"industry mapping ({100 * no_ind / n:.1f} %), {len(by_country)} countries.")
    files_note = ("Files: one NDJSON per UN region (`wikidata-<region>.ndjson`), one JSON object per line — see `docs/SEEDS.md` for the "
                  "schema, the diversification rules and how to add companies. `scripts/seed_edgar.py` rewrites the files in place with "
                  "SEC data; run it after every `assemble`.")
    lines = [
        "# Seed registry — Wikidata harvest", "", intro, "", files_note, "",
        "## Tiers", "", "| Tier | Companies |", "|---|---|",
        *[f"| {t} ({ {1: 'global', 2: 'major', 3: 'notable', 4: 'long tail'}[t] }) | {by_tier[t]} |" for t in sorted(by_tier)], "",
        "## Regions", "", "| Region | Companies |", "|---|---|", *[f"| {k} | {v} |" for k, v in region.most_common()], "",
        "## Countries", "", "| Code | Country | Companies | Share |", "|---|---|---|---|",
        *[f"| {c} | {countries.get(c, {}).get('name', 'unknown')} | {cnt} | {100 * cnt / n:.1f} % |" for c, cnt in by_country.most_common()], "",
        "## Stock exchanges (primary listing kept per company)", "", "| Exchange | Companies |", "|---|---|",
        *[f"| {k} | {v} |" for k, v in by_exchange.most_common(40)], "",
        "## Top-level industries (a company counts once per top-level sector)", "", "| Industry | Companies |", "|---|---|",
        *[f"| {names.get(k, k)} | {v} |" for k, v in by_top.most_common()], "",
        "## All industries", "", "| Slug | Companies |", "|---|---|", *[f"| {k} | {v} |" for k, v in by_ind.most_common()], "",
    ]
    lines += harvest_report(stats or {}, rows)
    (OUT_DIR / "README.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


def harvest_report(stats: dict[str, Any], rows: list[dict[str, Any]]) -> list[str]:
    """Human-readable summary of `data/seed/harvest-stats.json` (candidate bands, exclusions, selection passes, tiers)."""
    c = stats.get("candidates") or {}
    d = stats.get("dissolved") or {}
    s = stats.get("select") or {}
    a = stats.get("assemble") or {}
    if not (c or s or a):
        return []
    day = (a.get("harvested_at") or "")[:10]
    lines = [f"## Harvest report — {day}", "",
             ("* Pipeline: `scripts/seed_wikidata.py all --target N` (candidates → dissolved → select → details → assemble), then "
              "`scripts/seed_edgar.py`, then `catlas seed`. Every SPARQL result is cached under `data/seed/wikidata/sparql/`; per-stage "
              "counters in `data/seed/harvest-stats.json`.")]
    if c:
        bands = ", ".join(f"{k} {v:,}" for k, v in sorted((c.get("band_new") or {}).items(), key=lambda kv: -kv[1]))
        lines.append(f"* Candidates: **{c.get('total', 0):,}** unique items with a website from {c.get('classes')} P31 classes × sitelink bands "
                     f"(≥ 3), {c.get('exchanges')} stock exchanges scanned in {c.get('exchange_queries')} P414 queries ({c.get('listed', 0):,} "
                     f"currently listed), employees ≥ {EMPLOYEES_MIN} / revenue bands per class, plus country and industry boosts. "
                     f"First seen per band: {bands}.")
        if c.get("country_boost_failures"):
            lines.append(f"* Country-first boosts cut by the endpoint time limit (tolerated): {', '.join(c['country_boost_failures'])}.")
    if d:
        lines.append(f"* Dissolved check (P576 / P582, {d.get('queries')} queries of 300 QIDs): {d.get('dissolved', 0):,} of "
                     f"{d.get('candidates', 0):,} candidates excluded.")
    if s:
        passes = s.get("passes") or {}
        lines.append(f"* Prefilter: {s.get('dropped_generic_or_invalid_website', 0):,} generic / invalid websites dropped, "
                     f"{s.get('dropped_duplicate_domain', 0):,} duplicate registrable domains (highest sitelinks kept), {s.get('kept', 0):,} kept "
                     f"({s.get('listed_after_prefilter', 0):,} listed).")
        lines.append(f"* Selection (target {s.get('target', 0):,}; caps US ≤ {US_CAP_SHARE:.0%}, other countries ≤ {OTHER_CAP_SHARE:.0%}, "
                     f"unknown country ≤ {UNKNOWN_COUNTRY_SHARE:.0%}, narrow classes 2–5 %): country minimums {passes.get('country_minimums', 0):,}, "
                     f"then {passes.get('listed', 0):,} listed companies, then {passes.get(f'sitelinks_ge_{SELECT_MIN_SITELINKS}', 0):,} by sitelinks "
                     f"≥ {SELECT_MIN_SITELINKS}" + (f", then {passes[f'sitelinks_ge_{FALLBACK_MIN_SITELINKS}']:,} by sitelinks ≥ {FALLBACK_MIN_SITELINKS}"
                                                  if f"sitelinks_ge_{FALLBACK_MIN_SITELINKS}" in passes else "") +
                     f"; {s.get('listed_selected', 0):,} listed companies selected in total.")
    if a:
        t1, t2, t3 = a.get("tier_cutoffs") or (0, 0, 0)
        top = ", ".join(f"{k} {v}" for k, v in (a.get("non_company_top_classes") or {}).items())
        lines.append(f"* Non-company exclusion: **{a.get('non_company_excluded', 0):,} items removed** — {a.get('non_company_by_class', 0):,} for a "
                     f"non-company P31 class (museums, libraries, universities / schools, government agencies, NGOs / charities / foundations / "
                     f"nonprofits, religious organisations, political parties, trade unions, hospitals, sports clubs, YouTube channels; top: {top}) "
                     f"and {a.get('non_company_by_description', 0):,} for a description matching museum / university / school / ministry / agency / "
                     f"charity / foundation / association / club / church / channel with no ticker, exchange, employee count or revenue. "
                     f"Items with a current listing or a ticker are always kept (listed football clubs). List: `data/seed/dropped.json`.")
        lines.append(f"* Assemble: {a.get('selected_with_details', 0):,} selected companies with details + {a.get('industry_top_up', 0):,} from the "
                     f"industry top-up (every top-level industry ≥ {INDUSTRY_MINIMUM}); {a.get('dropped_shares_parent_domain', 0)} subsidiaries "
                     f"sharing their parent's domain and {a.get('dropped_duplicate_domain', 0)} duplicate domains dropped; "
                     f"{a.get('domain_conflict_notes', 0):,} `related_domain_conflict` notes → **{a.get('companies', 0):,} companies**.")
        lines.append(f"* Tiers by importance quantile ({TIER_SHARES[0]:.0%} / {TIER_SHARES[1]:.2%} / {TIER_SHARES[2]:.2%}): tier 1 = ranks 1–{t1:,}, "
                     f"tier 2 → {t2:,}, tier 3 → {t3:,}, tier 4 = rest.")
    mins = {iso: sum(1 for r in rows if r["country"] == iso) for iso in COUNTRY_MINIMUMS}
    short = {iso: n for iso, n in mins.items() if n < COUNTRY_MINIMUMS[iso]}
    lines.append("* Country minimums: " + ("all met" if not short else "short: " + ", ".join(f"{k} {v}/{COUNTRY_MINIMUMS[k]}" for k, v in short.items()))
                 + " (" + ", ".join(f"{k} {v}" for k, v in sorted(mins.items(), key=lambda kv: -kv[1])[:8]) + " …).")
    return lines + [""]


# ------------------------------------------------------------------------------------------------------------ main
def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("stage", nargs="?", default="all", choices=["candidates", "dissolved", "select", "details", "assemble", "all"])
    ap.add_argument("--target", type=int, default=30000, help="companies to select before detail fetching (default 30000)")
    ap.add_argument("--batch", type=int, default=100, help="QIDs per detail query")
    ap.add_argument("--refresh", action="store_true", help="ignore the SPARQL cache")
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args()
    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    logging.getLogger("httpx").setLevel(logging.WARNING)
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    sp = Sparql(refresh=args.refresh)
    t0 = time.monotonic()
    if args.stage in ("candidates", "all"):
        stage_candidates(sp)
    if args.stage in ("dissolved", "all"):
        stage_dissolved(sp)
    if args.stage in ("select", "all"):
        stage_select(target=args.target)
    if args.stage in ("details", "all"):
        stage_details(sp, batch=args.batch)
    if args.stage in ("assemble", "all"):
        stage_assemble()
    log.info("done in %.0fs (%d live queries, %d cached)", time.monotonic() - t0, sp.queries, sp.cached)
    return 0


if __name__ == "__main__":
    sys.exit(main())