#!/usr/bin/env python3 from __future__ import annotations import csv import json import re from pathlib import Path from urllib.parse import urlparse from extract_dlr import SAMPLE_ROW_IDS, ensure_sources, extract_rows BASE = Path(__file__).resolve().parent ROW_FILE = BASE / "dlr_structured_rows.json" CSV_FILE = BASE / "dlr_structured_rows.csv" MANIFEST_FILE = BASE / "source_manifest.csv" CACHE_MANIFEST = BASE / "source_cache_manifest.json" RESULT_FILE = BASE / "validation_results.json" VERIFY_CSV = BASE / "verification_sample_checks.csv" VERIFY_REPORT = BASE / "verification_report.json" REQUIRED_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_REQUIRED_FIELDS = [ "source_id", "company", "source_url", "source_type", "period", "expected_table_or_section", "access_caveats", ] VREQ = [ "row_id", "company", "period", "metric_name", "raw_value", "normalized_value", "unit", "source_url", "source_location", "snippet", "verdict", "confidence", "caveat", ] ALLOWED_UNITS = {"MW", "USD", "percent", "months", "text"} ALLOWED_BASIS = { "100% share", "Digital Realty share", "consolidated", "consolidated development projects", "unconsolidated/JV basis", } ALLOWED_CLASS = {"forecast", "actual"} ALLOWED_QUALIFIERS = { "reported", "reported_caveat", "derived_qoq", "derived_sum", "range", "weighted_average_lag", "greater_than", "less_than_or_equal", "approximate", } ALLOWED_SNIPPET_STATUS = {"supported", "partially_supported", "unsupported"} def load_csv(path: Path) -> list[dict]: with path.open(newline="", encoding="utf-8") as f: return list(csv.DictReader(f)) def is_https(url: str) -> bool: parsed = urlparse(str(url)) return parsed.scheme == "https" and bool(parsed.netloc) def compact(text: str) -> str: return re.sub(r"\s+", " ", str(text)).replace("$", "").replace(",", "").replace("%", "").strip().lower() def load_cached_text_by_url() -> tuple[dict[str, str], list[str]]: errors: list[str] = [] by_url: dict[str, str] = {} if not CACHE_MANIFEST.exists(): errors.append("source_cache_manifest.json missing") return by_url, errors manifest = json.loads(CACHE_MANIFEST.read_text(encoding="utf-8")) for src in manifest: txt_path = BASE / src["text_cache_path"] if not txt_path.exists() or not txt_path.stat().st_size: errors.append(f"missing cached text: {src['text_cache_path']}") continue by_url[src["source_url"]] = txt_path.read_text(encoding="utf-8", errors="ignore") return by_url, errors def normalized_ok(row: dict) -> bool: value, unit, qualifier = row.get("normalized_value"), row.get("unit"), row.get("value_qualifier") if unit == "text": if isinstance(value, dict): return bool(str(value.get("display", "")).strip()) or bool(str(value.get("basis_gap_flag", "")).strip()) return isinstance(value, str) and bool(value.strip()) if qualifier == "range": return ( isinstance(value, list) and len(value) == 2 and all(isinstance(v, (int, float)) for v in value) and value[0] <= value[1] ) return isinstance(value, (int, float)) and not isinstance(value, bool) def snippet_in_cached_text(snippet: str, source_url: str, cached_by_url: dict[str, str]) -> bool: text = cached_by_url.get(source_url, "") if not text: return False return compact(snippet) in compact(text) def anchor_tokens(raw_value: str) -> list[str]: tokens = [] for part in re.findall( r"\d[\d,]*\.?\d*|\$[\d,]+\.?\d*\s*(?:billion|million|thousand)?|over one gigawatt|up to \d+|nineteen months|ownership percentages vary", str(raw_value), re.I, ): tokens.append(part.strip()) if not tokens: tokens.append(str(raw_value).strip()) return tokens def verify_row_snippet(row: dict, cached_by_url: dict[str, str]) -> tuple[bool, str]: snippet = str(row.get("source_snippet", "")) source_url = str(row.get("source_url", "")) if len(snippet.strip()) < 20: return False, "snippet too short" if not snippet_in_cached_text(snippet, source_url, cached_by_url): return False, f"snippet not found in cached text for {source_url}" if row.get("value_qualifier") in {"derived_qoq", "derived_sum", "reported_caveat"}: return True, "derived or caveat row anchored to cached source text" if row.get("snippet_support_status") in {"partially_supported", "unsupported"}: return True, f"gradation row ({row.get('snippet_support_status')}) — snippet presence verified only" for token in anchor_tokens(str(row.get("raw_value", ""))): if compact(token) not in compact(snippet) and compact(token) not in compact(str(row.get("raw_value", ""))): return False, f"anchor token missing: {token}" return True, "snippet and anchor tokens verified against cached primary-source text" def csv_json_equivalent(rows: list[dict], csv_rows: list[dict]) -> bool: if len(rows) != len(csv_rows): return False for json_row, csv_row in zip(rows, csv_rows): for field in REQUIRED_FIELDS: json_val = json_row.get(field) csv_val = csv_row.get(field) if field == "is_derived": if str(csv_val).lower() != str(json_val).lower(): return False continue if field == "normalized_value" and isinstance(json_val, (list, dict)): if json.loads(csv_val) != json_val: return False continue if str(json_val) != str(csv_val): return False return True def validate() -> dict: rows = json.loads(ROW_FILE.read_text(encoding="utf-8")) csv_rows = load_csv(CSV_FILE) manifest = load_csv(MANIFEST_FILE) cached_by_url, cache_errors = load_cached_text_by_url() checks, row_errors, verification_samples = [], [], [] def add(name: str, passed: bool, detail: str) -> None: checks.append({"check_name": name, "status": "pass" if passed else "fail", "detail": detail}) for idx, row in enumerate(rows, 1): missing = [f for f in REQUIRED_FIELDS if f not in row or row[f] in (None, "")] if missing: row_errors.append(f"row {idx} missing {missing}") if row.get("company") != "DLR": row_errors.append(f"row {idx} company not DLR") if not re.fullmatch(r"20\d{2}Q[1-4]", str(row.get("period", ""))): row_errors.append(f"row {idx} period not quarter format") if row.get("unit") not in ALLOWED_UNITS: row_errors.append(f"row {idx} unit not allowed: {row.get('unit')}") if row.get("ownership_basis") not in ALLOWED_BASIS: row_errors.append(f"row {idx} ownership_basis not allowed: {row.get('ownership_basis')}") if row.get("forecast_or_actual") not in ALLOWED_CLASS: row_errors.append(f"row {idx} forecast_or_actual invalid: {row.get('forecast_or_actual')}") if not isinstance(row.get("is_derived"), bool): row_errors.append(f"row {idx} is_derived must be boolean") if row.get("value_qualifier") not in ALLOWED_QUALIFIERS: row_errors.append(f"row {idx} value_qualifier not allowed: {row.get('value_qualifier')}") if row.get("snippet_support_status") not in ALLOWED_SNIPPET_STATUS: row_errors.append(f"row {idx} snippet_support_status invalid") if row.get("snippet_support_status") == "supported" and len(str(row.get("source_snippet", "")).strip()) < 20: row_errors.append(f"row {idx} supported snippet too short") if not is_https(str(row.get("source_url", ""))): row_errors.append(f"row {idx} source_url not https") if not normalized_ok(row): row_errors.append(f"row {idx} normalized_value invalid for {row.get('unit')} / {row.get('value_qualifier')}") for i in SAMPLE_ROW_IDS: if i > len(rows): continue row = rows[i - 1] ok, detail = verify_row_snippet(row, cached_by_url) verification_samples.append({ "row_index": i, "metric_name": row.get("metric_name"), "ownership_basis": row.get("ownership_basis"), "raw_value": row.get("raw_value"), "source_url": row.get("source_url"), "status": "pass" if ok else "fail", "detail": detail, "source_location": row.get("source_location"), }) texts = ensure_sources() replay_rows = extract_rows(texts) replay_ok = replay_rows == rows cited_urls = {r["source_url"] for r in rows} manifest_urls = {s["source_url"] for s in manifest} support_statuses = {r.get("snippet_support_status") for r in rows} add("required_fields", not any("missing" in e for e in row_errors), "All structured rows must populate production schema fields.") add("row_count", len(rows) >= 20 and len(csv_rows) == len(rows), f"json_rows={len(rows)} csv_rows={len(csv_rows)} minimum=20") add("csv_json_parity", csv_json_equivalent(rows, csv_rows), "CSV and JSON row files must be field-equivalent.") add("extractor_replay", replay_ok, "Re-running extract_dlr.extract_rows() must reproduce persisted JSON rows.") add("source_cache_manifest", CACHE_MANIFEST.exists() and not cache_errors, "Cached primary sources must be fetched and listed in source_cache_manifest.json.") add("source_urls", not any("source_url" in e for e in row_errors), "Every structured row must use a direct https primary-source URL.") add("unit_normalization", not any("unit not allowed" in e or "normalized_value invalid" in e for e in row_errors), "Numeric units must be normalized; text caveat rows may use structured normalized_value.") add("ownership_basis", not any("ownership_basis" in e for e in row_errors), "Rows must distinguish ownership bases including unconsolidated/JV caveat.") add("forecast_or_actual", not any("forecast_or_actual" in e for e in row_errors), "Rows must be marked forecast or actual.") add("derived_flag", not any("is_derived" in e for e in row_errors) and sum(1 for r in rows if r.get("is_derived")) >= 2, "Derived rows must be boolean and include QoQ/sum examples.") add("snippet_presence", not any("snippet" in e for e in row_errors), "Supported rows must include primary-source snippets.") add( "snippet_support_gradations", {"partially_supported", "unsupported"}.issubset(support_statuses), "Persisted rows must exercise supported, partially_supported, and unsupported gradations.", ) add( "value_qualifiers", not any("value_qualifier" in e for e in row_errors) and {"approximate", "greater_than", "range", "derived_qoq", "derived_sum"}.issubset({r.get("value_qualifier") for r in rows}), "Rows must exercise approximate, greater-than, range, and derived qualifiers.", ) add("actual_vs_forecast_mix", {"actual", "forecast"}.issubset({r.get("forecast_or_actual") for r in rows}), "Rows must include actual and forecast observations.") add("manifest_required_fields", all(all(src.get(f) for f in MANIFEST_REQUIRED_FIELDS) for src in manifest), "Source manifest must include direct primary URLs, type, period, expected section, and access caveat.") add("manifest_primary_urls", all(is_https(src.get("source_url", "")) for src in manifest), "Manifest source URLs must be direct https primary-source URLs.") add("manifest_covers_cited_urls", cited_urls.issubset(manifest_urls), "Every cited source URL must appear in source_manifest.csv.") add( "multi_source_citations", len(cited_urls) >= 3, f"Structured rows must cite multiple manifest sources; cited_url_count={len(cited_urls)}", ) add( "dlr_specific_basis_lessons", {"100% share", "Digital Realty share", "consolidated", "consolidated development projects", "unconsolidated/JV basis"}.issubset({r.get("ownership_basis") for r in rows}), "Schema must represent 100% vs DLR share, consolidated capex, and unconsolidated/JV caveat rows.", ) add( "independent_sample_verification", all(s["status"] == "pass" for s in verification_samples) and len(verification_samples) >= 12, "Twelve sample rows verified independently against persisted cached source text (not extractor constants).", ) vrows = load_csv(VERIFY_CSV) if VERIFY_CSV.exists() else [] verify_report = { "status": "pass" if len(vrows) >= 12 and all(r.get("verdict") == "PASS" for r in vrows) else "fail", "sampled_row_count": len(vrows), "pass_count": sum(1 for r in vrows if r.get("verdict") == "PASS"), "fail_count": sum(1 for r in vrows if r.get("verdict") != "PASS"), "verification_csv": "verification_sample_checks.csv", "source_cache_manifest": "source_cache_manifest.json", "notes": [ "Verification samples are checked against sources/text/*.txt files listed in source_cache_manifest.json.", "Validator does not import embedded PRIMARY_TEXT from extract_dlr.py.", ], } VERIFY_REPORT.write_text(json.dumps(verify_report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") result = { "status": "pass" if all(c["status"] == "pass" for c in checks) else "fail", "company": "DLR", "period": "2026Q1", "row_count": len(rows), "csv_row_count": len(csv_rows), "manifest_row_count": len(manifest), "cited_source_url_count": len(cited_urls), "checks": checks, "verification_samples": verification_samples, "verification_report": verify_report, "row_errors": row_errors, "ownership_basis_counts": {b: sum(1 for r in rows if r.get("ownership_basis") == b) for b in sorted(ALLOWED_BASIS)}, "value_qualifier_counts": {q: sum(1 for r in rows if r.get("value_qualifier") == q) for q in sorted({r.get("value_qualifier") for r in rows})}, "forecast_or_actual_counts": {k: sum(1 for r in rows if r.get("forecast_or_actual") == k) for k in sorted(ALLOWED_CLASS)}, "snippet_support_status_counts": {k: sum(1 for r in rows if r.get("snippet_support_status") == k) for k in sorted(ALLOWED_SNIPPET_STATUS)}, } RESULT_FILE.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") return result if __name__ == "__main__": report = validate() if report["status"] != "pass": raise SystemExit(1)