#!/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"} 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(): with (BASE / "verification_sample_checks.csv").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(): return "", ["source_cache_manifest.json missing"] 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: errors.append(f"cached source text missing: {src['text_cache_path']}") 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()