"""Async SEC EDGAR client for the fundamentals importer (workstream A).

All access is batch (never at request time). This wraps the three SEC products
the A3 design uses — `company_tickers.json`, `submissions/`, `companyfacts/`, and
the daily filing index — behind one client that honors SEC's fair-access policy:

- an identifying ``User-Agent`` with a contact email on every request (config);
- request spacing well under the 10 req/s limit;
- exponential backoff + retry on 429;
- **403 → alert and stop** (raise ``SecForbiddenError``), never a retry-loop — a
  403 means the UA or request pattern is wrong and retrying won't fix it. The one
  exception is S3's ``AccessDenied`` on an ``/Archives/`` path, which is how the
  bucket reports an absent file (``_is_absent_archive_key``).

Parsing lives here (index fixed-width, submissions pagination); DB writes and the
snapshot mapping live in the importer. No conditional GETs — the companyfacts
endpoint exposes no ETag/Last-Modified (verified), which is why the importer is
daily-index driven rather than polling archives.
"""

from __future__ import annotations

import asyncio
import logging
import os
import re
from datetime import date, datetime
from pathlib import Path
from typing import Any

import httpx

from app.config import settings
from app.exceptions import ProviderError
from app.services.earnings_alignment import normalise_symbol

logger = logging.getLogger(__name__)

_WWW = "https://www.sec.gov"
_DATA = "https://data.sec.gov"

# Resolve CA bundle for explicit httpx verify (matches app/providers/fmp.py).
_CA = os.environ.get("SSL_CERT_FILE", "")
_CA_VERIFY: str | bool = _CA if _CA and Path(_CA).exists() else True

_FORMS_10 = frozenset({"10-K", "10-Q", "10-K/A", "10-Q/A"})


class SecError(ProviderError):
    """SEC request failed (403, exhausted 429/5xx, timeout, transport, parse)."""


class SecForbiddenError(SecError):
    """SEC returned 403 — User-Agent/pattern rejected. Alert and stop."""


class SecNotFoundError(SecError):
    """The resource does not exist (e.g. no daily index published for a day).

    The *only* error a caller may treat as 'missing' — every other SecError
    (fair-access rejection, exhausted retries, 5xx, timeout) must propagate so a
    fetch failure is never mistaken for an empty result.

    Raised for a 404, and for the one 403 that also means "absent": see
    ``_is_absent_archive_key``."""


def _is_absent_archive_key(url: str, resp: httpx.Response) -> bool:
    """True when a 403 means "this file does not exist", not "you are blocked".

    ``www.sec.gov/Archives`` is served straight out of an S3 bucket that grants
    no ``s3:ListBucket``, so a missing key cannot be answered with 404 — S3
    returns **403 with its ``AccessDenied`` XML** instead. SEC publishes a daily
    index only for business days, so every weekend and market holiday inside an
    incremental walk lands on exactly this response (verified 2026-07-30:
    ``form.20260725.idx``, a Saturday, 403s while the Friday and Monday files
    return 200 on the same User-Agent).

    A genuine fair-access rejection is distinguishable and must stay fatal: it is
    SEC's WAF interstitial — ``text/html``, "Your Request Originates from an
    Undeclared Automated Tool" — and it is returned for files that *do* exist,
    on any path. Hence the narrow gate: the Archives prefix plus S3's own error
    document. Nothing else may be downgraded to "missing"."""
    try:
        parsed = httpx.URL(url)
    except (TypeError, ValueError):  # pragma: no cover — url comes from us
        return False
    if parsed.host != "www.sec.gov" or not parsed.path.startswith("/Archives/"):
        return False
    if "xml" not in resp.headers.get("Content-Type", "").lower():
        return False
    try:
        return "<Code>AccessDenied</Code>" in resp.text
    except (UnicodeDecodeError, httpx.HTTPError):  # pragma: no cover
        return False


def _looks_like_contact_email(ua: str) -> bool:
    if "example.com" in ua.lower() or "set-a-real-email" in ua.lower():
        return False
    return re.search(r"[^@\s]+@[^@\s]+\.[^@\s]+", ua) is not None


# SEC asks callers to stay well under 10 req/s; enforce a floor on real clients.
_MIN_PROD_SPACING = 0.11


def cik10(cik: int | str) -> str:
    """Zero-pad a CIK to the 10-digit form SEC URLs use (320193 -> 0000320193)."""
    return str(int(cik)).zfill(10)


