#!/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 + per-country and per-industry boost queries → data/seed/candidates.json
  dissolved   P576 check over every candidate → data/seed/dissolved.json (excluded by `select`)
  select      normalise websites, drop generic hosts / duplicate domains, diversify (US ≤ 40 %, 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.
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": "pharmaceutical company", "Q210167": "video game developer",
    "Q1137109": "video game publisher", "Q1762059": "film production company", "Q18127": "record label", "Q2085381": "publisher",
    "Q613142": "law firm", "Q740752": "transport company", "Q2401749": "telecommunications company", "Q131734": "brewery",
    "Q507619": "retail chain", "Q936518": "aerospace manufacturer", "Q206361": "conglomerate", "Q249556": "railway company",
    "Q2005696": "real estate company", "Q708676": "shipyard", "Q1668024": "internet service", "Q1320047": "book publisher",
    "Q19967801": "online service",
}
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)]
MIN_SITELINKS = 5
BOOST_MIN_SITELINKS = 2
BOOST_LIMIT = 2500
INDUSTRY_BOOST_LIMIT = 600

# Diversification (spec: "seed diversified companies across US, Canada, Europe, UK, Japan, South Korea, India, Australia, LatAm, ME, Africa, SEA")
US_CAP_SHARE = 0.40
OTHER_CAP_SHARE = 0.12
UNKNOWN_COUNTRY_SHARE = 0.03
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, "Q708676": 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) -> 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}
}}"""


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) come back."""
    return f"""SELECT ?item ?dissolved WHERE {{ {values_clause("item", qids)} ?item wdt:P576 ?dissolved . }}"""


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"))


# ------------------------------------------------------------------------------------------------------------ stage: candidates
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()

    def add(row: dict[str, str], *, cls: str | None = None, boost_country: str | None = None, boost_industry: str | None = None) -> None:
        if row.get("dissolved"):
            dissolved.add(qid_of(row["item"]))
            return
        qid = qid_of(row["item"])
        c = cands.setdefault(qid, {"sitelinks": 0, "websites": [], "isos": [], "classes": [], "boost_countries": [], "boost_industries": []})
        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)

    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:
                raise
            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, cls=cls)

    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)
        log.info("candidates so far: %d", len(cands))
    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"])
    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)
            continue
        for r in rows:
            add(r, 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, boost_industry=slug)
    for qid in dissolved:
        cands.pop(qid, None)
    dump_json(out_path, cands)
    log.info("candidates: %d (%d dissolved dropped) → %s", len(cands), len(dissolved), 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)
    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"]})
    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 select_diversified(rows: list[dict[str, Any]], *, target: int, countries: dict[str, dict[str, str]]) -> list[dict[str, Any]]:
    """Sitelinks-ordered selection with country caps and minimum coverage. 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()

    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)

    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 by_country.get(iso, [])[:minimum]:
            take(r)
    for r in rows:
        if len(chosen) >= target:
            break
        if r["wikidata_id"] in chosen or r["sitelinks"] < MIN_SITELINKS:
            continue
        if per_country[r["country"]] >= caps[r["country"]] or class_capped(r):
            continue
        take(r)
    return list(chosen.values())


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 = select_diversified(kept, target=target, countries=countries)
    # 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})
    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)
    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 assign_tiers(rows: list[dict[str, Any]]) -> None:
    rows.sort(key=lambda r: (-r["importance"], -r["sitelinks"], r["wikidata_id"]))
    for i, r in enumerate(rows):
        r["tier"] = 1 if i < 150 else 2 if i < 950 else 3 if i < 3450 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])
    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)
    public = "Q891723" in sel.get("classes", []) 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 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]
    rows = [build_row(s, details[s["wikidata_id"]], harvested_at) for s in selected]
    # 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_row(s, details[s["wikidata_id"]], harvested_at) for s in 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()}
    dropped: list[dict[str, Any]] = []
    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)
    write_readme(final, countries)
    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]]) -> 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
    no_ind = sum(1 for r in rows if not r["industries"])
    public = sum(1 for r in rows if r["public_company"])
    region = Counter((countries.get(r["country"] or "", {}).get("region") or "other") for r in rows)
    generated = rows[0]["harvested_at"] if rows else ""
    intro = (f"Generated by `scripts/seed_wikidata.py` on {generated}. **{len(rows)} companies**, {public} public, {no_ind} without an "
             f"industry mapping, {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.")
    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')} | {n} | {100 * n / len(rows):.1f} % |" for c, n in by_country.most_common()], "",
        "## 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()], "",
    ]
    (OUT_DIR / "README.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


# ------------------------------------------------------------------------------------------------------------ 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=8000, help="companies to select before detail fetching (default 8000)")
    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())