#!/usr/bin/env python3
"""Amazon forecast-vintage extractor.

Reads source_manifest.csv, fetches primary SEC/IR HTML, and parses capex actuals
from cash-flow statements plus FY2025/FY2026 capex-guidance vintages from MD&A,
SEC 8-K ex99, and earnings materials. Amazon does not disclose MW/GW;
normalized_gw is blank.
"""

from __future__ import annotations

import argparse
import csv
import re
import urllib.request
from datetime import datetime
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
DEFAULT_MANIFEST = BASE_DIR / "source_manifest.csv"
DEFAULT_OUTPUT = BASE_DIR / "forecast_vintage.csv"

CANONICAL_FIELDS = [
    "company",
    "ticker",
    "as_of_period",
    "target_period",
    "metric_name",
    "raw_value",
    "normalized_gw",
    "unit",
    "forecast_or_actual",
    "is_derived",
    "source_url",
    "source_location",
    "source_snippet",
    "snippet_support_status",
    "confidence",
    "caveat",
]

GW_PROXY_CAVEAT = (
    "Amazon/AWS does not disclose data-center MW/GW. Row tracks USD capex proxy only; "
    "normalized_gw intentionally blank."
)

USER_AGENT = "AMZN-forecast-vintage-extractor/1.0 (research; contact@example.com)"

CAPEX_TAG = "PaymentsToAcquireProductiveAssets"

PERIOD_ENDS = {
    "FY2024-FQ1": "2024-03-31",
    "FY2024-FQ2": "2024-06-30",
    "FY2024-FQ3": "2024-09-30",
    "FY2024-FQ4": "2024-12-31",
    "FY2025-FQ1": "2025-03-31",
    "FY2025-FQ2": "2025-06-30",
    "FY2025-FQ3": "2025-09-30",
    "FY2025-FQ4": "2025-12-31",
    "FY2026-FQ1": "2026-03-31",
}

SEC_URL_BY_PERIOD = {
    "FY2024-FQ4": (
        "https://www.sec.gov/Archives/edgar/data/1018724/"
        "000101872425000004/amzn-20241231.htm"
    ),
    "FY2025-FQ4": (
        "https://www.sec.gov/Archives/edgar/data/1018724/"
        "000101872426000004/amzn-20251231.htm"
    ),
    "FY2025-FQ3": (
        "https://www.sec.gov/Archives/edgar/data/1018724/"
        "000101872425000123/amzn-20250930.htm"
    ),
}

GUIDANCE_SNIPPET_PATTERNS: dict[str, list[str]] = {
    "amzn_fy2025_q4_8k_ex99": [
        (
            r"we expect to invest about \$200 billion in capital expenditures "
            r"across Amazon in 2026"
        ),
    ],
    "amzn_fy2025_q4_ir_release": [
        (
            r"we expect to invest about \$200 billion in capital expenditures "
            r"across Amazon in 2026"
        ),
    ],
    "amzn_fy2024_10k": [
        (
            r"We expect cash capital expenditures to increase in 2025, primarily "
            r"driven by investments in technology infrastructure\."
        ),
    ],
    "amzn_fy2025_q1_10q": [
        (
            r"We expect cash capital expenditures to increase in 2025, primarily "
            r"driven by investments in technology infrastructure\."
        ),
    ],
    "amzn_fy2025_q2_10q": [
        (
            r"Cash capital expenditures were \$16\.4 billion and \$31\.4 billion "
            r"during Q2 2024 and Q2 2025"
        ),
    ],
    "amzn_fy2025_q3_10q": [
        (
            r"Cash capital expenditures were \$21\.3 billion and \$34\.2 billion "
            r"during Q3 2024 and Q3 2025"
        ),
    ],
    "amzn_fy2026_q1_10q": [
        (
            r"both of which we expect to increase in 2026"
        ),
    ],
}

SOURCE_PRIORITY = {
    "sec_10k": 0,
    "sec_10q": 1,
    "sec_8k_exhibit_99_1": 2,
    "official_ir_release": 3,
    "earnings_call_webcast": 4,
}


def load_manifest(manifest_path: Path) -> list[dict[str, str]]:
    with manifest_path.open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle))


def fetch_html(url: str) -> str:
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(request, timeout=90) as response:
        return response.read().decode("utf-8", errors="replace")