class SecClient:
    """Fair-access SEC HTTP client. Use as ``async with SecClient() as c:``."""

    def __init__(
        self,
        *,
        user_agent: str | None = None,
        spacing_seconds: float | None = None,
        max_retries: int | None = None,
        timeout: float | None = None,
        transport: httpx.AsyncBaseTransport | None = None,
    ) -> None:
        self._ua = user_agent or settings.sec_user_agent
        self._spacing = (
            spacing_seconds if spacing_seconds is not None else settings.sec_request_spacing_seconds
        )
        self._max_retries = (
            max_retries if max_retries is not None else settings.sec_max_retries
        )
        self._timeout = timeout if timeout is not None else settings.sec_request_timeout_seconds
        self._transport = transport  # injectable for tests
        self._client: httpx.AsyncClient | None = None
        self._lock = asyncio.Lock()
        self._last_request = 0.0

    def _validate_fair_access(self) -> None:
        """On a real (non-mocked) client, enforce SEC fair-access preconditions
        so we can't accidentally hammer SEC or get 403'd: a genuine contact-email
        User-Agent and a spacing floor. Mock transports skip this (tests use 0)."""
        if not _looks_like_contact_email(self._ua):
            raise SecError(
                "sec_user_agent must contain a real contact email (got "
                f"{self._ua!r}) — SEC fair-access requires it"
            )
        if self._spacing < _MIN_PROD_SPACING:
            raise SecError(
                f"sec_request_spacing_seconds {self._spacing} is below the "
                f"{_MIN_PROD_SPACING}s fair-access floor"
            )

    async def __aenter__(self) -> "SecClient":
        if self._transport is None:
            self._validate_fair_access()
        self._client = httpx.AsyncClient(
            headers={"User-Agent": self._ua, "Accept-Encoding": "gzip, deflate"},
            timeout=self._timeout,
            verify=_CA_VERIFY,
            transport=self._transport,
        )
        return self

    async def __aexit__(self, *exc) -> None:
        if self._client is not None:
            await self._client.aclose()
            self._client = None

    async def _throttle(self) -> None:
        async with self._lock:
            now = asyncio.get_event_loop().time()
            wait = self._spacing - (now - self._last_request)
            if wait > 0:
                await asyncio.sleep(wait)
            self._last_request = asyncio.get_event_loop().time()

    async def _get(self, url: str) -> httpx.Response:
        assert self._client is not None, "use `async with SecClient()`"
        attempt = 0
        while True:
            await self._throttle()
            try:
                resp = await self._client.get(url)
            except (httpx.TimeoutException, httpx.TransportError) as exc:
                attempt += 1
                if attempt > self._max_retries:
                    raise SecError(f"SEC network error for {url}: {exc}") from exc
                await asyncio.sleep(min(2.0**attempt, 30.0))
                continue

            code = resp.status_code
            if code == 403:
                if _is_absent_archive_key(url, resp):
                    raise SecNotFoundError(f"SEC 403/AccessDenied (absent) for {url}")
                raise SecForbiddenError(
                    f"SEC 403 for {url} — User-Agent/pattern rejected; set a real "
                    "sec_user_agent contact email"
                )
            if code == 404:
                raise SecNotFoundError(f"SEC 404 for {url}")
            # 429 and 5xx are transient — retry with backoff, honoring Retry-After.
            if code == 429 or 500 <= code < 600:
                attempt += 1
                if attempt > self._max_retries:
                    raise SecError(f"SEC {code} after {self._max_retries} retries: {url}")
                delay = _retry_after_seconds(resp) or min(2.0**attempt, 30.0)
                logger.warning("SEC %d for %s — backoff %.1fs (attempt %d)", code, url, delay, attempt)
                await asyncio.sleep(delay)
                continue
            if code >= 400:
                raise SecError(f"SEC {code} for {url}")
            return resp

    async def get_json(self, url: str) -> Any:
        return (await self._get(url)).json()

    async def get_text(self, url: str) -> str:
        return (await self._get(url)).text

    # -- domain fetchers ---------------------------------------------------

    async def company_tickers(self) -> dict[str, int]:
        """Map normalised ticker -> CIK (int). Multi-class tickers share a CIK."""
        data = await self.get_json(f"{_WWW}/files/company_tickers.json")
        out: dict[str, int] = {}
        for row in data.values():
            sym = normalise_symbol(row.get("ticker"))
            if sym:
                out[sym] = int(row["cik_str"])
        return out

    async def submissions(self, cik: int | str, *, include_history: bool = False) -> dict[str, Any]:
        """Issuer metadata + filing list.

        ``filings.recent`` caps at 1000; older accessions live in
        ``filings.files[]`` shards. Only ``include_history=True`` (the one-time
        full backfill) fetches those shards — SIC refresh and incremental runs
        use the recent list alone and make no extra requests.
        """
        base = await self.get_json(f"{_DATA}/submissions/CIK{cik10(cik)}.json")
        filings = _rows_from_arrays(base["filings"]["recent"])
        if include_history:
            for shard in base["filings"].get("files") or []:
                shard_data = await self.get_json(f"{_DATA}/submissions/{shard['name']}")
                filings.extend(_rows_from_arrays(shard_data))
        return {
            "cik": int(base["cik"]),
            "name": base.get("name"),
            "sic": base.get("sic"),
            "sic_description": base.get("sicDescription"),
            "fiscal_year_end": base.get("fiscalYearEnd"),
            "tickers": base.get("tickers") or [],
            "filings": filings,
        }

    async def companyfacts(self, cik: int | str) -> dict[str, Any]:
        """Raw companyfacts JSON ({cik, entityName, facts})."""
        return await self.get_json(f"{_DATA}/api/xbrl/companyfacts/CIK{cik10(cik)}.json")

    async def latest_index_date(self, today: date | None = None) -> date | None:
        """The most recent published daily-index date (drives the revision). Checks
        the current quarter, falling back to the previous one at a quarter boundary."""
        today = today or date.today()
        for year, qtr in _quarters_back(today, 2):
            url = f"{_WWW}/Archives/edgar/daily-index/{year}/QTR{qtr}/index.json"
            try:
                idx = await self.get_json(url)
            except SecNotFoundError:
                continue  # quarter dir absent — only 404 is "missing"
            dates = [
                d
                for item in idx.get("directory", {}).get("item", [])
                if (d := _index_file_date(item.get("name", ""))) is not None
                and d <= today
            ]
            if dates:
                return max(dates)
        return None

    async def daily_index(self, day: date) -> list[dict[str, Any]]:
        """Parse the daily form index into 10-K/10-Q(/A) rows for all issuers.

        Returns [{form, cik, accession, company}]. The caller filters to the
        tracked universe. A missing index (weekend/holiday/not-yet-published)
        returns [] rather than raising.
        """
        qtr = (day.month - 1) // 3 + 1
        url = f"{_WWW}/Archives/edgar/daily-index/{day.year}/QTR{qtr}/form.{day:%Y%m%d}.idx"
        try:
            text = await self.get_text(url)
        except SecNotFoundError:
            # Absent is normal on a weekend (SEC publishes business days only). On a
            # weekday it is not: a SEC hiccup — or a rejection page misread as absent
            # — would otherwise let the importer advance past real filings silently,
            # so surface it at WARNING instead of hiding it in the info stream.
            logger.log(
                logging.INFO if day.weekday() >= 5 else logging.WARNING,
                "no daily index published for %s (%s)",
                day,
                f"{day:%a}",
            )
            return []  # weekend/holiday/not-yet-published; other errors propagate
        return _parse_form_index(text)


