atc-data-quality/dq-api/main.py

"""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"}