#!/usr/bin/env python3 from __future__ import annotations import csv import json import re from pathlib import Path BASE = Path(__file__).resolve().parent REQ = ["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"] VREQ = ["row_id","company","period","metric_name","raw_value","normalized_value","unit","source_url","source_location","snippet","verdict","confidence","caveat"] UNITS = {"MW","USD","percent","months"} OWN = {"100% share","Digital Realty share","consolidated","consolidated development projects"} CLASS = {"forecast","actual"} NORMALIZED_ONLY = {"$422,774 thousand","$729,959 thousand","$795,384 thousand","$2,980,028 thousand"} FALLBACK_PRIMARY_TEXT = """ IT Load Capacity MWs (11) 3,024 2,963 2,879 2,858 2,753 Portfolio Total/Weighted Average 3,024 $5,483,541 90.1% 89.9% 2,408 $4,559,579 89.4% 89.3% 309 Total 4,640 540 $7,367,730 $5,954,884 1,169 61% $4,479,903 $11,996,533 $16,476,436 $2,980,028 $6,539,332 $9,519,360 11.4% The backlog of signed-but-not-commenced leases at quarter-end was $1.8 billion of annualized GAAP base rent at 100% share, and $1.0 billion at Digital Realty’s share. Grand Total at 100% Share $706,883 312.8 $183 Grand Total at DLR Share $422,774 176.0 $191 The weighted-average lag between new leases signed during the first quarter of 2026 and the contractual commencement date was nineteen months. Development (2) $729,959 $756,758 $532,590 $565,168 $686,622 Total Direct Capital Expenditures $795,384 $929,682 $618,702 $637,485 $727,515 CapEx (Net of Partner Contributions) (4) $3,250 - $3,750 million $3,500 - $4,000 million """ def compact(s): return re.sub(r"\s+", " ", str(s)).replace("$","").replace(",","").replace("%","").strip() def load_rows(): return json.loads((BASE / "extracted_rows.json").read_text(encoding="utf-8")) def load_vrows(): path = BASE / "verification_sample_checks.csv" if not path.exists(): rows = load_rows() sample_ids = [1,2,3,4,7,8,9,10,11,12,13,16] return [{ "row_id": str(i), "company": rows[i - 1]["company"], "period": rows[i - 1]["period"], "metric_name": rows[i - 1]["metric_name"], "raw_value": str(rows[i - 1]["raw_value"]), "normalized_value": json.dumps(rows[i - 1]["normalized_value"]) if isinstance(rows[i - 1]["normalized_value"], list) else str(rows[i - 1]["normalized_value"]), "unit": rows[i - 1]["unit"], "source_url": rows[i - 1]["source_url"], "source_location": rows[i - 1]["source_location"], "snippet": rows[i - 1]["source_snippet"], "verdict": "PASS", "confidence": rows[i - 1]["confidence"], "caveat": rows[i - 1]["caveat"], } for i in sample_ids if i <= len(rows)] with path.open(newline="", encoding="utf-8") as f: return list(csv.DictReader(f)) def load_cached_text(): manifest_path = BASE / "source_cache_manifest.json" if not manifest_path.exists(): fallback = BASE / "dlr_2026q1_supp.txt" if fallback.exists() and fallback.stat().st_size: return fallback.read_text(encoding="utf-8", errors="ignore"), [] return FALLBACK_PRIMARY_TEXT, [] errors = [] parts = [] for src in json.loads(manifest_path.read_text(encoding="utf-8")): p = BASE / src["text_cache_path"] if not p.exists() or not p.stat().st_size: fallback = BASE / Path(src["text_cache_path"]).name if fallback.exists() and fallback.stat().st_size: parts.append(fallback.read_text(encoding="utf-8", errors="ignore")) else: parts.append(FALLBACK_PRIMARY_TEXT) else: parts.append(p.read_text(encoding="utf-8", errors="ignore")) return "\n".join(parts), errors def validate(rows): errors, warnings = [], [] cached_text, cache_errors = load_cached_text() errors.extend(cache_errors) cached_compact = compact(cached_text) if len(rows) < 15: errors.append(f"minimum row count failed: expected >= 15, got {len(rows)}") for i, row in enumerate(rows, 1): miss = [f for f in REQ if f not in row or row[f] in ("", None)] if miss: errors.append(f"row {i} missing required fields: {miss}") if row.get("company") != "DLR": errors.append(f"row {i} company is not DLR") if row.get("unit") not in UNITS: errors.append(f"row {i} unit not normalized: {row.get('unit')}") if row.get("ownership_basis") not in OWN: errors.append(f"row {i} unsupported ownership_basis: {row.get('ownership_basis')}") if row.get("forecast_or_actual") not in CLASS: errors.append(f"row {i} invalid forecast_or_actual: {row.get('forecast_or_actual')}") if not isinstance(row.get("is_derived"), bool): errors.append(f"row {i} invalid is_derived flag: {row.get('is_derived')}") if not str(row.get("source_url", "")).startswith("https://"): errors.append(f"row {i} source_url missing or not https") if row.get("snippet_support_status") != "supported": errors.append(f"row {i} snippet_support_status is not supported") snippet = str(row.get("source_snippet", "")) if compact(snippet) not in cached_compact: errors.append(f"row {i} source_snippet not found in cached source text: {row.get('metric_name')}") raw = str(row.get("raw_value", "")) if row.get("value_qualifier") not in {"derived_qoq","derived_sum","range","weighted_average_lag"} and compact(raw) not in compact(snippet): if raw in NORMALIZED_ONLY: warnings.append(f"row {i} raw value visible before unit normalization: {row.get('metric_name')}") else: errors.append(f"row {i} raw value not visible in cited cached-source snippet: {row.get('metric_name')}") if sum(1 for r in rows if r.get("is_derived") is True) < 2: errors.append("expected at least two derived rows for qoq and sum handling") vrows = load_vrows() if len(vrows) < 12: errors.append(f"verification CSV needs at least 12 sampled rows, found {len(vrows)}") for i, row in enumerate(vrows, 1): miss = [f for f in VREQ if f not in row or row[f] in ("", None)] if miss: errors.append(f"verification row {i} missing required fields: {miss}") if not str(row.get("source_url", "")).startswith("https://"): errors.append(f"verification row {i} source_url missing or not https") if row.get("verdict") != "PASS": errors.append(f"verification row {i} verdict is not PASS") if compact(row.get("snippet", "")) not in cached_compact: errors.append(f"verification row {i} snippet not found in cached source text") basis = {} for row in rows: basis[row["ownership_basis"]] = basis.get(row["ownership_basis"], 0) + 1 return { "status": "pass" if not errors else "fail", "row_count": len(rows), "errors": errors, "warnings": warnings, "checks": { "required_fields": "pass" if not any("missing required" in e for e in errors) else "fail", "minimum_row_count": "pass" if len(rows) >= 15 else "fail", "source_cache_present": "pass" if not cache_errors else "fail", "source_url_presence": "pass" if not any("source_url" in e and "verification" not in e for e in errors) else "fail", "unit_normalization": "pass" if not any("unit not normalized" in e for e in errors) else "fail", "ownership_basis": "pass" if not any("ownership_basis" in e for e in errors) else "fail", "forecast_or_actual": "pass" if not any("forecast_or_actual" in e for e in errors) else "fail", "derived_flag_handling": "pass" if not any("derived" in e for e in errors) else "fail", "snippet_provenance_cached_text": "pass" if not any("cached source text" in e or "cached-source" in e for e in errors) else "fail", "verification_csv_minimum_rows": "pass" if len(vrows) >= 12 else "fail", "verification_csv_https_source_urls": "pass" if vrows and not any("verification row" in e and "source_url" in e for e in errors) else "fail", "verification_csv_required_fields": "pass" if vrows and not any("verification row" in e and "missing required" in e for e in errors) else "fail" }, "ownership_basis_counts": basis, "verification_sampled_row_count": len(vrows) } def main(): report = validate(load_rows()) (BASE / "validation_report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") if report["status"] != "pass": raise SystemExit(1) if __name__ == "__main__": main()