def _rows_from_arrays(arrays: dict[str, list]) -> list[dict[str, Any]]:
    """Turn SEC's parallel-array filing block into row dicts (keeping only 10-K/10-Q
    family filings — the ones that carry XBRL fundamentals)."""
    forms = arrays.get("form", [])
    out: list[dict[str, Any]] = []
    for i, form in enumerate(forms):
        if form not in _FORMS_10:
            continue
        out.append(
            {
                "accession": arrays["accessionNumber"][i],
                "form": form,
                "report_date": arrays["reportDate"][i] or None,
                "filing_date": arrays["filingDate"][i] or None,
                "acceptance_datetime": arrays["acceptanceDateTime"][i] or None,
                "is_xbrl": bool(arrays.get("isXBRL", [0] * len(forms))[i]),
            }
        )
    return out


def _parse_form_index(text: str) -> list[dict[str, Any]]:
    """Parse a daily ``form.YYYYMMDD.idx`` (fixed columns: Form / Company / CIK /
    Date Filed / File Name-with-accession)."""
    rows: list[dict[str, Any]] = []
    started = False
    for line in text.splitlines():
        if not started:
            if set(line.strip()) == {"-"}:  # the dashed separator row
                started = True
            continue
        parts = line.split()
        if len(parts) < 5:
            continue
        form = parts[0]
        if form not in _FORMS_10:
            continue
        path = parts[-1]  # edgar/data/<cik>/<accession>.txt
        cik = _cik_from_path(path)
        accession = _accession_from_path(path)
        if cik is None or accession is None:
            continue
        rows.append({"form": form, "cik": cik, "accession": accession, "path": path})
    return rows


def _retry_after_seconds(resp: httpx.Response) -> float | None:
    """Parse a numeric-seconds Retry-After header (SEC uses seconds), capped."""
    raw = resp.headers.get("Retry-After")
    if not raw:
        return None
    try:
        return min(float(raw), 60.0)
    except (TypeError, ValueError):
        return None


def _index_file_date(name: str) -> date | None:
    if name.startswith("form.") and name.endswith(".idx"):
        try:
            return datetime.strptime(name[5:13], "%Y%m%d").date()
        except ValueError:
            return None
    return None


def _quarters_back(today: date, n: int) -> list[tuple[int, int]]:
    """(year, quarter) for `today`'s quarter and the previous n-1, newest first."""
    q = (today.month - 1) // 3 + 1
    out = []
    y = today.year
    for _ in range(n):
        out.append((y, q))
        q -= 1
        if q == 0:
            q = 4
            y -= 1
    return out


def _cik_from_path(path: str) -> int | None:
    segs = path.split("/")
    if len(segs) >= 3 and segs[2].isdigit():
        return int(segs[2])
    return None


def _accession_from_path(path: str) -> str | None:
    stem = path.rsplit("/", 1)[-1]
    if stem.endswith(".txt"):
        stem = stem[:-4]
    return stem or None