def html_to_text(html: str) -> str:
    text = re.sub(r"<script[^>]*>.*?</script>", " ", html, flags=re.I | re.S)
    text = re.sub(r"<style[^>]*>.*?</style>", " ", html, flags=re.I | re.S)
    text = re.sub(r"<[^>]+>", " ", text)
    text = re.sub(r"&#\d+;", " ", text)
    return re.sub(r"\s+", " ", text)


def find_snippet_in_text(text: str, patterns: list[str]) -> str | None:
    for pattern in patterns:
        match = re.search(pattern, text, flags=re.I)
        if match:
            return match.group(0)
    return None


def parse_contexts(html: str) -> dict[str, tuple[str, str]]:
    contexts: dict[str, tuple[str, str]] = {}
    for match in re.finditer(
        r'<xbrli:context id="([^"]+)"[^>]*>(.*?)</xbrli:context>', html, re.S
    ):
        body = match.group(2)
        start = re.search(r"<xbrli:startDate>([^<]+)", body)
        end = re.search(r"<xbrli:endDate>([^<]+)", body)
        if start and end:
            contexts[match.group(1)] = (start.group(1), end.group(1))
    return contexts


def parse_capex_by_context(html: str) -> dict[str, int]:
    values: dict[str, int] = {}
    pattern = re.compile(
        r'contextRef="([^"]*)"[^>]*name="us-gaap:' + CAPEX_TAG + r'"[^>]*>([^<]+)<'
        r'|name="us-gaap:' + CAPEX_TAG + r'"[^>]*contextRef="([^"]*)"[^>]*>([^<]+)<',
        re.I,
    )
    for match in pattern.finditer(html):
        ctx = match.group(1) or match.group(3)
        raw = (match.group(2) or match.group(4)).replace(",", "")
        values[ctx] = int(raw)
    return values


def days_between(start: str, end: str) -> int:
    s = datetime.strptime(start, "%Y-%m-%d")
    e = datetime.strptime(end, "%Y-%m-%d")
    return (e - s).days


def quarterly_capex_millions(html: str, period: str) -> tuple[int, str] | None:
    end_date = PERIOD_ENDS.get(period)
    if not end_date:
        return None
    contexts = parse_contexts(html)
    capex = parse_capex_by_context(html)
    year = end_date[:4]
    for ctx, (start, end) in contexts.items():
        if end != end_date or ctx not in capex:
            continue
        span = days_between(start, end)
        if 80 <= span <= 100 and start.startswith(year):
            val = capex[ctx]
            raw = f"{val:,}"
            return val, f"Purchases of property and equipment ({raw})"
    return None


def annual_capex_millions(html: str, fiscal_year: str) -> tuple[int, str] | None:
    end_date = f"{fiscal_year}-12-31"
    contexts = parse_contexts(html)
    capex = parse_capex_by_context(html)
    for ctx, (start, end) in contexts.items():
        if end != end_date or ctx not in capex:
            continue
        if start == f"{fiscal_year}-01-01":
            val = capex[ctx]
            raw = f"{val:,}"
            return val, f"Purchases of property and equipment ({raw})"
    return None


def ytd_capex_millions(html: str, year: str, months: int) -> tuple[int, str] | None:
    end_month = {6: "06-30", 9: "09-30"}[months]
    end_date = f"{year}-{end_month}"
    contexts = parse_contexts(html)
    capex = parse_capex_by_context(html)
    for ctx, (start, end) in contexts.items():
        if end == end_date and start == f"{year}-01-01" and ctx in capex:
            val = capex[ctx]
            raw = f"{val:,}"
            label = "H1" if months == 6 else "9M"
            return val, f"Purchases of property and equipment YTD {label} ({raw})"
    return None


def billions_from_millions(millions: int) -> str:
    return f"{millions / 1000:.3f}".rstrip("0").rstrip(".")


