"""Instrument master and identifier crosswalk: FinData entity/instrument ↔ ticker ↔ SEC CIK/SIC.

SEC data are public; requests carry the fair-access User-Agent (no secret) and are throttled to <= 8 req/s.
"""
from __future__ import annotations

import gzip
import json
import os
import time
from pathlib import Path

import httpx
import polars as pl

SEC_TICKERS_URL = "https://www.sec.gov/files/company_tickers.json"
SEC_SUBMISSIONS_URL = "https://data.sec.gov/submissions/CIK{cik:010d}.json"

# SIC divisions (US Department of Labor): used as the coarse "sector" for fixed effects and demeaning.
SIC_DIVISIONS = [
    (100, 999, "A_agriculture"), (1000, 1499, "B_mining"), (1500, 1799, "C_construction"), (2000, 3999, "D_manufacturing"),
    (4000, 4999, "E_transport_utilities"), (5000, 5199, "F_wholesale"), (5200, 5999, "G_retail"),
    (6000, 6799, "H_finance_insurance_realestate"), (7000, 8999, "I_services"), (9100, 9729, "J_public_admin"),
]


def sic_division(sic: int | None) -> str:
    if sic is None:
        return "unknown"
    for lo, hi, name in SIC_DIVISIONS:
        if lo <= sic <= hi:
            return name
    return "unknown"


def _ua() -> str:
    return os.environ.get("SEC_USER_AGENT", "UQO WP39 research (mailto:simon-pierre.boucher@uqo.ca)")


def _get_json(client: httpx.Client, url: str) -> dict:
    r = client.get(url, headers={"User-Agent": _ua(), "Accept-Encoding": "gzip, deflate"})
    r.raise_for_status()
    raw = r.content
    if raw[:2] == b"\x1f\x8b":
        raw = gzip.decompress(raw)
    return json.loads(raw)


def fetch_sec_tickers(cache_dir: Path) -> pl.DataFrame:
    """Current SEC ticker→CIK table (snapshot cached with its download date)."""
    cache_dir.mkdir(parents=True, exist_ok=True)
    f = cache_dir / "sec_company_tickers.json"
    if not f.exists():
        with httpx.Client(timeout=60) as c:
            d = _get_json(c, SEC_TICKERS_URL)
        f.write_text(json.dumps({"downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "data": d}))
    d = json.loads(f.read_text())["data"]
    rows = [{"ticker": v["ticker"].upper(), "cik": int(v["cik_str"]), "sec_name": v["title"]} for v in d.values()]
    return pl.DataFrame(rows)


def fetch_sec_submissions(ciks: list[int], cache_dir: Path, rate_per_s: float = 8.0, log=None, workers: int = 6) -> pl.DataFrame:
    """SIC code, exchanges, fiscal year end and former names per CIK (cached per CIK; global token-bucket rate limit)."""
    import threading
    from concurrent.futures import ThreadPoolExecutor

    out_dir = cache_dir / "sec_submissions"
    out_dir.mkdir(parents=True, exist_ok=True)
    lock, state = threading.Lock(), {"last": 0.0}
    min_gap = 1.0 / rate_per_s

    def fetch_one(cik: int) -> None:
        f = out_dir / f"CIK{cik:010d}.json"
        if f.exists():
            return
        with lock:
            wait = min_gap - (time.time() - state["last"])
            if wait > 0:
                time.sleep(wait)
            state["last"] = time.time()
        with httpx.Client(timeout=60) as c:
            try:
                d = _get_json(c, SEC_SUBMISSIONS_URL.format(cik=cik))
            except httpx.HTTPStatusError as exc:
                d = {"error": exc.response.status_code}
            except Exception as exc:  # noqa: BLE001
                d = {"error": type(exc).__name__}
        slim = {k: d.get(k) for k in ("cik", "sic", "sicDescription", "tickers", "exchanges", "fiscalYearEnd",
                                        "formerNames", "name", "stateOfIncorporation", "error")}
        f.write_text(json.dumps(slim))

    with ThreadPoolExecutor(max_workers=workers) as ex:
        for i, _ in enumerate(ex.map(fetch_one, ciks)):
            if log and i % 500 == 0:
                log.info("SEC submissions %d/%d", i, len(ciks))
    rows = []
    for cik in ciks:
        d = json.loads((out_dir / f"CIK{cik:010d}.json").read_text())
        if d.get("error") and not d.get("sic"):
            (out_dir / f"CIK{cik:010d}.json").unlink(missing_ok=True) if d.get("error") not in (404,) else None
        sic = int(d["sic"]) if d.get("sic") not in (None, "") else None
        rows.append({"cik": cik, "sic": sic, "sic_description": d.get("sicDescription"),
                     "sic_division": sic_division(sic), "sec_tickers": d.get("tickers") or [],
                     "exchanges": d.get("exchanges") or [], "fiscal_year_end": d.get("fiscalYearEnd"),
                     "n_former_names": len(d.get("formerNames") or []), "sec_error": d.get("error")})
    return pl.DataFrame(rows)


def build_crosswalk(entities: pl.DataFrame, sec_tickers: pl.DataFrame, submissions: pl.DataFrame | None) -> pl.DataFrame:
    """entities: entity_id, security_id, ticker (from FinData). Effective-date ranges are the fundamentals span."""
    xw = entities.join(sec_tickers.unique(subset=["ticker"]), on="ticker", how="left")
    if submissions is not None:
        xw = xw.join(submissions, on="cik", how="left")
    xw = xw.with_columns([
        pl.col("cik").is_not_null().alias("has_cik"),
        pl.when(pl.col("cik").is_null()).then(pl.lit("unknown")).otherwise(pl.col("sic_division").fill_null("unknown")).alias("sic_division")
        if submissions is not None else pl.lit("unknown").alias("sic_division"),
    ])
    return xw