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

Reads source_manifest.csv, fetches primary SEC/IR HTML, and parses capex actuals
from cash-flow statements plus FY2026 capex-guidance vintages from earnings-call
prepared remarks. Microsoft does not disclose MW/GW; normalized_gw is blank.
"""

from __future__ import annotations

import argparse
import csv
import re
import urllib.request
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 = (
    "Microsoft does not disclose data-center MW/GW. Row tracks USD capex proxy only; "
    "normalized_gw intentionally blank."
)

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

FY2025_ANNUAL_CAPEX_B = 64.551

# Regex anchors for verbatim snippets on company-hosted earnings webcast pages
# (repeatable source family: Microsoft IR transcript pages; not filed on SEC).
GUIDANCE_SNIPPET_PATTERNS: dict[str, list[str]] = {
    "msft_fy2025_q4_earnings_webcast": [
        (
            r"Capital expenditure growth, as we shared last quarter, will moderate "
            r"compared to FY25 with a greater mix of short-lived assets\."
        ),
    ],
    "msft_fy2026_q1_earnings_webcast": [
        (
            r"Therefore, total spend will increase sequentially, and we now expect the "
            r"FY26 growth rate to be higher than FY25\."
        ),
    ],
    "msft_fy2026_q2_earnings_webcast": [
        (
            r"Capital expenditures were \$37\.5 billion, and this quarter, roughly two "
            r"thirds of our capex was on short-lived assets, primarily GPUs and CPUs\."
        ),
    ],
    "msft_fy2026_q3_earnings_webcast": [
        (
            r"Capital expenditures were \$31\.9 billion, down sequentially due to the "
            r"normal variability from cloud infrastructure buildouts and the timing of "
            r"delivery of finance leases\."
        ),
    ],
}

Q1_FLOOR_SNIPPET_PATTERN = (
    r"We expect Q1 capital expenditures to be over \$30 billion driven by the "
    r"continued strong demand signals we see\."
)

SEC_URL_BY_PERIOD: dict[str, str] = {
    "FY2025-FQ4": (
        "https://www.sec.gov/Archives/edgar/data/789019/"
        "000095017025100235/msft-20250630.htm"
    ),
    "FY2026-FQ2": (
        "https://www.sec.gov/Archives/edgar/data/789019/"
        "000119312526027207/msft-20251231.htm"
    ),
    "FY2026-FQ3": (
        "https://www.sec.gov/Archives/edgar/data/789019/"
        "000119312526191507/msft-20260331.htm"
    ),
}

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


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>", " ", text, flags=re.I | re.S)
    text = re.sub(r"<[^>]+>", " ", 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)
        if match:
            return match.group(0)
    return None


def capex_millions_snippet(value_b: float) -> str:
    millions = int(round(value_b * 1000))
    return f"Additions to property and equipment ({millions:,})"


def guidance_metadata(
    source_id: str,
    snippet: str,
    carry_forward_value: str | None = None,
) -> dict[str, str]:
    if source_id == "msft_fy2025_q4_earnings_webcast":
        return {
            "raw_value": "77.5",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                "Microsoft gives qualitative FY2026 capex growth guidance only. "
                "Implied ~$77.5B derived as FY2025 actual $64.551B × 1.20 (~20% YoY, "
                "interpretation of 'moderate' vs FY2025's 45% growth). Not MW/GW capacity."
            ),
        }
    if source_id == "msft_fy2026_q1_earnings_webcast":
        return {
            "raw_value": "96.8",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                "FY2026 capex guide revised upward qualitatively. Implied ~$96.8B derived "
                "as FY2025 $64.551B × 1.50 (~50% YoY, above FY2025's 45% growth rate). "
                "Not MW/GW capacity."
            ),
        }
    if source_id in ("msft_fy2026_q2_earnings_webcast", "msft_fy2026_q3_earnings_webcast"):
        quarter = "FQ2" if "q2" in source_id else "FQ3"
        return {
            "raw_value": carry_forward_value or "96.8",
            "is_derived": "1",
            "confidence": "medium",
            "caveat": (
                f"No new full-year FY2026 numeric capex guide at {quarter}; vintage carries "
                f"forward Q1 FY2026 implied ~${carry_forward_value or '96.8'}B. "
                "Snippet is quarter-specific capex commentary from earnings call. "
                "Not MW/GW capacity."
            ),
        }
    raise ValueError(f"Unknown guidance source_id: {source_id}")


def extract_call_guidance(
    manifest_row: dict[str, str],
    html: str,
    carry_forward_value: 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, snippet, carry_forward_value)
    period_label = manifest_row["period"].replace("FY", "FY ").replace("-FQ", " Q")
    return {
        "raw_value": meta["raw_value"],
        "source_location": f"{period_label} earnings call prepared remarks (Amy Hood)",
        "source_snippet": snippet,
        "is_derived": meta["is_derived"],
        "confidence": meta["confidence"],
        "caveat": meta["caveat"],
    }


def billions_from_millions(raw: str) -> str:
    value = int(raw.replace(",", ""))
    return f"{value / 1000:.3f}".rstrip("0").rstrip(".")


def parse_capex_values(html: str) -> list[int]:
    values: list[int] = []
    for match in re.finditer(
        r'PaymentsToAcquirePropertyPlantAndEquipment"[^>]*>([0-9,]+)<',
        html,
    ):
        values.append(int(match.group(1).replace(",", "")))
    return values


def extract_annual_capex(html: str, fiscal_year_label: str) -> tuple[str, str] | None:
    values = parse_capex_values(html)
    if fiscal_year_label == "FY2025" and values:
        raw = f"{values[0]:,}"
        return billions_from_millions(raw), f"Additions to property and equipment ({raw})"
    if fiscal_year_label == "FY2024" and values:
        raw = f"{values[0]:,}"
        return billions_from_millions(raw), f"Additions to property and equipment ({raw})"
    return None


def extract_quarterly_capex(
    html: str,
    derived: bool,
    prior_ytd_millions: int | None,
) -> tuple[str, str, str, int] | None:
    values = parse_capex_values(html)
    if not values:
        return None
    if not derived:
        raw = f"{values[0]:,}"
        return (
            billions_from_millions(raw),
            f"Additions to property and equipment ({raw})",
            "0",
            values[0],
        )
    if len(values) >= 3 and prior_ytd_millions is not None:
        current_ytd = values[2]
        quarter = current_ytd - prior_ytd_millions
        raw = f"{quarter:,}"
        return (
            billions_from_millions(raw),
            f"Additions to property and equipment ({raw})",
            "1",
            current_ytd,
        )
    if prior_ytd_millions is not None:
        current_ytd = values[0]
        quarter = current_ytd - prior_ytd_millions
        raw = f"{quarter:,}"
        return (
            billions_from_millions(raw),
            f"Additions to property and equipment ({raw})",
            "1",
            current_ytd,
        )
    return None


def extract_ytd_capex(html: str, label: str) -> tuple[str, str] | None:
    values = parse_capex_values(html)
    if not values:
        return None
    if len(values) >= 3:
        raw = f"{values[2]:,}"
    else:
        raw = f"{values[0]:,}"
    return billions_from_millions(raw), f"Additions to property and equipment YTD ({raw})"


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_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 doc_type == "earnings_call_webcast":
        if "capex_guidance_fy2026" not in metrics:
            return rows
        call = extract_call_guidance(manifest_row, html, carry_forward_guidance)
        if not call:
            raise ValueError(
                f"Could not extract FY2026 guidance snippet from {url} "
                f"({manifest_row['source_id']})"
            )
        rows.append(
            base_row(
                manifest_row,
                target_period="FY2026",
                metric_name="capex_guidance",
                raw_value=call["raw_value"],
                forecast_or_actual="forecast",
                is_derived=call["is_derived"],
                source_location=call["source_location"],
                source_snippet=call["source_snippet"],
                snippet_support_status="verified_primary",
                confidence=call["confidence"],
                caveat=call["caveat"],
            )
        )
        return rows

    if "capex_actual_fy" in metrics and doc_type == "sec_10k":
        if manifest_row["period"] == "FY2024-FQ4":
            parsed = extract_annual_capex(html, "FY2024")
            if parsed:
                value, snippet = parsed
                rows.append(
                    base_row(
                        manifest_row,
                        target_period="FY2024",
                        metric_name="capex_actual",
                        raw_value=value,
                        forecast_or_actual="actual",
                        is_derived="0",
                        source_location=(
                            "Consolidated Statements of Cash Flows; "
                            "FY2024 additions to property and equipment"
                        ),
                        source_snippet=snippet,
                        snippet_support_status="verified_primary",
                        confidence="high",
                        caveat=GW_PROXY_CAVEAT,
                    )
                )
        if manifest_row["period"] == "FY2025-FQ4":
            parsed = extract_annual_capex(html, "FY2025")
            if parsed:
                value, snippet = parsed
                rows.append(
                    base_row(
                        manifest_row,
                        target_period="FY2025",
                        metric_name="capex_actual",
                        raw_value=value,
                        forecast_or_actual="actual",
                        is_derived="0",
                        source_location=(
                            "Consolidated Statements of Cash Flows; "
                            "FY2025 additions to property and equipment"
                        ),
                        source_snippet=snippet,
                        snippet_support_status="verified_primary",
                        confidence="high",
                        caveat=GW_PROXY_CAVEAT,
                    )
                )

    if "capex_actual_ytd" in metrics:
        period = manifest_row["period"]
        ytd_label = {
            "FY2026-FQ2": "FY2026-H1",
            "FY2026-FQ3": "FY2026-9M",
        }.get(period, period)
        parsed = extract_ytd_capex(html, ytd_label)
        if parsed:
            value, snippet = parsed
            values = parse_capex_values(html)
            ytd_millions = values[2] if len(values) >= 3 else values[0]
            ytd_state[period.split("-")[0]] = ytd_millions
            rows.append(
                base_row(
                    manifest_row,
                    target_period=ytd_label,
                    metric_name="capex_actual_ytd",
                    raw_value=value,
                    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 or "capex_actual_quarter_derived" in metrics:
        derived = "capex_actual_quarter_derived" in metrics
        fiscal_year = manifest_row["period"].split("-")[0]
        prior_ytd = ytd_state.get(fiscal_year)
        parsed = extract_quarterly_capex(html, derived=derived, prior_ytd_millions=prior_ytd)
        if parsed:
            value, snippet, is_derived, ytd_millions = parsed
            ytd_state[fiscal_year] = ytd_millions
            rows.append(
                base_row(
                    manifest_row,
                    target_period=manifest_row["period"],
                    metric_name="capex_actual",
                    raw_value=value,
                    forecast_or_actual="actual",
                    is_derived=is_derived,
                    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]],
    html_cache: dict[str, str],
) -> list[dict[str, str]]:
    """Add derived quarterly actuals and Q1 FY2026 capex floor guidance."""
    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,
        metric_name: str,
        value: float,
        source_url: str,
        source_location: str,
        source_snippet: str,
        caveat: str,
    ) -> None:
        template = next((r for r in rows if r["as_of_period"] == template_period), rows[0])
        key = (template_period, target_period, metric_name)
        if key in by_key:
            return
        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=metric_name,
            raw_value=raw,
            forecast_or_actual="actual" if "actual" in metric_name else "forecast",
            is_derived="1",
            source_location=source_location,
            source_snippet=source_snippet,
            snippet_support_status="verified_primary",
            confidence="high" if "actual" in metric_name else "medium",
            caveat=caveat,
            source_url=source_url,
        )
        rows.append(row)
        by_key[key] = row

    annual_fy25 = get_value("FY2025-FQ4", "FY2025", "capex_actual")
    if annual_fy25 is not None:
        nine_m_total = 0.0
        for p in ("FY2025-FQ1", "FY2025-FQ2", "FY2025-FQ3"):
            v = get_value(p, p, "capex_actual")
            if v is not None:
                nine_m_total += v
        if nine_m_total > 0:
            q4 = annual_fy25 - nine_m_total
            add_derived(
                "FY2025-FQ4",
                "FY2025-FQ4",
                "capex_actual",
                q4,
                SEC_URL_BY_PERIOD["FY2025-FQ4"],
                "Derived from FY2025 annual minus Q1–Q3 quarterly actuals",
                capex_millions_snippet(q4),
                GW_PROXY_CAVEAT,
            )

    q1_fy26 = get_value("FY2026-FQ1", "FY2026-FQ1", "capex_actual")
    h1_fy26 = get_value("FY2026-FQ2", "FY2026-H1", "capex_actual_ytd")
    nine_m_fy26 = get_value("FY2026-FQ3", "FY2026-9M", "capex_actual_ytd")
    if q1_fy26 is not None and h1_fy26 is not None:
        q2_val = h1_fy26 - q1_fy26
        add_derived(
            "FY2026-FQ2",
            "FY2026-FQ2",
            "capex_actual",
            q2_val,
            SEC_URL_BY_PERIOD["FY2026-FQ2"],
            "Derived from H1 YTD minus Q1 FY2026 (SEC 10-Q cash-flow YTD)",
            capex_millions_snippet(q2_val),
            GW_PROXY_CAVEAT,
        )
    if h1_fy26 is not None and nine_m_fy26 is not None:
        q3_val = nine_m_fy26 - h1_fy26
        add_derived(
            "FY2026-FQ3",
            "FY2026-FQ3",
            "capex_actual",
            q3_val,
            SEC_URL_BY_PERIOD["FY2026-FQ3"],
            "Derived from 9M YTD minus H1 FY2026 (SEC 10-Q cash-flow YTD)",
            capex_millions_snippet(q3_val),
            GW_PROXY_CAVEAT,
        )

    q4_guide = by_key.get(("FY2025-FQ4", "FY2026", "capex_guidance"))
    if q4_guide and ("FY2025-FQ4", "FY2026-FQ1", "capex_guidance") not in by_key:
        q4_url = q4_guide["source_url"]
        html = html_cache.get(q4_url)
        if html is None:
            html = fetch_html(q4_url)
            html_cache[q4_url] = html
        floor_snippet = find_snippet_in_text(html_to_text(html), [Q1_FLOOR_SNIPPET_PATTERN])
        if not floor_snippet:
            raise ValueError(f"Could not extract Q1 FY2026 capex floor snippet from {q4_url}")
        rows.append(
            base_row(
                {
                    "company": q4_guide["company"],
                    "ticker": q4_guide["ticker"],
                    "period": "FY2025-FQ4",
                    "source_url": q4_url,
                },
                target_period="FY2026-FQ1",
                metric_name="capex_guidance",
                raw_value="30",
                forecast_or_actual="forecast",
                is_derived="0",
                source_location="Q4 FY2025 earnings call prepared remarks (Amy Hood)",
                source_snippet=floor_snippet,
                snippet_support_status="verified_primary",
                confidence="high",
                caveat=(
                    "Quarterly Q1 FY2026 capex floor guidance (> $30B); not full-year "
                    "FY2026 guide. Not MW/GW capacity."
                ),
            )
        )

    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"] != "MSFT":
            raise ValueError(f"Row {idx} ticker must be MSFT")
        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" and r["target_period"] == "FY2026"]
    if len(guidance) < 4:
        raise ValueError(f"Need >=4 FY2026 capex_guidance vintages, got {len(guidance)}")
    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]] = []
    carry_forward: str | None = None

    # Process non-earnings rows first so YTD state and carry-forward value exist.
    for manifest_row in manifest_rows:
        if manifest_row["document_type"] == "earnings_call_webcast":
            continue
        extracted.extend(extract_from_manifest_row(manifest_row, html_cache, ytd_state))

    interim = dedupe_rows(manifest_rows, extracted)
    q1_guide = next(
        (
            r
            for r in interim
            if r["metric_name"] == "capex_guidance"
            and r["target_period"] == "FY2026"
            and r["as_of_period"] == "FY2026-FQ1"
        ),
        None,
    )

    for manifest_row in manifest_rows:
        if manifest_row["document_type"] != "earnings_call_webcast":
            continue
        cf = carry_forward
        if manifest_row["source_id"] in (
            "msft_fy2026_q2_earnings_webcast",
            "msft_fy2026_q3_earnings_webcast",
        ):
            cf = q1_guide["raw_value"] if q1_guide else "96.8"
        extracted.extend(
            extract_from_manifest_row(manifest_row, html_cache, ytd_state, cf)
        )

    rows = dedupe_rows(manifest_rows, extracted)
    rows = append_derived_rows(rows, html_cache)
    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 Microsoft 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()