def guidance_metadata(source_id: str, carry_forward: str | None = None) -> dict[str, str]:
    if source_id in ("amzn_fy2024_10k", "amzn_fy2025_q1_10q"):
        return {
            "target_period": "FY2025",
            "raw_value": "105",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                "Amazon gives qualitative FY2025 capex growth guidance in 10-K/10-Q MD&A. "
                "Implied ~$105B derived from Q4 FY2024 earnings-call commentary that "
                "$26.3B quarterly cash capex is representative of 2025 annualized rate "
                "(~$105B). Not MW/GW capacity."
            ),
        }
    if source_id == "amzn_fy2025_q2_10q":
        return {
            "target_period": "FY2025",
            "raw_value": "118",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                "Implied ~$118B FY2025 capex from CFO commentary that $31.4B Q2 cash "
                "capex is representative of H2 2025 rate (H1 $55.6B + H2 ~$62.8B). "
                "Snippet is SEC MD&A cash capex actuals. Not MW/GW capacity."
            ),
        }
    if source_id == "amzn_fy2025_q3_10q":
        return {
            "target_period": "FY2025",
            "raw_value": "125",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                "Implied ~$125B FY2025 capex from Q3 FY2025 earnings-call commentary "
                "(CFO Olsavsky). SEC MD&A reports $89.9B cash capex for 9M 2025. "
                "Not MW/GW capacity."
            ),
        }
    if source_id in ("amzn_fy2025_q4_8k_ex99", "amzn_fy2025_q4_ir_release"):
        return {
            "target_period": "FY2026",
            "raw_value": "200",
            "is_derived": "0",
            "confidence": "high",
            "caveat": (
                "Management FY2026 capex guidance (~$200B) from SEC-filed earnings "
                "release / IR. Not MW/GW capacity."
            ),
        }
    if source_id == "amzn_fy2026_q1_10q":
        return {
            "target_period": "FY2026",
            "raw_value": carry_forward or "200",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                "No new full-year FY2026 numeric capex guide at FQ1; vintage carries "
                f"forward Q4 FY2025 ~${carry_forward or '200'}B. MD&A expects cash "
                "capex to increase in 2026. Not MW/GW capacity."
            ),
        }
    raise ValueError(f"Unknown guidance source_id: {source_id}")


def row_key(row: dict[str, str]) -> tuple[str, str, str]:
    return (row["as_of_period"], row["target_period"], row["metric_name"])


def base_row(
    manifest_row: dict[str, str],
    *,
    target_period: str,
    metric_name: str,
    raw_value: str,
    forecast_or_actual: str,
    is_derived: str,
    source_location: str,
    source_snippet: str,
    snippet_support_status: str,
    confidence: str,
    caveat: str,
    source_url: str | None = None,
) -> dict[str, str]:
    return {
        "company": manifest_row["company"],
        "ticker": manifest_row["ticker"],
        "as_of_period": manifest_row["period"],
        "target_period": target_period,
        "metric_name": metric_name,
        "raw_value": raw_value,
        "normalized_gw": "",
        "unit": "USD_billions",
        "forecast_or_actual": forecast_or_actual,
        "is_derived": is_derived,
        "source_url": source_url or manifest_row["source_url"],
        "source_location": source_location,
        "source_snippet": source_snippet,
        "snippet_support_status": snippet_support_status,
        "confidence": confidence,
        "caveat": caveat,
    }


def extract_guidance_row(
    manifest_row: dict[str, str],
    html: str,
    carry_forward: str | None = None,
) -> dict[str, str] | None:
    source_id = manifest_row["source_id"]
    patterns = GUIDANCE_SNIPPET_PATTERNS.get(source_id)
    if not patterns:
        return None
    text = html_to_text(html)
    snippet = find_snippet_in_text(text, patterns)
    if not snippet:
        return None
    meta = guidance_metadata(source_id, carry_forward)
    period_label = manifest_row["period"].replace("FY", "FY ").replace("-FQ", " Q")
    return base_row(
        manifest_row,
        target_period=meta["target_period"],
        metric_name="capex_guidance",
        raw_value=meta["raw_value"],
        forecast_or_actual="forecast",
        is_derived=meta["is_derived"],
        source_location=f"{period_label} MD&A / earnings materials",
        source_snippet=snippet,
        snippet_support_status="verified_primary",
        confidence=meta["confidence"],
        caveat=meta["caveat"],
    )


