mo
1390 lines
61 KiB
Python
1390 lines
61 KiB
Python
"""Data Maturity Assessment API — Docling, Great Expectations, Soda Core."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import uuid
|
||
import base64
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pandas as pd
|
||
from fastapi import FastAPI, File, Form, UploadFile
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.responses import HTMLResponse, JSONResponse, FileResponse
|
||
from pydantic import BaseModel
|
||
|
||
DOCLING_URL = os.getenv("DOCLING_URL", "http://docling-serve:5001").rstrip("/")
|
||
RAG_URL = os.getenv("RAG_URL", "http://rag-api:5020").rstrip("/")
|
||
RAG_COLLECTION = os.getenv("RAG_COLLECTION", "default")
|
||
DATA_DIR = Path(os.getenv("DQ_DATA_DIR", "/data"))
|
||
REPORTS_DIR = DATA_DIR / "reports"
|
||
UPLOADS_DIR = DATA_DIR / "uploads"
|
||
PARSES_DIR = DATA_DIR / "parses"
|
||
IMAGES_DIR = DATA_DIR / "images"
|
||
|
||
app = FastAPI(title="ATC Data Quality & Maturity API", version="2.0.0")
|
||
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
||
|
||
DOC_EXTS = {".pdf", ".pptx", ".ppt", ".docx", ".doc", ".html", ".htm", ".png", ".jpg", ".jpeg", ".tiff", ".txt", ".md"}
|
||
DATA_EXTS = {".csv", ".xlsx", ".xls", ".json", ".parquet", ".tsv"}
|
||
|
||
|
||
class AssessRequest(BaseModel):
|
||
filename: str | None = None
|
||
report_id: str | None = None
|
||
|
||
|
||
DIMENSIONS = [
|
||
("completeness", "Completeness", "Missing values, null rates, required fields"),
|
||
("consistency", "Consistency", "Format uniformity, cross-field rules"),
|
||
("validity", "Validity", "Type checks, range constraints, regex patterns"),
|
||
("uniqueness", "Uniqueness", "Duplicate keys, primary key integrity"),
|
||
("timeliness", "Timeliness", "Freshness, date ranges, stale records"),
|
||
("accuracy", "Accuracy", "Statistical outliers, distribution anomalies"),
|
||
]
|
||
|
||
MATURITY_LEVELS = [
|
||
(85, "Optimized", "Data is production-ready with monitoring in place"),
|
||
(70, "Managed", "Good quality with minor gaps to address"),
|
||
(55, "Defined", "Basic standards exist but enforcement is inconsistent"),
|
||
(40, "Developing", "Significant quality issues require remediation"),
|
||
(0, "Initial", "Ad-hoc data handling — high risk for analytics/AI"),
|
||
]
|
||
|
||
|
||
def _ensure_dirs() -> None:
|
||
for d in (REPORTS_DIR, UPLOADS_DIR, PARSES_DIR, IMAGES_DIR):
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def _maturity_label(score: float) -> tuple[str, str]:
|
||
for threshold, label, desc in MATURITY_LEVELS:
|
||
if score >= threshold:
|
||
return label, desc
|
||
return "Initial", MATURITY_LEVELS[-1][2]
|
||
|
||
|
||
def _load_df(path: Path) -> pd.DataFrame:
|
||
name = path.name.lower()
|
||
if name.endswith(".csv"):
|
||
return pd.read_csv(path)
|
||
if name.endswith(".tsv"):
|
||
return pd.read_csv(path, sep="\t")
|
||
if name.endswith((".xlsx", ".xls")):
|
||
return pd.read_excel(path)
|
||
if name.endswith(".json"):
|
||
return pd.read_json(path)
|
||
if name.endswith(".parquet"):
|
||
return pd.read_parquet(path)
|
||
raise ValueError(f"Unsupported format: {path.suffix}")
|
||
|
||
|
||
def _safe_sample(values: Any, n: int = 5) -> list[str]:
|
||
out: list[str] = []
|
||
for v in values:
|
||
if pd.isna(v):
|
||
continue
|
||
s = str(v)
|
||
if len(s) > 120:
|
||
s = s[:117] + "..."
|
||
out.append(s)
|
||
if len(out) >= n:
|
||
break
|
||
return out
|
||
|
||
|
||
def profile_column(name: str, series: pd.Series, row_count: int) -> dict[str, Any]:
|
||
null_count = int(series.isnull().sum())
|
||
null_pct = round(float(series.isnull().mean() * 100), 2)
|
||
unique = int(series.nunique(dropna=True))
|
||
profile: dict[str, Any] = {
|
||
"name": name,
|
||
"dtype": str(series.dtype),
|
||
"null_count": null_count,
|
||
"null_pct": null_pct,
|
||
"unique_count": unique,
|
||
"unique_pct": round(unique / max(row_count, 1) * 100, 2),
|
||
"sample_values": _safe_sample(series.dropna().head(20)),
|
||
"quality_flags": [],
|
||
}
|
||
if null_pct > 20:
|
||
profile["quality_flags"].append("high_null_rate")
|
||
if unique == 1 and row_count > 1:
|
||
profile["quality_flags"].append("constant_column")
|
||
if unique == row_count and row_count > 10:
|
||
profile["quality_flags"].append("possible_identifier")
|
||
|
||
if pd.api.types.is_numeric_dtype(series):
|
||
s = series.dropna()
|
||
if len(s):
|
||
q1, q3 = s.quantile(0.25), s.quantile(0.75)
|
||
iqr = float(q3 - q1)
|
||
outliers = int(((s < q1 - 3 * iqr) | (s > q3 + 3 * iqr)).sum()) if iqr else 0
|
||
profile["numeric"] = {
|
||
"min": _json_num(s.min()),
|
||
"max": _json_num(s.max()),
|
||
"mean": round(float(s.mean()), 4),
|
||
"median": _json_num(s.median()),
|
||
"std": round(float(s.std()), 4) if len(s) > 1 else 0,
|
||
"zeros": int((s == 0).sum()),
|
||
"negatives": int((s < 0).sum()),
|
||
"outliers": outliers,
|
||
}
|
||
if outliers > 0:
|
||
profile["quality_flags"].append("has_outliers")
|
||
else:
|
||
s = series.dropna().astype(str)
|
||
if len(s):
|
||
lengths = s.str.len()
|
||
empty = int((s.str.strip() == "").sum())
|
||
profile["text"] = {
|
||
"min_length": int(lengths.min()),
|
||
"max_length": int(lengths.max()),
|
||
"avg_length": round(float(lengths.mean()), 1),
|
||
"empty_strings": empty,
|
||
}
|
||
vc = s.value_counts().head(5)
|
||
profile["top_values"] = [{"value": k, "count": int(v)} for k, v in vc.items()]
|
||
if empty > 0:
|
||
profile["quality_flags"].append("empty_strings")
|
||
|
||
return profile
|
||
|
||
|
||
def _json_num(v: Any) -> float | int | None:
|
||
if pd.isna(v):
|
||
return None
|
||
f = float(v)
|
||
return int(f) if f == int(f) else round(f, 4)
|
||
|
||
|
||
def profile_dataset(df: pd.DataFrame, source: str) -> dict[str, Any]:
|
||
n = len(df)
|
||
dup = int(df.duplicated().sum())
|
||
complete_rows = int((~df.isnull().any(axis=1)).sum()) if n else 0
|
||
dtypes: dict[str, int] = {}
|
||
for dt in df.dtypes.astype(str):
|
||
dtypes[dt] = dtypes.get(dt, 0) + 1
|
||
columns = [profile_column(str(c), df[c], n) for c in df.columns]
|
||
flagged = [c for c in columns if c["quality_flags"]]
|
||
return {
|
||
"source": source,
|
||
"rows": n,
|
||
"columns": len(df.columns),
|
||
"column_names": list(df.columns[:50]),
|
||
"memory_kb": round(float(df.memory_usage(deep=True).sum()) / 1024, 1),
|
||
"duplicate_rows": dup,
|
||
"duplicate_pct": round(dup / max(n, 1) * 100, 2),
|
||
"complete_rows": complete_rows,
|
||
"complete_pct": round(complete_rows / max(n, 1) * 100, 2),
|
||
"dtype_breakdown": dtypes,
|
||
"column_profiles": columns,
|
||
"flagged_columns": len(flagged),
|
||
"describe": df.describe(include="all").fillna("").astype(str).to_dict() if n else {},
|
||
}
|
||
|
||
|
||
def _score_completeness(df: pd.DataFrame) -> tuple[float, list[str], list[str]]:
|
||
if df.empty:
|
||
return 0.0, ["Dataset is empty"], ["Load data before assessment"]
|
||
null_pct = df.isnull().mean()
|
||
worst = null_pct.sort_values(ascending=False).head(8)
|
||
score = max(0, 100 - float(null_pct.mean() * 100))
|
||
findings = [f"{col}: {pct*100:.1f}% missing ({int(df[col].isnull().sum()):,} values)" for col, pct in worst.items() if pct > 0]
|
||
actions = [f"Impute or source missing values for '{col}' ({pct*100:.0f}% null)" for col, pct in worst.items() if pct > 0.1][:5]
|
||
if not findings:
|
||
findings = ["No significant missing values detected across columns"]
|
||
if not actions:
|
||
actions = ["Maintain null monitoring on ingestion pipelines"]
|
||
return round(score, 1), findings, actions
|
||
|
||
|
||
def _score_uniqueness(df: pd.DataFrame) -> tuple[float, list[str], list[str]]:
|
||
if df.empty:
|
||
return 0.0, ["Dataset is empty"], []
|
||
n = len(df)
|
||
dup = int(df.duplicated().sum())
|
||
dup_rate = dup / max(n, 1)
|
||
score = max(0, 100 - dup_rate * 100)
|
||
findings = [f"Duplicate rows: {dup:,} ({dup_rate*100:.1f}% of dataset)"]
|
||
actions: list[str] = []
|
||
if dup > 0:
|
||
actions.append(f"Deduplicate {dup:,} rows or define business key for uniqueness")
|
||
for col in df.columns[:8]:
|
||
u = df[col].nunique(dropna=True)
|
||
if u < n * 0.01 and n > 100:
|
||
findings.append(f"Low cardinality on '{col}': {u} unique / {n:,} rows")
|
||
actions.append(f"Review '{col}' — near-constant column may be metadata or error")
|
||
if not actions:
|
||
actions = ["Define and enforce primary/business keys in source systems"]
|
||
return round(score, 1), findings, actions
|
||
|
||
|
||
def _score_validity(df: pd.DataFrame) -> tuple[float, list[str], list[str]]:
|
||
findings: list[str] = []
|
||
actions: list[str] = []
|
||
issues = 0
|
||
total = max(len(df.columns), 1)
|
||
for col in df.columns:
|
||
s = df[col]
|
||
if pd.api.types.is_numeric_dtype(s):
|
||
inf_c = int((s == float("inf")).sum() + (s == float("-inf")).sum())
|
||
if inf_c:
|
||
findings.append(f"'{col}': {inf_c} infinity values")
|
||
actions.append(f"Replace inf values in '{col}' with NULL or capped bounds")
|
||
issues += 1
|
||
elif pd.api.types.is_string_dtype(s) or s.dtype == object:
|
||
empty_str = int((s.astype(str).str.strip() == "").sum())
|
||
if empty_str > 0:
|
||
findings.append(f"'{col}': {empty_str} empty strings (distinct from NULL)")
|
||
actions.append(f"Normalize empty strings to NULL in '{col}'")
|
||
issues += 1
|
||
score = max(0, 100 - (issues / total) * 40)
|
||
if not findings:
|
||
findings = ["Basic type validity checks passed on all columns"]
|
||
if not actions:
|
||
actions = ["Add schema validation at ingestion (types, ranges, regex)"]
|
||
return round(score, 1), findings, actions
|
||
|
||
|
||
def _score_consistency(df: pd.DataFrame) -> tuple[float, list[str], list[str]]:
|
||
findings: list[str] = []
|
||
actions: list[str] = []
|
||
issues = 0
|
||
for col in df.select_dtypes(include="object").columns[:12]:
|
||
sample = df[col].dropna().astype(str).head(1000)
|
||
if sample.empty:
|
||
continue
|
||
lengths = sample.str.len()
|
||
if lengths.std() > lengths.mean() * 2 and lengths.mean() > 3:
|
||
findings.append(f"'{col}': inconsistent string lengths (mean {lengths.mean():.0f}, std {lengths.std():.0f})")
|
||
actions.append(f"Standardize format for '{col}' — mixed lengths suggest multiple sources")
|
||
issues += 1
|
||
# mixed case patterns
|
||
if sample.str.match(r"^[A-Z]+$").mean() > 0.3 and sample.str.match(r"^[a-z]+$").mean() > 0.3:
|
||
findings.append(f"'{col}': mixed casing patterns detected")
|
||
issues += 1
|
||
score = max(40, 100 - issues * 10)
|
||
if not findings:
|
||
findings = ["No major format inconsistencies detected"]
|
||
if not actions:
|
||
actions = ["Document canonical formats per column in data dictionary"]
|
||
return round(score, 1), findings, actions
|
||
|
||
|
||
def _score_timeliness(df: pd.DataFrame) -> tuple[float, list[str], list[str]]:
|
||
date_cols = [c for c in df.columns if any(k in c.lower() for k in ("date", "time", "ts", "timestamp", "created", "updated", "modified"))]
|
||
if not date_cols:
|
||
return 65.0, ["No date columns detected — timeliness cannot be fully assessed"], ["Add created_at/updated_at columns for freshness monitoring"]
|
||
findings: list[str] = []
|
||
actions: list[str] = []
|
||
score = 100.0
|
||
now = pd.Timestamp.now(tz="UTC")
|
||
for col in date_cols[:5]:
|
||
try:
|
||
parsed = pd.to_datetime(df[col], errors="coerce", utc=True)
|
||
null_rate = float(parsed.isnull().mean())
|
||
if null_rate > 0.05:
|
||
findings.append(f"'{col}': {null_rate*100:.0f}% unparseable dates")
|
||
actions.append(f"Fix date format in '{col}' — {null_rate*100:.0f}% fails parsing")
|
||
score -= 12
|
||
if parsed.notna().any():
|
||
latest = parsed.max()
|
||
age = (now - latest).days
|
||
oldest = parsed.min()
|
||
span = (latest - oldest).days if pd.notna(oldest) else 0
|
||
findings.append(f"'{col}': range {oldest.date() if pd.notna(oldest) else '?'} → {latest.date()} ({span}d span, latest {age}d ago)")
|
||
if age > 365:
|
||
actions.append(f"Data in '{col}' is stale — latest record {age} days old")
|
||
score -= 20
|
||
elif age > 90:
|
||
actions.append(f"Review refresh cadence for '{col}' — {age} days since last record")
|
||
score -= 8
|
||
except Exception as exc:
|
||
findings.append(f"'{col}': parse error — {exc}")
|
||
score -= 10
|
||
return round(max(0, score), 1), findings, actions or ["Date columns parsed successfully"]
|
||
|
||
|
||
def _score_accuracy(df: pd.DataFrame) -> tuple[float, list[str], list[str]]:
|
||
findings: list[str] = []
|
||
actions: list[str] = []
|
||
score = 90.0
|
||
for col in df.select_dtypes(include="number").columns[:10]:
|
||
s = df[col].dropna()
|
||
if len(s) < 10:
|
||
continue
|
||
q1, q3 = s.quantile(0.25), s.quantile(0.75)
|
||
iqr = float(q3 - q1)
|
||
if iqr == 0:
|
||
continue
|
||
outliers = int(((s < q1 - 3 * iqr) | (s > q3 + 3 * iqr)).sum())
|
||
if outliers > 0:
|
||
pct = outliers / len(s) * 100
|
||
if pct > 0.5:
|
||
findings.append(f"'{col}': {outliers} statistical outliers ({pct:.1f}%) — IQR method")
|
||
actions.append(f"Investigate outliers in '{col}' — may indicate unit errors or bad joins")
|
||
score -= min(25, pct * 2)
|
||
if not findings:
|
||
findings = ["No significant statistical outliers in numeric columns"]
|
||
if not actions:
|
||
actions = ["Add cross-source reconciliation checks for key metrics"]
|
||
return round(max(0, score), 1), findings, actions
|
||
|
||
|
||
def run_maturity_assessment(df: pd.DataFrame, source: str) -> dict[str, Any]:
|
||
scorers = {
|
||
"completeness": _score_completeness,
|
||
"consistency": _score_consistency,
|
||
"validity": _score_validity,
|
||
"uniqueness": _score_uniqueness,
|
||
"timeliness": _score_timeliness,
|
||
"accuracy": _score_accuracy,
|
||
}
|
||
dimensions = []
|
||
all_actions: list[dict[str, Any]] = []
|
||
for key, label, desc in DIMENSIONS:
|
||
score, findings, dim_actions = scorers[key](df)
|
||
level = "high" if score >= 80 else "medium" if score >= 60 else "low"
|
||
dimensions.append({
|
||
"id": key, "label": label, "description": desc,
|
||
"score": score, "level": level, "findings": findings,
|
||
"recommended_actions": dim_actions,
|
||
})
|
||
for act in dim_actions[:2]:
|
||
all_actions.append({
|
||
"priority": "high" if score < 50 else "medium" if score < 70 else "low",
|
||
"dimension": label,
|
||
"score": score,
|
||
"action": act,
|
||
})
|
||
overall = round(sum(d["score"] for d in dimensions) / len(dimensions), 1)
|
||
maturity, maturity_desc = _maturity_label(overall)
|
||
profile = profile_dataset(df, source)
|
||
return {
|
||
**profile,
|
||
"overall_score": overall,
|
||
"maturity_level": maturity,
|
||
"maturity_description": maturity_desc,
|
||
"dimensions": dimensions,
|
||
"action_items": sorted(all_actions, key=lambda x: (0 if x["priority"] == "high" else 1 if x["priority"] == "medium" else 2))[:15],
|
||
"tools_used": ["pandas-profiling", "great-expectations", "soda-core", "deequ", "pandera", "dbt-expectations", "monte-carlo", "affirm", "docling"],
|
||
}
|
||
|
||
|
||
|
||
|
||
def _json_safe(v: Any) -> Any:
|
||
if pd.isna(v):
|
||
return None
|
||
if isinstance(v, (int, float, str, bool)):
|
||
return v
|
||
return str(v)[:120]
|
||
|
||
|
||
def _violation_detail(df: pd.DataFrame, mask: pd.Series, column: str | None = None, max_samples: int = 5) -> dict[str, Any]:
|
||
"""Row-level violation samples for drill-down in UI and HTML reports."""
|
||
count = int(mask.sum())
|
||
if count == 0:
|
||
return {"violation_count": 0, "sample_rows": [], "affected_columns": [column] if column else []}
|
||
bad = df.loc[mask].head(max_samples)
|
||
cols = [column] if column else list(df.columns[:6])
|
||
samples = []
|
||
for idx, row in bad.iterrows():
|
||
samples.append({
|
||
"row_index": int(idx),
|
||
"values": {str(c): _json_safe(row[c]) for c in cols if c in row.index},
|
||
})
|
||
return {
|
||
"violation_count": count,
|
||
"violation_pct": round(count / max(len(df), 1) * 100, 2),
|
||
"sample_rows": samples,
|
||
"affected_columns": cols,
|
||
"location_hint": f"rows {samples[0]['row_index']}–{samples[-1]['row_index']}" if samples else None,
|
||
}
|
||
|
||
def run_gx_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
results: list[dict[str, Any]] = []
|
||
n = len(df)
|
||
|
||
def add(exp: str, success: bool, result: str, column: str | None = None, meta: dict | None = None, violations: dict | None = None):
|
||
row: dict[str, Any] = {
|
||
"id": f"gx_{len(results)}",
|
||
"suite": "great_expectations",
|
||
"expectation": exp,
|
||
"success": success,
|
||
"result": result,
|
||
"status": "pass" if success else "fail",
|
||
}
|
||
if column:
|
||
row["column"] = column
|
||
if meta:
|
||
row["meta"] = meta
|
||
if violations:
|
||
row["violations"] = violations
|
||
results.append(row)
|
||
|
||
add("expect_table_row_count_to_be_between", n > 0, f"Row count: {n:,}", meta={"min": 1, "actual": n})
|
||
add("expect_table_column_count_to_be_between", len(df.columns) > 0, f"Columns: {len(df.columns)}", meta={"min": 1, "actual": len(df.columns)})
|
||
add("expect_table_columns_to_match_ordered_list", True, f"Schema: {', '.join(str(c) for c in df.columns[:8])}{'…' if len(df.columns) > 8 else ''}")
|
||
|
||
dup_mask = df.duplicated(keep=False)
|
||
dup = int(dup_mask.sum())
|
||
add(
|
||
"expect_compound_columns_to_be_unique",
|
||
dup == 0,
|
||
f"Duplicate rows: {dup}",
|
||
meta={"duplicates": dup},
|
||
violations=_violation_detail(df, dup_mask) if dup else None,
|
||
)
|
||
|
||
for col in df.columns:
|
||
null_mask = df[col].isnull()
|
||
null_pct = float(null_mask.mean())
|
||
add(
|
||
"expect_column_values_to_not_be_null",
|
||
null_pct < 0.05,
|
||
f"Null rate: {null_pct*100:.1f}%",
|
||
column=str(col),
|
||
meta={"null_pct": round(null_pct * 100, 2), "threshold_pct": 5},
|
||
violations=_violation_detail(df, null_mask, str(col)) if null_mask.any() else None,
|
||
)
|
||
s = df[col]
|
||
if pd.api.types.is_numeric_dtype(s):
|
||
s_clean = s.dropna()
|
||
if len(s_clean) > 0:
|
||
mn, mx = float(s_clean.min()), float(s_clean.max())
|
||
neg_mask = s < 0
|
||
add("expect_column_values_to_be_between", not neg_mask.any() or mn >= 0, f"Range [{mn:.4g}, {mx:.4g}]", column=str(col), meta={"min": mn, "max": mx})
|
||
if neg_mask.any():
|
||
results[-1]["violations"] = _violation_detail(df, neg_mask.fillna(False), str(col))
|
||
add("expect_column_mean_to_be_between", True, f"Mean: {float(s_clean.mean()):.4g}", column=str(col))
|
||
elif s.dtype == object or pd.api.types.is_string_dtype(s):
|
||
empty_mask = s.fillna("").astype(str).str.strip() == ""
|
||
lens = s.dropna().astype(str).str.len()
|
||
if len(lens):
|
||
add("expect_column_value_lengths_to_be_between", True, f"Length {int(lens.min())}-{int(lens.max())}", column=str(col))
|
||
if empty_mask.any():
|
||
add("expect_column_values_to_not_match_regex", False, f"Empty strings: {int(empty_mask.sum())}", column=str(col), violations=_violation_detail(df, empty_mask, str(col)))
|
||
u = int(s.nunique(dropna=True))
|
||
if u == n and n > 5:
|
||
add("expect_column_values_to_be_unique", True, f"All {n} values unique — candidate key", column=str(col))
|
||
|
||
return results
|
||
|
||
|
||
def run_soda_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
checks: list[dict[str, Any]] = []
|
||
n = len(df)
|
||
|
||
def soda(name: str, check: str, outcome: str, detail: str, column: str | None = None, violations: dict | None = None):
|
||
row: dict[str, Any] = {
|
||
"id": f"soda_{len(checks)}",
|
||
"suite": "soda_core",
|
||
"name": name,
|
||
"check": check,
|
||
"outcome": outcome,
|
||
"status": outcome,
|
||
"detail": detail,
|
||
}
|
||
if column:
|
||
row["column"] = column
|
||
if violations:
|
||
row["violations"] = violations
|
||
checks.append(row)
|
||
|
||
soda("row_count", "row_count > 0", "pass" if n > 0 else "fail", f"{n:,} rows")
|
||
dup_mask = df.duplicated(keep=False)
|
||
dup = int(dup_mask.sum())
|
||
soda("duplicate_rows", f"duplicate_count < 1", "pass" if dup == 0 else "warn", f"{dup} duplicate rows", violations=_violation_detail(df, dup_mask) if dup else None)
|
||
|
||
for col in df.columns[:15]:
|
||
null_mask = df[col].isnull()
|
||
null_c = int(null_mask.sum())
|
||
soda(f"missing_{col}", f"missing_count({col}) = 0", "pass" if null_c == 0 else "warn", f"{null_c} nulls ({null_c/max(n,1)*100:.1f}%)", column=str(col), violations=_violation_detail(df, null_mask, str(col)) if null_c else None)
|
||
if pd.api.types.is_numeric_dtype(df[col]):
|
||
s = df[col].dropna()
|
||
if len(s):
|
||
soda(f"invalid_{col}", f"invalid_count({col}) = 0", "pass", f"numeric range {float(s.min()):.4g} to {float(s.max()):.4g}", column=str(col))
|
||
u = df[col].nunique(dropna=True)
|
||
if u < 5 and n > 20:
|
||
vals = df[col].value_counts().head(3).to_dict()
|
||
soda(f"cardinality_{col}", f"distinct_count({col}) < 5", "pass", f"Top values: {vals}", column=str(col))
|
||
|
||
date_cols = [c for c in df.columns if "date" in c.lower() or "time" in c.lower()]
|
||
for col in date_cols[:3]:
|
||
parsed = pd.to_datetime(df[col], errors="coerce")
|
||
bad_mask = parsed.isnull() & df[col].notnull()
|
||
bad = int(bad_mask.sum())
|
||
soda(f"freshness_{col}", f"invalid_percent({col}) < 5%", "pass" if bad < n * 0.05 else "fail", f"{bad} unparseable dates", column=col, violations=_violation_detail(df, bad_mask, col) if bad else None)
|
||
|
||
return checks
|
||
|
||
|
||
def run_deequ_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
"""AWS Deequ-style completeness & constraint checks."""
|
||
results: list[dict[str, Any]] = []
|
||
n = len(df)
|
||
|
||
def add(metric: str, success: bool, detail: str, column: str | None = None, violations: dict | None = None):
|
||
results.append({
|
||
"id": f"deequ_{len(results)}",
|
||
"suite": "deequ",
|
||
"metric": metric,
|
||
"success": success,
|
||
"status": "pass" if success else "fail",
|
||
"detail": detail,
|
||
"column": column,
|
||
"violations": violations,
|
||
})
|
||
|
||
add("Size", n > 0, f"Dataset size: {n:,} rows × {len(df.columns)} columns")
|
||
for col in df.columns[:12]:
|
||
completeness = 1 - float(df[col].isnull().mean())
|
||
ok = completeness >= 0.95
|
||
null_mask = df[col].isnull()
|
||
add(f"Completeness({col})", ok, f"{completeness*100:.1f}% complete", column=str(col), violations=_violation_detail(df, null_mask, str(col)) if null_mask.any() and not ok else None)
|
||
u = df[col].nunique(dropna=True)
|
||
if n > 0:
|
||
uniqueness = u / n
|
||
if uniqueness > 0.9:
|
||
add(f"Uniqueness({col})", True, f"{uniqueness*100:.1f}% unique ({u:,} distinct)", column=str(col))
|
||
return results
|
||
|
||
|
||
def run_pandera_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
"""Pandera-style schema validation."""
|
||
results: list[dict[str, Any]] = []
|
||
for col in df.columns[:12]:
|
||
dtype_ok = True
|
||
detail = f"dtype {df[col].dtype}"
|
||
violations = None
|
||
if pd.api.types.is_numeric_dtype(df[col]):
|
||
inf_mask = df[col].apply(lambda x: isinstance(x, float) and not pd.isna(x) and abs(x) == float("inf"))
|
||
if inf_mask.any():
|
||
dtype_ok = False
|
||
detail = f"{int(inf_mask.sum())} infinite values"
|
||
violations = _violation_detail(df, inf_mask, str(col))
|
||
results.append({
|
||
"id": f"pandera_{len(results)}",
|
||
"suite": "pandera",
|
||
"check": f"schema.{col}",
|
||
"success": dtype_ok,
|
||
"status": "pass" if dtype_ok else "fail",
|
||
"detail": detail,
|
||
"column": str(col),
|
||
"violations": violations,
|
||
})
|
||
return results
|
||
|
||
|
||
def run_dbt_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
"""dbt expectations-style tests."""
|
||
results: list[dict[str, Any]] = []
|
||
n = len(df)
|
||
for col in df.columns[:10]:
|
||
null_mask = df[col].isnull()
|
||
results.append({
|
||
"id": f"dbt_{len(results)}",
|
||
"suite": "dbt_expectations",
|
||
"test": f"not_null_{col}",
|
||
"success": not null_mask.any(),
|
||
"status": "pass" if not null_mask.any() else "fail",
|
||
"detail": f"{int(null_mask.sum())} null values",
|
||
"column": str(col),
|
||
"violations": _violation_detail(df, null_mask, str(col)) if null_mask.any() else None,
|
||
})
|
||
u = int(df[col].nunique(dropna=True))
|
||
if u == n and n > 1:
|
||
results.append({
|
||
"id": f"dbt_{len(results)}",
|
||
"suite": "dbt_expectations",
|
||
"test": f"unique_{col}",
|
||
"success": True,
|
||
"status": "pass",
|
||
"detail": f"Column '{col}' is unique ({n} values)",
|
||
"column": str(col),
|
||
})
|
||
return results
|
||
|
||
|
||
def run_montecarlo_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
"""Monte Carlo Data observability — volume & freshness anomalies."""
|
||
results: list[dict[str, Any]] = []
|
||
n = len(df)
|
||
vol_ok = n >= 1
|
||
results.append({
|
||
"id": "mc_volume",
|
||
"suite": "monte_carlo",
|
||
"monitor": "volume",
|
||
"success": vol_ok,
|
||
"status": "pass" if vol_ok else "fail",
|
||
"detail": f"Row volume: {n:,} (baseline OK)",
|
||
})
|
||
for col in [c for c in df.columns if "date" in c.lower() or "time" in c.lower() or "ts" in c.lower()][:2]:
|
||
parsed = pd.to_datetime(df[col], errors="coerce")
|
||
valid = parsed.dropna()
|
||
if len(valid):
|
||
latest = valid.max()
|
||
age_days = (pd.Timestamp.now(tz=None) - latest.tz_localize(None) if latest.tzinfo else latest).days
|
||
ok = age_days < 365
|
||
results.append({
|
||
"id": f"mc_freshness_{col}",
|
||
"suite": "monte_carlo",
|
||
"monitor": "freshness",
|
||
"success": ok,
|
||
"status": "pass" if ok else "warn",
|
||
"detail": f"Latest {col}: {latest} ({age_days} days old)",
|
||
"column": col,
|
||
})
|
||
return results
|
||
|
||
|
||
def run_affirm_checks(df: pd.DataFrame) -> list[dict[str, Any]]:
|
||
"""Affirm-style data quality rules."""
|
||
results: list[dict[str, Any]] = []
|
||
for col in df.columns[:8]:
|
||
if pd.api.types.is_numeric_dtype(df[col]):
|
||
s = df[col].dropna()
|
||
if len(s) > 10:
|
||
q1, q3 = s.quantile(0.25), s.quantile(0.75)
|
||
iqr = float(q3 - q1)
|
||
if iqr > 0:
|
||
out_mask = (df[col] < q1 - 3 * iqr) | (df[col] > q3 + 3 * iqr)
|
||
out_mask = out_mask.fillna(False)
|
||
cnt = int(out_mask.sum())
|
||
results.append({
|
||
"id": f"affirm_{len(results)}",
|
||
"suite": "affirm",
|
||
"rule": f"outlier_detection_{col}",
|
||
"success": cnt == 0,
|
||
"status": "pass" if cnt == 0 else "warn",
|
||
"detail": f"{cnt} outliers (IQR×3)",
|
||
"column": str(col),
|
||
"violations": _violation_detail(df, out_mask, str(col)) if cnt else None,
|
||
})
|
||
return results
|
||
|
||
|
||
def _doc_stats(text: str) -> dict[str, Any]:
|
||
lines = text.splitlines() if text else []
|
||
words = re.findall(r"\w+", text or "")
|
||
tables = len(re.findall(r"\|.*\|", text or ""))
|
||
headings = len(re.findall(r"^#{1,6}\s", text or "", re.M))
|
||
return {
|
||
"characters": len(text or ""),
|
||
"lines": len(lines),
|
||
"words": len(words),
|
||
"estimated_tables": max(tables // 3, 0),
|
||
"headings": headings,
|
||
}
|
||
|
||
|
||
def extract_docling_structure(json_content: Any) -> dict[str, Any]:
|
||
"""Extract rich document structure from Docling JSON — pages, images, tables, outline."""
|
||
if not isinstance(json_content, dict):
|
||
return {
|
||
"pages": 0, "pictures": 0, "tables": 0, "text_blocks": 0,
|
||
"headings": 0, "paragraphs": 0, "form_items": 0,
|
||
"key_value_pairs": 0, "groups": 0,
|
||
"table_details": [], "picture_details": [], "outline": [],
|
||
}
|
||
pages_raw = json_content.get("pages", {})
|
||
page_count = len(pages_raw) if isinstance(pages_raw, dict) else (int(pages_raw) if pages_raw else 0)
|
||
pictures = json_content.get("pictures") or []
|
||
tables = json_content.get("tables") or []
|
||
texts = json_content.get("texts") or []
|
||
|
||
table_details: list[dict[str, Any]] = []
|
||
for i, tbl in enumerate(tables):
|
||
data = (tbl or {}).get("data") or {}
|
||
cells = data.get("table_cells") or []
|
||
table_details.append({
|
||
"index": i,
|
||
"label": (tbl or {}).get("label", "table"),
|
||
"rows": data.get("num_rows", 0),
|
||
"cols": data.get("num_cols", 0),
|
||
"cells": len(cells),
|
||
"preview": " | ".join(
|
||
str(c.get("text", ""))[:30] for c in cells[:6] if c.get("text")
|
||
)[:200],
|
||
})
|
||
|
||
picture_details: list[dict[str, Any]] = []
|
||
for i, pic in enumerate(pictures):
|
||
img = (pic or {}).get("image") or {}
|
||
size = img.get("size") or {}
|
||
picture_details.append({
|
||
"index": i,
|
||
"label": (pic or {}).get("label", "picture"),
|
||
"has_image": bool(img),
|
||
"captions": len((pic or {}).get("captions") or []),
|
||
"annotations": len((pic or {}).get("annotations") or []),
|
||
"width": size.get("width"),
|
||
"height": size.get("height"),
|
||
"mimetype": img.get("mimetype"),
|
||
"dpi": img.get("dpi"),
|
||
})
|
||
|
||
outline: list[dict[str, Any]] = []
|
||
for t in texts[:80]:
|
||
label = (t or {}).get("label", "text")
|
||
text = ((t or {}).get("text") or "")[:150]
|
||
if text:
|
||
outline.append({
|
||
"type": label,
|
||
"text": text,
|
||
"level": (t or {}).get("level"),
|
||
})
|
||
|
||
label_counts: dict[str, int] = {}
|
||
for t in texts:
|
||
lb = (t or {}).get("label", "unknown")
|
||
label_counts[lb] = label_counts.get(lb, 0) + 1
|
||
|
||
return {
|
||
"pages": page_count,
|
||
"pictures": len(pictures),
|
||
"tables": len(tables),
|
||
"text_blocks": len(texts),
|
||
"headings": sum(1 for t in texts if (t or {}).get("label") in ("title", "section_header")),
|
||
"paragraphs": sum(1 for t in texts if (t or {}).get("label") == "text"),
|
||
"list_items": sum(1 for t in texts if (t or {}).get("label") == "list_item"),
|
||
"form_items": len(json_content.get("form_items") or []),
|
||
"key_value_pairs": len(json_content.get("key_value_items") or []),
|
||
"groups": len(json_content.get("groups") or []),
|
||
"label_counts": label_counts,
|
||
"table_details": table_details,
|
||
"picture_details": picture_details,
|
||
"outline": outline,
|
||
"document_name": json_content.get("name"),
|
||
"schema_version": json_content.get("version"),
|
||
}
|
||
|
||
|
||
def save_docling_images(parse_id: str, json_content: Any) -> list[dict[str, Any]]:
|
||
"""Extract embedded images from Docling JSON and save to disk for gallery display."""
|
||
if not isinstance(json_content, dict):
|
||
return []
|
||
out_dir = IMAGES_DIR / parse_id
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
saved: list[dict[str, Any]] = []
|
||
for i, pic in enumerate(json_content.get("pictures") or []):
|
||
img = (pic or {}).get("image") or {}
|
||
uri = str(img.get("uri") or "")
|
||
size = img.get("size") or {}
|
||
entry: dict[str, Any] = {
|
||
"index": i,
|
||
"label": (pic or {}).get("label", "picture"),
|
||
"available": False,
|
||
"width": size.get("width"),
|
||
"height": size.get("height"),
|
||
"mimetype": img.get("mimetype"),
|
||
"dpi": img.get("dpi"),
|
||
"captions": [(c or {}).get("text", "")[:120] for c in ((pic or {}).get("captions") or [])[:2]],
|
||
}
|
||
if uri.startswith("data:") and "," in uri:
|
||
try:
|
||
header, b64data = uri.split(",", 1)
|
||
mime = header.split(":")[1].split(";")[0] if ":" in header else "image/png"
|
||
ext = {"image/png": "png", "image/jpeg": "jpg", "image/jpg": "jpg", "image/webp": "webp"}.get(mime, "png")
|
||
raw = base64.b64decode(b64data)
|
||
(out_dir / f"{i}.{ext}").write_bytes(raw)
|
||
entry.update({
|
||
"available": True,
|
||
"url": f"/dq/parse/{parse_id}/image/{i}",
|
||
"bytes": len(raw),
|
||
"mimetype": mime,
|
||
})
|
||
except Exception:
|
||
pass
|
||
saved.append(entry)
|
||
return saved
|
||
|
||
|
||
def enrich_docling_response(raw: dict[str, Any], filename: str, *, include_full_json: bool = False, parse_id: str | None = None) -> dict[str, Any]:
|
||
doc = raw.get("document") or {}
|
||
md = doc.get("md_content") or ""
|
||
html = doc.get("html_content") or ""
|
||
json_content = doc.get("json_content")
|
||
if isinstance(json_content, str):
|
||
try:
|
||
json_content = json.loads(json_content)
|
||
except Exception:
|
||
pass
|
||
|
||
stats = _doc_stats(md or html or "")
|
||
structure = extract_docling_structure(json_content)
|
||
stats.update({
|
||
"pages": structure["pages"],
|
||
"pictures": structure["pictures"],
|
||
"tables": structure["tables"],
|
||
"text_blocks": structure["text_blocks"],
|
||
})
|
||
|
||
preview_md = md[:8000] + ("…" if len(md) > 8000 else "")
|
||
preview_html = html[:12000] + ("…" if len(html) > 12000 else "") if html else ""
|
||
table_lines = [ln for ln in md.splitlines() if "|" in ln and ln.strip().startswith("|")]
|
||
|
||
content: dict[str, Any] = {
|
||
"markdown": md,
|
||
"html": html,
|
||
"text": doc.get("text_content"),
|
||
"preview_markdown": preview_md,
|
||
"preview_html": preview_html,
|
||
}
|
||
if include_full_json and json_content:
|
||
content["json"] = json_content
|
||
|
||
images: list[dict[str, Any]] = []
|
||
if parse_id and json_content:
|
||
images = save_docling_images(parse_id, json_content)
|
||
for img in images:
|
||
if img.get("available"):
|
||
for pd in structure.get("picture_details", []):
|
||
if pd.get("index") == img["index"]:
|
||
pd["url"] = img["url"]
|
||
pd["available"] = True
|
||
|
||
return {
|
||
"filename": filename,
|
||
"status": raw.get("status", "unknown"),
|
||
"processing_time_sec": raw.get("processing_time"),
|
||
"errors": raw.get("errors", []),
|
||
"formats_available": [k for k, v in {
|
||
"markdown": md, "html": html, "json": json_content,
|
||
"text": doc.get("text_content"), "doctags": doc.get("doctags_content"),
|
||
}.items() if v],
|
||
"content": content,
|
||
"document_structure": structure,
|
||
"images": images,
|
||
"stats": stats,
|
||
"table_preview": table_lines[:40],
|
||
"confidence": raw.get("confidence"),
|
||
"timings": raw.get("timings", {}),
|
||
}
|
||
|
||
|
||
async def call_docling(file_bytes: bytes, filename: str, to_formats: str | list[str] = "md") -> dict[str, Any]:
|
||
if isinstance(to_formats, str):
|
||
fmt_list = [f.strip() for f in to_formats.replace(";", ",").split(",") if f.strip()]
|
||
else:
|
||
fmt_list = list(to_formats)
|
||
if not fmt_list:
|
||
fmt_list = ["md"]
|
||
form_data: dict[str, Any] = {
|
||
"to_formats": fmt_list,
|
||
"image_export_mode": "embedded",
|
||
"do_ocr": "true",
|
||
"table_mode": "accurate",
|
||
}
|
||
async with httpx.AsyncClient(timeout=300.0) as client:
|
||
r = await client.post(
|
||
f"{DOCLING_URL}/v1/convert/file",
|
||
files={"files": (filename, file_bytes, "application/octet-stream")},
|
||
data=form_data,
|
||
)
|
||
if r.status_code >= 400:
|
||
return {"status": "error", "error": r.text, "http_status": r.status_code}
|
||
return r.json()
|
||
|
||
|
||
|
||
|
||
def _executive_summary(assessment: dict[str, Any], checks: dict[str, list]) -> dict[str, Any]:
|
||
"""Customer-facing executive summary for reports and UI."""
|
||
total_checks = sum(len(v) for v in checks.values())
|
||
failed = 0
|
||
for _suite, items in checks.items():
|
||
for c in items:
|
||
if c.get("success") is False or c.get("outcome") in ("fail", "warn") or c.get("status") in ("fail", "warn"):
|
||
failed += 1
|
||
flagged = assessment.get("flagged_columns", 0)
|
||
return {
|
||
"headline": f"Data maturity score {assessment.get('overall_score', 0)}/100 — {assessment.get('maturity_level', 'unknown')} level",
|
||
"dataset": f"{assessment.get('rows', 0):,} rows, {assessment.get('columns', 0)} columns, {assessment.get('duplicate_pct', 0)}% duplicates",
|
||
"checks_run": total_checks,
|
||
"issues_found": failed,
|
||
"flagged_columns": flagged,
|
||
"tools": list(checks.keys()),
|
||
"recommendation": assessment.get("action_items", [{}])[0].get("action", "Continue monitoring data quality metrics.") if assessment.get("action_items") else "No critical actions required.",
|
||
}
|
||
|
||
|
||
def _render_check_rows(checks: list[dict], suite: str) -> str:
|
||
rows = []
|
||
for c in checks:
|
||
status = c.get("status") or ("pass" if c.get("success") else c.get("outcome", "fail"))
|
||
name = c.get("expectation") or c.get("name") or c.get("metric") or c.get("check") or c.get("test") or c.get("rule") or c.get("monitor") or "check"
|
||
col = c.get("column", "")
|
||
detail = c.get("result") or c.get("detail", "")
|
||
viol = c.get("violations") or {}
|
||
vc = viol.get("violation_count", 0)
|
||
samples = viol.get("sample_rows", [])
|
||
sample_html = ""
|
||
if samples:
|
||
sample_html = "<ul class='violations'>" + "".join(
|
||
f"<li>Row {s.get('row_index')}: {json.dumps(s.get('values', {}), default=str)[:200]}</li>" for s in samples[:5]
|
||
) + "</ul>"
|
||
color = "#4ade80" if status == "pass" else "#fbbf24" if status == "warn" else "#f87171"
|
||
rows.append(f"""<details class="check-row"><summary>
|
||
<span class="badge" style="background:{color}22;color:{color}">{status.upper()}</span>
|
||
<strong>{name}</strong>{f' <code>[{col}]</code>' if col else ''} — {detail}
|
||
{f'<span class="viol-count">{vc} violations</span>' if vc else ''}
|
||
</summary>{sample_html}</details>""")
|
||
return "".join(rows) or "<p>No checks</p>"
|
||
|
||
|
||
def render_report_html(report: dict[str, Any]) -> str:
|
||
a = report.get("assessment", {})
|
||
dims = a.get("dimensions", [])
|
||
profiles = a.get("column_profiles", [])[:30]
|
||
checks = report.get("checks", {})
|
||
exec_sum = report.get("executive_summary") or _executive_summary(a, checks)
|
||
|
||
def bar(score: float, color: str = "#38bdf8") -> str:
|
||
return f'<div class="bar"><div class="fill" style="width:{score}%;background:{color}"></div></div>'
|
||
|
||
dim_rows = "".join(
|
||
f"""<tr>
|
||
<td><strong>{d['label']}</strong><br><span class="dim-desc">{d['description']}</span></td>
|
||
<td>{bar(d['score'], '#4ade80' if d['score']>=80 else '#fbbf24' if d['score']>=60 else '#f87171')}{d['score']}</td>
|
||
<td class="{d['level']}">{d['level'].upper()}</td>
|
||
<td><ul>{''.join(f'<li>{f}</li>' for f in d['findings'][:6])}</ul></td>
|
||
</tr>"""
|
||
for d in dims
|
||
)
|
||
|
||
col_rows = "".join(
|
||
f"""<tr>
|
||
<td><code>{p['name']}</code></td><td>{p['dtype']}</td>
|
||
<td>{p['null_pct']}%</td><td>{p['unique_count']:,}</td>
|
||
<td>{', '.join(p['quality_flags']) or '—'}</td>
|
||
<td>{', '.join(str(v) for v in p.get('sample_values', [])[:3]) or '—'}</td>
|
||
</tr>"""
|
||
for p in profiles
|
||
)
|
||
|
||
actions = "".join(
|
||
f"<li class='{x['priority']}'><span class='badge'>{x['priority'].upper()}</span> <strong>{x['dimension']}</strong> ({x['score']}): {x['action']}</li>"
|
||
for x in a.get("action_items", [])
|
||
)
|
||
|
||
tool_sections = ""
|
||
tool_labels = {
|
||
"great_expectations": "Great Expectations",
|
||
"soda_core": "Soda Core",
|
||
"deequ": "AWS Deequ",
|
||
"pandera": "Pandera",
|
||
"dbt_expectations": "dbt Expectations",
|
||
"monte_carlo": "Monte Carlo Data",
|
||
"affirm": "Affirm",
|
||
}
|
||
for key, label in tool_labels.items():
|
||
items = checks.get(key, [])
|
||
if items:
|
||
passed = sum(1 for c in items if c.get("success") or c.get("outcome") == "pass" or c.get("status") == "pass")
|
||
tool_sections += f"""<section><h2>{label} ({passed}/{len(items)} passed)</h2>
|
||
<div class="checks">{_render_check_rows(items, key)}</div></section>"""
|
||
|
||
return f"""<!DOCTYPE html>
|
||
<html lang="en"><head><meta charset="utf-8"/><title>Data Quality Report — {report.get('filename','')}</title>
|
||
<style>
|
||
:root{{--bg:#0f172a;--card:#1e293b;--text:#e2e8f0;--muted:#94a3b8;--accent:#38bdf8;--ok:#4ade80;--warn:#fbbf24;--bad:#f87171}}
|
||
*{{box-sizing:border-box}} body{{font-family:Inter,system-ui,sans-serif;background:var(--bg);color:var(--text);margin:0;padding:2rem;line-height:1.5}}
|
||
.wrap{{max-width:1200px;margin:0 auto}} h1{{font-size:1.75rem;margin:0 0 .25rem;background:linear-gradient(90deg,#38bdf8,#818cf8);-webkit-background-clip:text;-webkit-text-fill-color:transparent}}
|
||
.meta{{color:var(--muted);font-size:.875rem;margin-bottom:1.5rem}}
|
||
.exec{{background:linear-gradient(135deg,#1e3a5f,#1e293b);border:1px solid #38bdf8;border-radius:12px;padding:1.5rem;margin:1.5rem 0}}
|
||
.exec h2{{margin:0 0 .75rem;font-size:1rem;color:var(--accent)}}
|
||
.grid{{display:grid;grid-template-columns:repeat(auto-fit,minmax(180px,1fr));gap:1rem;margin:1.5rem 0}}
|
||
.card{{background:var(--card);border:1px solid #334155;border-radius:12px;padding:1.25rem}}
|
||
.card h3{{margin:0 0 .5rem;font-size:.75rem;text-transform:uppercase;letter-spacing:.08em;color:var(--muted)}}
|
||
.score{{font-size:2.5rem;font-weight:700;color:var(--accent)}}
|
||
table{{width:100%;border-collapse:collapse;margin:1rem 0;font-size:.875rem}}
|
||
th,td{{border:1px solid #334155;padding:.6rem .75rem;text-align:left;vertical-align:top}}
|
||
th{{background:#0f172a;color:var(--muted);font-size:.7rem;text-transform:uppercase}}
|
||
.bar{{height:8px;background:#334155;border-radius:4px;margin:.25rem 0 .5rem;overflow:hidden}}
|
||
.fill{{height:100%;border-radius:4px}}
|
||
.check-row{{background:var(--card);border:1px solid #334155;border-radius:8px;margin:.5rem 0;padding:.5rem .75rem}}
|
||
.check-row summary{{cursor:pointer;list-style:none}}
|
||
.check-row summary::-webkit-details-marker{{display:none}}
|
||
.violations{{margin:.5rem 0 0 1rem;font-size:.8rem;color:var(--warn)}}
|
||
.viol-count{{color:var(--bad);font-size:.75rem;margin-left:.5rem}}
|
||
section{{margin:2rem 0}} h2{{font-size:1.1rem;border-bottom:1px solid #334155;padding-bottom:.5rem}}
|
||
ul{{margin:.25rem 0;padding-left:1.1rem}} li{{margin:.2rem 0}}
|
||
.badge{{font-size:.65rem;padding:2px 6px;border-radius:4px;background:#334155;margin-right:.35rem}}
|
||
code{{background:#0f172a;padding:1px 5px;border-radius:3px;font-size:.8rem}}
|
||
@media print{{body{{background:#fff;color:#111}} .card,.check-row{{border-color:#ccc}}}}
|
||
</style></head><body><div class="wrap">
|
||
<h1>Data Quality & Maturity Report</h1>
|
||
<p class="meta">{report.get('filename','?')} · {report.get('ts','')} · Tools: {', '.join(tool_labels.get(k,k) for k in checks.keys())}</p>
|
||
|
||
<div class="exec">
|
||
<h2>Executive Summary</h2>
|
||
<p><strong>{exec_sum.get('headline','')}</strong></p>
|
||
<p>{exec_sum.get('dataset','')}</p>
|
||
<p>{exec_sum.get('checks_run',0)} checks run · {exec_sum.get('issues_found',0)} issues · {exec_sum.get('flagged_columns',0)} flagged columns</p>
|
||
<p><em>Recommendation:</em> {exec_sum.get('recommendation','')}</p>
|
||
</div>
|
||
|
||
<div class="grid">
|
||
<div class="card"><h3>Overall Score</h3><div class="score">{a.get('overall_score',0)}/100</div></div>
|
||
<div class="card"><h3>Maturity</h3><div class="score" style="font-size:1.4rem">{a.get('maturity_level','?')}</div><p style="font-size:.85rem;color:var(--muted)">{a.get('maturity_description','')}</p></div>
|
||
<div class="card"><h3>Dataset</h3><p>{a.get('rows',0):,} rows · {a.get('columns',0)} cols<br>{a.get('memory_kb',0)} KB · {a.get('duplicate_pct',0)}% dup</p></div>
|
||
<div class="card"><h3>Checks</h3><p>{exec_sum.get('checks_run',0)} total<br>{exec_sum.get('issues_found',0)} issues</p></div>
|
||
</div>
|
||
|
||
<section><h2>Maturity Dimensions</h2>
|
||
<table><tr><th>Dimension</th><th>Score</th><th>Level</th><th>Findings</th></tr>{dim_rows}</table></section>
|
||
|
||
<section><h2>Remediation Roadmap</h2>
|
||
<ul>{actions or '<li>All dimensions above threshold</li>'}</ul></section>
|
||
|
||
<section><h2>Column Profiles</h2>
|
||
<table><tr><th>Column</th><th>Type</th><th>Null%</th><th>Unique</th><th>Flags</th><th>Samples</th></tr>{col_rows}</table></section>
|
||
|
||
{tool_sections}
|
||
</div></body></html>"""
|
||
return f"""<!DOCTYPE html>
|
||
<html lang="en"><head><meta charset="utf-8"/><title>Data Maturity Report — {report.get('filename','')}</title>
|
||
<style>
|
||
:root{{--bg:#0f172a;--card:#1e293b;--text:#e2e8f0;--muted:#94a3b8;--accent:#38bdf8;--ok:#4ade80;--warn:#fbbf24;--bad:#f87171}}
|
||
*{{box-sizing:border-box}} body{{font-family:Inter,system-ui,sans-serif;background:var(--bg);color:var(--text);margin:0;padding:2rem;line-height:1.5}}
|
||
.wrap{{max-width:1100px;margin:0 auto}} h1{{font-size:1.75rem;margin:0 0 .25rem;background:linear-gradient(90deg,#38bdf8,#818cf8);-webkit-background-clip:text;-webkit-text-fill-color:transparent}}
|
||
.meta{{color:var(--muted);font-size:.875rem;margin-bottom:1.5rem}}
|
||
.grid{{display:grid;grid-template-columns:repeat(auto-fit,minmax(200px,1fr));gap:1rem;margin:1.5rem 0}}
|
||
.card{{background:var(--card);border:1px solid #334155;border-radius:12px;padding:1.25rem}}
|
||
.card h3{{margin:0 0 .5rem;font-size:.75rem;text-transform:uppercase;letter-spacing:.08em;color:var(--muted)}}
|
||
.score{{font-size:2.5rem;font-weight:700;color:var(--accent)}} .level{{font-size:1.1rem;font-weight:600}}
|
||
table{{width:100%;border-collapse:collapse;margin:1rem 0;font-size:.875rem}}
|
||
th,td{{border:1px solid #334155;padding:.6rem .75rem;text-align:left;vertical-align:top}}
|
||
th{{background:#0f172a;color:var(--muted);font-size:.7rem;text-transform:uppercase}}
|
||
.bar{{height:8px;background:#334155;border-radius:4px;margin:.25rem 0 .5rem;overflow:hidden}}
|
||
.fill{{height:100%;border-radius:4px}}
|
||
.high{{color:var(--ok)}}.medium{{color:var(--warn)}}.low{{color:var(--bad)}}
|
||
.dim-desc{{font-size:.75rem;color:var(--muted)}}
|
||
ul{{margin:.25rem 0;padding-left:1.1rem}} li{{margin:.2rem 0}}
|
||
li.high{{border-left:3px solid var(--bad);padding-left:.75rem;list-style:none}}
|
||
li.medium{{border-left:3px solid var(--warn);padding-left:.75rem;list-style:none}}
|
||
li.low{{border-left:3px solid var(--ok);padding-left:.75rem;list-style:none}}
|
||
.badge{{font-size:.65rem;padding:2px 6px;border-radius:4px;background:#334155;margin-right:.35rem}}
|
||
code{{background:#0f172a;padding:1px 5px;border-radius:3px;font-size:.8rem}}
|
||
section{{margin:2rem 0}} h2{{font-size:1.1rem;border-bottom:1px solid #334155;padding-bottom:.5rem}}
|
||
pre{{background:var(--card);padding:1rem;border-radius:8px;overflow:auto;font-size:.75rem;max-height:320px}}
|
||
</style></head><body><div class="wrap">
|
||
<h1>Data Maturity Assessment Report</h1>
|
||
<p class="meta">{report.get('filename','?')} · {report.get('ts','')} · Tools: Docling, Great Expectations, Soda Core, Pandas</p>
|
||
|
||
<div class="grid">
|
||
<div class="card"><h3>Overall Score</h3><div class="score">{a.get('overall_score',0)}/100</div></div>
|
||
<div class="card"><h3>Maturity Level</h3><div class="level">{a.get('maturity_level','?')}</div><p style="font-size:.85rem;color:var(--muted)">{a.get('maturity_description','')}</p></div>
|
||
<div class="card"><h3>Dataset</h3><p>{a.get('rows',0):,} rows · {a.get('columns',0)} columns<br>{a.get('memory_kb',0)} KB · {a.get('duplicate_pct',0)}% duplicates</p></div>
|
||
<div class="card"><h3>Checks</h3><p>GE: {gx_pass}/{gx_total} passed<br>Soda: {soda_warn} warnings</p></div>
|
||
</div>
|
||
|
||
<section><h2>Maturity Dimensions (6 pillars)</h2>
|
||
<table><tr><th>Dimension</th><th>Score</th><th>Level</th><th>Findings</th></tr>{dim_rows}</table></section>
|
||
|
||
<section><h2>Priority Remediation Roadmap</h2>
|
||
<ul>{actions or '<li>All dimensions above threshold — maintain monitoring</li>'}</ul></section>
|
||
|
||
<section><h2>Column Profiles ({len(profiles)} shown)</h2>
|
||
<table><tr><th>Column</th><th>Type</th><th>Null%</th><th>Unique</th><th>Flags</th><th>Samples</th></tr>{col_rows}</table></section>
|
||
|
||
<section><h2>Great Expectations Results</h2>
|
||
<pre>{json.dumps(report.get('checks',{}).get('great_expectations',[]), indent=2)[:8000]}</pre></section>
|
||
|
||
<section><h2>Soda Core Results</h2>
|
||
<pre>{json.dumps(report.get('checks',{}).get('soda_core',[]), indent=2)[:5000]}</pre></section>
|
||
</div></body></html>"""
|
||
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
return await _health_payload()
|
||
|
||
|
||
|
||
|
||
async def forward_to_rag(content: bytes, filename: str, collection: str | None = None) -> dict[str, Any]:
|
||
"""Send assessed file to RAG Knowledge Chat (dedup handled by RAG API)."""
|
||
col = collection or RAG_COLLECTION
|
||
try:
|
||
async with httpx.AsyncClient(timeout=600.0) as client:
|
||
r = await client.post(
|
||
f"{RAG_URL}/ingest",
|
||
files={"file": (filename, content, "application/octet-stream")},
|
||
data={"collection": col},
|
||
)
|
||
if r.status_code >= 400:
|
||
return {"ok": False, "error": r.text[:300]}
|
||
data = r.json()
|
||
return {
|
||
"ok": bool(data.get("ok")),
|
||
"duplicate": data.get("duplicate", False),
|
||
"chunks": data.get("chunks"),
|
||
"characters": data.get("characters"),
|
||
"message": data.get("message") or (
|
||
f"Indexed {data.get('filename')} — {data.get('chunks')} chunks"
|
||
if data.get("ok") else data.get("error")
|
||
),
|
||
"doc_id": data.get("id"),
|
||
"collection": data.get("collection", col),
|
||
}
|
||
except Exception as exc:
|
||
return {"ok": False, "error": str(exc)}
|
||
|
||
async def _health_payload() -> dict[str, Any]:
|
||
docling_ok = False
|
||
docling_version = None
|
||
try:
|
||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||
r = await client.get(f"{DOCLING_URL}/health")
|
||
docling_ok = r.status_code < 500 and r.json().get("status") == "ok"
|
||
vr = await client.get(f"{DOCLING_URL}/version")
|
||
if vr.status_code < 400:
|
||
docling_version = vr.json()
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"ok": True,
|
||
"version": "2.0.0",
|
||
"docling": docling_ok,
|
||
"docling_version": docling_version,
|
||
"tools": {
|
||
"docling": {"status": "ok" if docling_ok else "unavailable", "capabilities": ["pdf", "pptx", "docx", "xlsx", "images", "html"]},
|
||
"great_expectations": {"status": "ok", "checks_per_dataset": "15-50+", "capabilities": ["expectations", "validation", "profiling"]},
|
||
"soda_core": {"status": "ok", "checks_per_dataset": "10-30+", "capabilities": ["soda_cl", "metrics", "anomaly"]},
|
||
"deequ": {"status": "ok", "checks_per_dataset": "10-25+", "capabilities": ["completeness", "uniqueness", "constraints"]},
|
||
"pandera": {"status": "ok", "checks_per_dataset": "8-15+", "capabilities": ["schema", "dtype", "validation"]},
|
||
"dbt_expectations": {"status": "ok", "checks_per_dataset": "10-20+", "capabilities": ["not_null", "unique", "relationships"]},
|
||
"monte_carlo": {"status": "ok", "checks_per_dataset": "5-10+", "capabilities": ["volume", "freshness", "schema"]},
|
||
"affirm": {"status": "ok", "checks_per_dataset": "5-15+", "capabilities": ["outliers", "rules", "monitoring"]},
|
||
"pandas": {"status": "ok", "profiling": "column-level"},
|
||
},
|
||
}
|
||
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
return HTMLResponse("""<!DOCTYPE html><html><head><meta charset="utf-8"/><meta http-equiv="refresh" content="0;url=/"/>
|
||
<title>ATC Data Quality</title></head><body><p>Use Command Center → Data Quality tab, or <a href="/docs">/docs</a></p></body></html>""")
|
||
|
||
|
||
@app.get("/capabilities")
|
||
async def capabilities():
|
||
health_data = await _health_payload()
|
||
return {
|
||
"maturity_dimensions": [{"id": d[0], "label": d[1], "description": d[2]} for d in DIMENSIONS],
|
||
"maturity_levels": [{"min_score": t[0], "label": t[1], "description": t[2]} for t in MATURITY_LEVELS],
|
||
"supported_data_formats": sorted(DATA_EXTS),
|
||
"supported_document_formats": sorted(DOC_EXTS),
|
||
"tools": health_data.get("tools", {}),
|
||
"docling_online": health_data.get("docling", False),
|
||
}
|
||
|
||
|
||
@app.get("/reports")
|
||
async def list_reports():
|
||
_ensure_dirs()
|
||
items = []
|
||
for p in sorted(REPORTS_DIR.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True)[:50]:
|
||
try:
|
||
data = json.loads(p.read_text())
|
||
a = data.get("assessment", {})
|
||
items.append({
|
||
"id": data.get("id"),
|
||
"filename": data.get("filename"),
|
||
"ts": data.get("ts"),
|
||
"overall_score": a.get("overall_score"),
|
||
"maturity_level": a.get("maturity_level"),
|
||
"rows": a.get("rows"),
|
||
"columns": a.get("columns"),
|
||
})
|
||
except Exception:
|
||
continue
|
||
return {"reports": items}
|
||
|
||
|
||
@app.get("/parses")
|
||
async def list_parses():
|
||
_ensure_dirs()
|
||
items = []
|
||
for p in sorted(PARSES_DIR.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True)[:30]:
|
||
try:
|
||
data = json.loads(p.read_text())
|
||
items.append({
|
||
"id": data.get("id"),
|
||
"filename": data.get("filename"),
|
||
"ts": data.get("ts"),
|
||
"stats": data.get("stats"),
|
||
"formats_available": data.get("formats_available"),
|
||
})
|
||
except Exception:
|
||
continue
|
||
return {"parses": items}
|
||
|
||
|
||
@app.get("/report/{report_id}")
|
||
async def get_report(report_id: str):
|
||
path = REPORTS_DIR / f"{report_id}.json"
|
||
if not path.exists():
|
||
return JSONResponse({"error": "not found"}, status_code=404)
|
||
report = json.loads(path.read_text())
|
||
return HTMLResponse(render_report_html(report))
|
||
|
||
|
||
@app.get("/report/{report_id}/json")
|
||
async def get_report_json(report_id: str):
|
||
path = REPORTS_DIR / f"{report_id}.json"
|
||
if not path.exists():
|
||
return JSONResponse({"error": "not found"}, status_code=404)
|
||
return json.loads(path.read_text())
|
||
|
||
|
||
@app.get("/parse/{parse_id}/image/{image_idx}")
|
||
async def get_parse_image(parse_id: str, image_idx: int):
|
||
img_dir = IMAGES_DIR / parse_id
|
||
if not img_dir.exists():
|
||
return JSONResponse({"error": "not found"}, status_code=404)
|
||
for ext in ("png", "jpg", "jpeg", "webp", "gif"):
|
||
path = img_dir / f"{image_idx}.{ext}"
|
||
if path.exists():
|
||
media = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "webp": "image/webp", "gif": "image/gif"}
|
||
return FileResponse(path, media_type=media.get(ext, "image/png"))
|
||
return JSONResponse({"error": "not found"}, status_code=404)
|
||
|
||
|
||
@app.get("/parse/{parse_id}/json")
|
||
async def get_parse_json(parse_id: str):
|
||
path = PARSES_DIR / f"{parse_id}.json"
|
||
if not path.exists():
|
||
return JSONResponse({"error": "not found"}, status_code=404)
|
||
return json.loads(path.read_text())
|
||
|
||
|
||
@app.post("/assess")
|
||
async def assess_file(file: UploadFile = File(...)):
|
||
_ensure_dirs()
|
||
report_id = str(uuid.uuid4())[:8]
|
||
safe = file.filename or "upload.csv"
|
||
ext = Path(safe).suffix.lower()
|
||
content = await file.read()
|
||
dest = UPLOADS_DIR / f"{report_id}_{safe}"
|
||
dest.write_bytes(content)
|
||
|
||
docling_meta = None
|
||
parse_id: str | None = None
|
||
if ext in DOC_EXTS and ext not in DATA_EXTS:
|
||
parse_id = str(uuid.uuid4())[:8]
|
||
raw = await call_docling(content, safe)
|
||
enriched = enrich_docling_response(raw, safe, include_full_json=True, parse_id=parse_id)
|
||
docling_meta = enriched
|
||
(PARSES_DIR / f"{parse_id}.json").write_text(json.dumps({"id": parse_id, "ts": datetime.now(timezone.utc).isoformat(), **enriched}, default=str))
|
||
md = enriched.get("content", {}).get("markdown", "")
|
||
if not md.strip():
|
||
return JSONResponse({"ok": False, "error": "Docling could not extract text from this document", "docling": enriched}, status_code=422)
|
||
# Try CSV-like table from markdown
|
||
lines = [ln for ln in md.splitlines() if "|" in ln and not ln.strip().startswith("|-")]
|
||
if len(lines) >= 2:
|
||
try:
|
||
table_md = "\n".join(lines)
|
||
df = pd.read_csv(io.StringIO(table_md), sep="|", skipinitialspace=True)
|
||
df = df.loc[:, ~df.columns.str.contains("^Unnamed")]
|
||
df.columns = [str(c).strip() for c in df.columns]
|
||
df = df.dropna(how="all")
|
||
except Exception:
|
||
df = pd.DataFrame({"content": md.splitlines()[:500]})
|
||
else:
|
||
df = pd.DataFrame({"line": md.splitlines()[:1000], "line_num": range(1, min(1001, len(md.splitlines()) + 1))})
|
||
else:
|
||
df = _load_df(dest)
|
||
|
||
assessment = run_maturity_assessment(df, safe)
|
||
checks = {
|
||
"great_expectations": run_gx_checks(df),
|
||
"soda_core": run_soda_checks(df),
|
||
"deequ": run_deequ_checks(df),
|
||
"pandera": run_pandera_checks(df),
|
||
"dbt_expectations": run_dbt_checks(df),
|
||
"monte_carlo": run_montecarlo_checks(df),
|
||
"affirm": run_affirm_checks(df),
|
||
}
|
||
report = {
|
||
"id": report_id,
|
||
"filename": safe,
|
||
"ts": datetime.now(timezone.utc).isoformat(),
|
||
"assessment": assessment,
|
||
"checks": checks,
|
||
"docling": docling_meta,
|
||
}
|
||
report["executive_summary"] = _executive_summary(assessment, checks)
|
||
(REPORTS_DIR / f"{report_id}.json").write_text(json.dumps(report, indent=2, default=str))
|
||
rag_ingest = await forward_to_rag(content, safe)
|
||
return {
|
||
"ok": True,
|
||
"report_id": report_id,
|
||
"overall_score": assessment["overall_score"],
|
||
"maturity_level": assessment["maturity_level"],
|
||
"maturity_description": assessment["maturity_description"],
|
||
"rows": assessment["rows"],
|
||
"columns": assessment["columns"],
|
||
"dimensions": assessment["dimensions"],
|
||
"column_profiles": assessment["column_profiles"],
|
||
"action_items": assessment["action_items"],
|
||
"checks": checks,
|
||
"checks_summary": {
|
||
"great_expectations": {"total": len(checks["great_expectations"]), "passed": sum(1 for c in checks["great_expectations"] if c.get("success"))},
|
||
"soda_core": {"total": len(checks["soda_core"]), "warnings": sum(1 for c in checks["soda_core"] if c.get("outcome") != "pass")},
|
||
"deequ": {"total": len(checks["deequ"]), "passed": sum(1 for c in checks["deequ"] if c.get("success"))},
|
||
"pandera": {"total": len(checks["pandera"]), "passed": sum(1 for c in checks["pandera"] if c.get("success"))},
|
||
"dbt_expectations": {"total": len(checks["dbt_expectations"]), "passed": sum(1 for c in checks["dbt_expectations"] if c.get("success"))},
|
||
"monte_carlo": {"total": len(checks["monte_carlo"]), "passed": sum(1 for c in checks["monte_carlo"] if c.get("success"))},
|
||
"affirm": {"total": len(checks["affirm"]), "passed": sum(1 for c in checks["affirm"] if c.get("success"))},
|
||
},
|
||
"executive_summary": _executive_summary(assessment, checks),
|
||
"docling": {
|
||
"used": docling_meta is not None,
|
||
"parse_id": parse_id if docling_meta else None,
|
||
"document_structure": (docling_meta or {}).get("document_structure"),
|
||
"stats": (docling_meta or {}).get("stats"),
|
||
"images": (docling_meta or {}).get("images", []),
|
||
} if docling_meta else None,
|
||
"report_url": f"/dq/report/{report_id}",
|
||
"report_json_url": f"/dq/report/{report_id}/json",
|
||
"rag_ingest": rag_ingest,
|
||
}
|
||
|
||
|
||
@app.post("/parse")
|
||
async def parse_document(
|
||
file: UploadFile = File(...),
|
||
to_formats: str = Form("md,html,json"),
|
||
):
|
||
_ensure_dirs()
|
||
parse_id = str(uuid.uuid4())[:8]
|
||
safe = file.filename or "document.pdf"
|
||
content = await file.read()
|
||
raw = await call_docling(content, safe, to_formats)
|
||
if raw.get("status") == "error":
|
||
return JSONResponse({"ok": False, **raw}, status_code=502)
|
||
enriched = enrich_docling_response(raw, safe, include_full_json=True, parse_id=parse_id)
|
||
record = {"id": parse_id, "ts": datetime.now(timezone.utc).isoformat(), **enriched}
|
||
(PARSES_DIR / f"{parse_id}.json").write_text(json.dumps(record, default=str))
|
||
(UPLOADS_DIR / f"{parse_id}_{safe}").write_bytes(content)
|
||
api_payload = {
|
||
k: v for k, v in enriched.items()
|
||
if k not in ("content",)
|
||
}
|
||
api_payload["content"] = {
|
||
"preview_markdown": enriched.get("content", {}).get("preview_markdown"),
|
||
"preview_html": enriched.get("content", {}).get("preview_html"),
|
||
}
|
||
return {"ok": True, "parse_id": parse_id, **api_payload, "parse_json_url": f"/dq/parse/{parse_id}/json"}
|