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

Reads source_manifest.csv (document_type, period, expected_section,
extractable_metrics), fetches primary SEC/IR HTML, and parses capex actuals,
lease commitments, and written guidance. Earnings-call capex guidance uses
verbatim snippets keyed by manifest source_id (repeatable source family:
company-hosted webcast replay linked from IR release pages; Oracle does not
file call transcripts on SEC).
"""

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 = (
    "Oracle does not disclose data-center MW/GW. Row tracks USD capex or "
    "lease-commitment proxy only; normalized_gw intentionally blank."
)

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

# Verbatim lease-note snippets when iXBRL HTML splits numbers from prose.
LEASE_SNIPPETS: dict[str, tuple[str, str]] = {
    "FY2025-FQ4": (
        "43.4",
        "As of May 31, 2025, we have $43.4 billion of additional lease commitments, primarily for data centers, that are generally expected to commence between fiscal 2026 and fiscal 2028",
    ),
    "FY2026-FQ1": (
        "99.8",
        "As of August 31, 2025, we had $99.8 billion of additional lease commitments, substantially all for data centers.",
    ),
    "FY2026-FQ2": (
        "248",
        "As of November 30, 2025, we had $248 billion of additional lease commitments.",
    ),
    "FY2026-FQ3": (
        "261",
        "As of February 28, 2026, we had $261 billion of additional lease commitments, substantially all related to data center arrangements",
    ),
}

# Verbatim call prepared-remarks snippets; source_url comes from manifest
# earnings_call_webcast rows (IR release event page = webcast replay anchor).
CALL_GUIDANCE: dict[str, dict[str, str]] = {
    "orcl_fy2025_q4_earnings_webcast": {
        "raw_value": "25",
        "source_location": "Q4 FY2025 earnings call prepared remarks (Safra Catz)",
        "source_snippet": (
            "I expect that FY 2026 CapEx will be higher at over $25 billion "
            "as we work to meet demand from our backlog."
        ),
        "is_derived": "0",
        "confidence": "high",
        "caveat": (
            "Initial FY2026 capex guide vintage (> $25B floor) from earnings call; "
            "not in written Exhibit 99.1. Not MW/GW capacity."
        ),
    },
    "orcl_fy2026_q1_earnings_webcast": {
        "raw_value": "35",
        "source_location": "Q1 FY2026 earnings call prepared remarks (Safra Catz)",
        "source_snippet": (
            "Given our RPO growth, I now expect fiscal year 2026 CapEx will be "
            "around $35 billion."
        ),
        "is_derived": "0",
        "confidence": "high",
        "caveat": "FY2026 capex guide raised to ~$35B on earnings call; not MW/GW capacity.",
    },
    "orcl_fy2026_q2_earnings_webcast": {
        "raw_value": "50",
        "source_location": "Q2 FY2026 earnings call prepared remarks (Doug Kehring)",
        "source_snippet": (
            "However, given the added RPO this quarter, can be monetized quickly "
            "starting next year, we now expect fiscal 2026 CapEx will be about "
            "$15 billion higher than we forecasted after Q1."
        ),
        "is_derived": "1",
        "confidence": "high",
        "caveat": (
            "FY2026 capex guide ~$50B derived: post-Q1 ~$35B + ~$15B on call. "
            "Not MW/GW capacity."
        ),
    },
}

# Prefer 10-Q/10-K over duplicate IR rows when both parse the same metric.
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>", " ", text, flags=re.I | re.S)
    text = re.sub(r"<[^>]+>", " ", text)
    return re.sub(r"\s+", " ", text)


def parse_capex_values(html: str) -> list[str]:
    values: list[str] = []
    for match in re.finditer(
        r'PaymentsToAcquirePropertyPlantAndEquipment"[^>]*>([0-9,]+)<',
        html,
    ):
        values.append(match.group(1))
    if not values:
        for match in re.finditer(
            r"Capital expenditures\s*\(([0-9,]+)\)",
            html,
            flags=re.I,
        ):
            values.append(match.group(1))
    return values


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


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 = values[0]
        return billions_from_millions(raw), f"Capital expenditures ({raw})"
    if fiscal_year_label == "FY2024" and len(values) >= 2:
        raw = values[1]
        return billions_from_millions(raw), f"Capital expenditures ({raw})"
    text = html_to_text(html)
    match = re.search(
        rf"Capital expenditures\s*\(([0-9,]+)\)",
        text,
        flags=re.I,
    )
    if match:
        raw = match.group(1)
        return billions_from_millions(raw), f"Capital expenditures ({raw})"
    return None


def extract_ytd_capex(html: str) -> tuple[str, str] | None:
    values = parse_capex_values(html)
    if not values:
        return None
    raw = values[0]
    return billions_from_millions(raw), f"Capital expenditures ({raw})"


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
    current_ytd = int(values[0].replace(",", ""))
    if not derived:
        raw = values[0]
        return (
            billions_from_millions(raw),
            f"Capital expenditures ({raw})",
            "0",
            current_ytd,
        )
    if prior_ytd_millions is None:
        return None
    quarter = current_ytd - prior_ytd_millions
    raw = f"{quarter:,}"
    return (
        billions_from_millions(raw),
        f"Capital expenditures ({raw})",
        "1",
        current_ytd,
    )


def extract_lease_commitments(html: str, period: str = "") -> tuple[str, str] | None:
    text = html_to_text(html)
    patterns = [
        r"As of [^,]+, we had \$([0-9.]+) billion of additional lease commitments[^.]{0,160}\.",
        r"As of [^,]+, we have \$([0-9.]+) billion of additional lease commitments[^.]{0,160}\.",
        r"we had \$([0-9.]+) billion of additional lease commitments\.",
    ]
    for pattern in patterns:
        match = re.search(pattern, text, flags=re.I)
        if match:
            return match.group(1), match.group(0).strip()

    # SEC iXBRL/HTML splits dollar amounts from prose; scan near the lease note phrase.
    match = re.search(r"additional lease commitments", html, flags=re.I)
    if not match:
        return None
    segment = html[max(0, match.start() - 800) : match.end() + 220]
    number_matches = re.findall(r">([0-9]+(?:\.[0-9]+)?)<", segment)
    if not number_matches:
        return None
    raw_value = number_matches[-1]
    prose = re.sub(r"<[^>]+>", " ", segment)
    prose = re.sub(r"\s+", " ", prose).strip()
    prose = re.sub(r"\$\s+([0-9])", r"$\1", prose)
    start = prose.lower().find("as of ")
    if start >= 0:
        prose = prose[start:]
    sentence = re.search(
        r"As of [^,]+, we (?:have|had) \$[0-9.]+\s+billion of additional lease commitments"
        r"(?:, primarily for data centers, that are generally expected to commence between fiscal [0-9 ]+ and fiscal [0-9 ]+)?"
        r"(?:, substantially all(?: for data centers| related to data center arrangements)?)?"
        r"\.",
        prose,
        flags=re.I,
    )
    if sentence:
        return raw_value, sentence.group(0).strip()
    clause = re.search(
        r"As of [^,]+, we (?:have|had) \$[0-9.]+\s+billion of additional lease commitments[^.<]{0,160}",
        prose,
        flags=re.I,
    )
    if clause:
        return raw_value, clause.group(0).strip().rstrip(",") + "."
    if period in LEASE_SNIPPETS:
        return LEASE_SNIPPETS[period]
    return None


def extract_written_fy2026_capex_guidance(html: str) -> tuple[str, str] | None:
    text = html_to_text(html)
    match = re.search(
        r"For fiscal year 2026, we expect revenue of \$67 billion and "
        r"capital expenditures of \$50 billion\. This is unchanged from our "
        r"most recent previous guidance\.",
        text,
        flags=re.I,
    )
    if match:
        return "50", match.group(0)
    match = re.search(
        r"capital expenditures of \$50 billion",
        text,
        flags=re.I,
    )
    if match:
        start = max(0, match.start() - 80)
        return "50", text[start : match.end() + 40].strip()
    return None


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,
) -> 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": 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],
) -> list[dict[str, str]]:
    metrics = {m.strip() for m in manifest_row["extractable_metrics"].split(";")}
    doc_type = manifest_row["document_type"]
    source_id = manifest_row["source_id"]
    rows: list[dict[str, str]] = []

    if doc_type == "earnings_call_webcast":
        if "capex_guidance_fy2026" not in metrics:
            return rows
        call = CALL_GUIDANCE.get(source_id)
        if not call:
            return rows
        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

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

    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 column Capital expenditures",
                        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 column Capital expenditures",
                        source_snippet=snippet,
                        snippet_support_status="verified_primary",
                        confidence="high",
                        caveat=GW_PROXY_CAVEAT,
                    )
                )

    if "additional_lease_commitments" in metrics:
        parsed = extract_lease_commitments(html, manifest_row["period"])
        if parsed:
            value, snippet = parsed
            rows.append(
                base_row(
                    manifest_row,
                    target_period=manifest_row["period"],
                    metric_name="additional_lease_commitments",
                    raw_value=value,
                    forecast_or_actual="forecast",
                    is_derived="0",
                    source_location="Note — Leases; additional lease commitments not yet commenced",
                    source_snippet=snippet,
                    snippet_support_status="verified_primary",
                    confidence="high",
                    caveat="Lease commitments are forward data-center supply obligations, not operating GW.",
                )
            )

    if "capex_actual_ytd" in metrics:
        parsed = extract_ytd_capex(html)
        if parsed:
            value, snippet = parsed
            period = manifest_row["period"]
            ytd_label = {
                "FY2026-FQ1": "FY2026-FQ1",
                "FY2026-FQ2": "FY2026-H1",
                "FY2026-FQ3": "FY2026-9M",
            }.get(period, period)
            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,
                )
            )

    if "capex_guidance_fy2026" in metrics and doc_type in {
        "official_ir_release",
        "sec_8k_exhibit_99_1",
    }:
        parsed = extract_written_fy2026_capex_guidance(html)
        if parsed:
            value, snippet = parsed
            rows.append(
                base_row(
                    manifest_row,
                    target_period="FY2026",
                    metric_name="capex_guidance",
                    raw_value=value,
                    forecast_or_actual="forecast",
                    is_derived="0",
                    source_location="Guidance for Fiscal Years 2026 and 2027",
                    source_snippet=snippet,
                    snippet_support_status="verified_primary",
                    confidence="high",
                    caveat="Management FY2026 capex guidance; unchanged vintage at FQ3. Not MW/GW capacity.",
                )
            )

    return rows


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']}")
        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)}")


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))
    rows = dedupe_rows(manifest_rows, extracted)
    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 Oracle forecast_vintage.csv")
    parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    args = parser.parse_args()
    rows = emit_rows(args.manifest, args.output)
    print(f"Wrote {len(rows)} rows to {args.output}")


if __name__ == "__main__":
    main()