def extract_from_manifest_row(
    manifest_row: dict[str, str],
    html_cache: dict[str, str],
    ytd_state: dict[str, int],
    carry_forward_guidance: str | None = None,
) -> list[dict[str, str]]:
    metrics = {m.strip() for m in manifest_row["extractable_metrics"].split(";")}
    doc_type = manifest_row["document_type"]
    rows: list[dict[str, str]] = []

    url = manifest_row["source_url"]
    html = html_cache.get(url)
    if html is None:
        html = fetch_html(url)
        html_cache[url] = html

    if any(m.startswith("capex_guidance") for m in metrics):
        guide = extract_guidance_row(manifest_row, html, carry_forward_guidance)
        if guide:
            rows.append(guide)
        elif doc_type in ("sec_10k", "sec_10q") and any(
            m.startswith("capex_guidance") for m in metrics
        ):
            raise ValueError(
                f"Could not extract guidance snippet from {url} ({manifest_row['source_id']})"
            )

    period = manifest_row["period"]

    if "capex_actual_fy" in metrics and doc_type == "sec_10k":
        year = "2024" if period == "FY2024-FQ4" else "2025"
        parsed = annual_capex_millions(html, year)
        if parsed:
            millions, snippet = parsed
            rows.append(
                base_row(
                    manifest_row,
                    target_period=f"FY{year}",
                    metric_name="capex_actual",
                    raw_value=billions_from_millions(millions),
                    forecast_or_actual="actual",
                    is_derived="0",
                    source_location=(
                        "Consolidated Statements of Cash Flows; "
                        f"FY{year} purchases of property and equipment"
                    ),
                    source_snippet=snippet,
                    snippet_support_status="verified_primary",
                    confidence="high",
                    caveat=GW_PROXY_CAVEAT,
                )
            )

    if "capex_actual_ytd" in metrics and doc_type == "sec_10q":
        parsed = ytd_capex_millions(html, "2025", 9)
        if parsed:
            millions, snippet = parsed
            ytd_state["2025"] = millions
            rows.append(
                base_row(
                    manifest_row,
                    target_period="FY2025-9M",
                    metric_name="capex_actual_ytd",
                    raw_value=billions_from_millions(millions),
                    forecast_or_actual="actual",
                    is_derived="0",
                    source_location=manifest_row["expected_section"],
                    source_snippet=snippet,
                    snippet_support_status="verified_primary",
                    confidence="high",
                    caveat=GW_PROXY_CAVEAT,
                )
            )

    if "capex_actual_quarter" in metrics and doc_type in ("sec_10q", "sec_10k"):
        if period != "FY2024-FQ4":
            parsed = quarterly_capex_millions(html, period)
            if parsed:
                millions, snippet = parsed
                rows.append(
                    base_row(
                        manifest_row,
                        target_period=period,
                        metric_name="capex_actual",
                        raw_value=billions_from_millions(millions),
                        forecast_or_actual="actual",
                        is_derived="0",
                        source_location=manifest_row["expected_section"],
                        source_snippet=snippet,
                        snippet_support_status="verified_primary",
                        confidence="high",
                        caveat=GW_PROXY_CAVEAT,
                    )
                )

    return rows


def append_derived_rows(rows: list[dict[str, str]]) -> list[dict[str, str]]:
    by_key = {row_key(r): r for r in rows}

    def get_value(period: str, target: str, metric: str) -> float | None:
        row = by_key.get((period, target, metric))
        return float(row["raw_value"]) if row else None

    def add_derived(
        template_period: str,
        target_period: str,
        value: float,
        source_url: str,
        source_location: str,
        source_snippet: str,
    ) -> None:
        key = (template_period, target_period, "capex_actual")
        if key in by_key:
            return
        template = next((r for r in rows if r["as_of_period"] == template_period), rows[0])
        raw = f"{value:.3f}".rstrip("0").rstrip(".")
        row = base_row(
            {
                "company": template["company"],
                "ticker": template["ticker"],
                "period": template_period,
                "source_url": source_url,
            },
            target_period=target_period,
            metric_name="capex_actual",
            raw_value=raw,
            forecast_or_actual="actual",
            is_derived="1",
            source_location=source_location,
            source_snippet=source_snippet,
            snippet_support_status="verified_primary",
            confidence="high",
            caveat=GW_PROXY_CAVEAT,
            source_url=source_url,
        )
        rows.append(row)
        by_key[key] = row

    annual_fy25 = get_value("FY2025-FQ4", "FY2025", "capex_actual")
    nine_m = get_value("FY2025-FQ3", "FY2025-9M", "capex_actual_ytd")
    if annual_fy25 is not None and nine_m is not None:
        q4_m = int(round(annual_fy25 * 1000 - nine_m * 1000))
        add_derived(
            "FY2025-FQ4",
            "FY2025-FQ4",
            q4_m / 1000,
            SEC_URL_BY_PERIOD["FY2025-FQ4"],
            "Derived from FY2025 annual minus 9M YTD (SEC 10-K vs Q3 10-Q cash-flow YTD)",
            f"Purchases of property and equipment ({q4_m:,})",
        )

    return sorted(rows, key=lambda r: (r["as_of_period"], r["target_period"], r["metric_name"]))


def dedupe_rows(
    manifest_rows: list[dict[str, str]],
    extracted: list[dict[str, str]],
) -> list[dict[str, str]]:
    priority_by_url = {
        row["source_url"]: SOURCE_PRIORITY.get(row["document_type"], 99)
        for row in manifest_rows
    }
    best: dict[tuple[str, str, str], dict[str, str]] = {}
    for row in extracted:
        key = row_key(row)
        existing = best.get(key)
        if existing is None:
            best[key] = row
            continue
        if priority_by_url[row["source_url"]] < priority_by_url[existing["source_url"]]:
            best[key] = row
    return sorted(
        best.values(),
        key=lambda r: (r["as_of_period"], r["target_period"], r["metric_name"]),
    )


def validate_rows(
    rows: list[dict[str, str]],
    manifest_rows: list[dict[str, str]],
) -> None:
    manifest_urls = {row["source_url"] for row in manifest_rows}
    required = set(CANONICAL_FIELDS)
    as_of_periods: set[str] = set()
    for idx, row in enumerate(rows, start=1):
        missing = required - row.keys()
        if missing:
            raise ValueError(f"Row {idx} missing columns: {sorted(missing)}")
        if not row["source_url"].startswith("https://"):
            raise ValueError(f"Row {idx} source_url must be https")
        if row["source_url"] not in manifest_urls:
            raise ValueError(f"Row {idx} source_url not in manifest: {row['source_url']}")
        if row["ticker"] != "AMZN":
            raise ValueError(f"Row {idx} ticker must be AMZN")
        as_of_periods.add(row["as_of_period"])
    if len(as_of_periods) < 8:
        raise ValueError(f"Need >=8 as_of_period vintages, got {len(as_of_periods)}")
    guidance = [r for r in rows if r["metric_name"] == "capex_guidance"]
    if len(guidance) < 6:
        raise ValueError(f"Need >=6 capex_guidance vintages, got {len(guidance)}")
    fy26 = [r for r in guidance if r["target_period"] == "FY2026"]
    if len(fy26) < 2:
        raise ValueError(f"Need >=2 FY2026 capex_guidance vintages, got {len(fy26)}")
    if len(rows) < 16:
        raise ValueError(f"Need >=16 vintage rows, got {len(rows)}")


def emit_rows(
    manifest_path: Path = DEFAULT_MANIFEST,
    output_path: Path = DEFAULT_OUTPUT,
) -> list[dict[str, str]]:
    manifest_rows = load_manifest(manifest_path)
    html_cache: dict[str, str] = {}
    ytd_state: dict[str, int] = {}
    extracted: list[dict[str, str]] = []

    for manifest_row in manifest_rows:
        extracted.extend(
            extract_from_manifest_row(manifest_row, html_cache, ytd_state)
        )

    interim = dedupe_rows(manifest_rows, extracted)
    fy26_guide = next(
        (
            r
            for r in interim
            if r["metric_name"] == "capex_guidance" and r["target_period"] == "FY2026"
        ),
        None,
    )
    carry = fy26_guide["raw_value"] if fy26_guide else "200"

    # FY2026-FQ1 guidance carry-forward (no new numeric guide; 10-Q MD&A processed above).
    q1_fy26_row = next(
        (r for r in manifest_rows if r["source_id"] == "amzn_fy2026_q1_10q"),
        None,
    )
    if q1_fy26_row and not any(
        r["as_of_period"] == "FY2026-FQ1"
        and r["target_period"] == "FY2026"
        and r["metric_name"] == "capex_guidance"
        for r in interim
    ):
        html = html_cache.get(q1_fy26_row["source_url"])
        if html:
            guide = extract_guidance_row(q1_fy26_row, html, carry)
            if guide:
                extracted.append(guide)

    rows = dedupe_rows(manifest_rows, extracted)
    rows = append_derived_rows(rows)
    validate_rows(rows, manifest_rows)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with output_path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=CANONICAL_FIELDS)
        writer.writeheader()
        writer.writerows(rows)
    return rows


def main() -> None:
    parser = argparse.ArgumentParser(description="Emit Amazon forecast_vintage.csv")
    parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument(
        "--build-timeseries",
        action="store_true",
        help="Also regenerate capacity_timeseries.csv and company_gw_chart.html",
    )
    args = parser.parse_args()
    rows = emit_rows(args.manifest, args.output)
    print(f"Wrote {len(rows)} rows to {args.output}")
    if args.build_timeseries:
        from build_timeseries import main as build_main

        build_main()


if __name__ == "__main__":
    main()