commit db863f96e6e2e36bc4153849aaf491524aaa0562 Author: mo Date: Thu Jun 25 00:28:10 2026 +0000 Add DQ + RAG APIs with Docling, ChromaDB, persistent ingest diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b6cf5f0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.env diff --git a/README.md b/README.md new file mode 100644 index 0000000..c809f61 --- /dev/null +++ b/README.md @@ -0,0 +1,8 @@ +# ATC Data Quality + RAG + +| Service | Port (internal) | Route | +|---------|-----------------|-------| +| dq-api | 5010 | `/dq/*` | +| rag-api | 5020 | `/rag/*` | + +Deploy with sibling repo `mo/atc-agents` — see `config/data-quality/README.md` there. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d2f0e42 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,18 @@ +services: + docling-serve: + image: quay.io/docling-project/docling-serve-cpu + restart: unless-stopped + environment: + DOCLING_SERVE_ENABLE_UI: "1" + DOCLING_SERVE_MAX_SYNC_WAIT: "300" + + dq-api: + build: ./dq-api + restart: unless-stopped + environment: + DOCLING_URL: http://docling-serve:5001 + DQ_DATA_DIR: /data + volumes: + - dq_data:/data + depends_on: + - docling-serve diff --git a/dq-api/Dockerfile b/dq-api/Dockerfile new file mode 100644 index 0000000..e086320 --- /dev/null +++ b/dq-api/Dockerfile @@ -0,0 +1,11 @@ +FROM python:3.12-slim +WORKDIR /app +RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY main.py . +RUN mkdir -p /data/reports /data/uploads +ENV DQ_DATA_DIR=/data +ENV DOCLING_URL=http://docling-serve:5001 +EXPOSE 5010 +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "5010"] diff --git a/dq-api/main.py b/dq-api/main.py new file mode 100644 index 0000000..c4ff2bb --- /dev/null +++ b/dq-api/main.py @@ -0,0 +1,1031 @@ +"""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", "docling"], + } + + +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): + row: dict[str, Any] = {"suite": "great_expectations", "expectation": exp, "success": success, "result": result} + if column: + row["column"] = column + if meta: + row["meta"] = meta + results.append(row) + + add("expect_table_row_count_to_be_between", n > 0, f"Row count: {n:,}", meta={"min": 1}) + add("expect_table_column_count_to_be_between", len(df.columns) > 0, f"Columns: {len(df.columns)}", meta={"min": 1}) + 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 = int(df.duplicated().sum()) + add("expect_table_row_count_to_equal", dup == 0, f"Duplicate rows: {dup}", meta={"duplicates": dup}) + + for col in df.columns: + null_pct = float(df[col].isnull().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)}) + 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()) + add("expect_column_values_to_be_between", True, f"Range [{mn:.4g}, {mx:.4g}]", column=str(col), meta={"min": mn, "max": mx}) + 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): + 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)) + 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)) + + passed = sum(1 for r in results if r["success"]) + 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): + checks.append({"suite": "soda_core", "name": name, "check": check, "outcome": outcome, "detail": detail}) + + soda("row_count", "row_count > 0", "pass" if n > 0 else "fail", f"{n:,} rows") + soda("duplicate_rows", f"duplicate_count = {int(df.duplicated().sum())}", "pass" if df.duplicated().sum() == 0 else "warn", f"{int(df.duplicated().sum())} duplicate rows") + + for col in df.columns[:15]: + null_c = int(df[col].isnull().sum()) + soda(f"missing_{col}", f"missing_count({col}) = {null_c}", "pass" if null_c == 0 else "warn", f"{null_c} nulls ({null_c/max(n,1)*100:.1f}%)") + 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}") + 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}") + + 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 = int(parsed.isnull().sum() - df[col].isnull().sum()) + soda(f"freshness_{col}", f"invalid_percent({col}) < 5%", "pass" if bad < n * 0.05 else "fail", f"{bad} unparseable dates") + + return checks + + +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 render_report_html(report: dict[str, Any]) -> str: + a = report.get("assessment", {}) + dims = a.get("dimensions", []) + profiles = a.get("column_profiles", [])[:20] + + def bar(score: float, color: str = "#38bdf8") -> str: + return f'
' + + dim_rows = "".join( + f""" + {d['label']}
{d['description']} + {bar(d['score'], '#4ade80' if d['score']>=80 else '#fbbf24' if d['score']>=60 else '#f87171')}{d['score']} + {d['level'].upper()} + + """ + for d in dims + ) + + col_rows = "".join( + f""" + {p['name']}{p['dtype']} + {p['null_pct']}%{p['unique_count']:,} + {', '.join(p['quality_flags']) or '—'} + {', '.join(p.get('sample_values', [])[:3]) or '—'} + """ + for p in profiles + ) + + actions = "".join( + f"
  • {x['priority'].upper()} {x['dimension']} ({x['score']}): {x['action']}
    • " + for x in a.get("action_items", []) + ) + + gx_pass = sum(1 for c in report.get("checks", {}).get("great_expectations", []) if c.get("success")) + gx_total = len(report.get("checks", {}).get("great_expectations", [])) + soda_warn = sum(1 for c in report.get("checks", {}).get("soda_core", []) if c.get("outcome") != "pass") + + return f""" +Data Maturity Report — {report.get('filename','')} +
    +

    Data Maturity Assessment Report

    +

    {report.get('filename','?')} · {report.get('ts','')} · Tools: Docling, Great Expectations, Soda Core, Pandas

    + +
    +

    Overall Score

    {a.get('overall_score',0)}/100
    +

    Maturity Level

    {a.get('maturity_level','?')}

    {a.get('maturity_description','')}

    +

    Dataset

    {a.get('rows',0):,} rows · {a.get('columns',0)} columns
    {a.get('memory_kb',0)} KB · {a.get('duplicate_pct',0)}% duplicates

    +

    Checks

    GE: {gx_pass}/{gx_total} passed
    Soda: {soda_warn} warnings

    +
    + +

    Maturity Dimensions (6 pillars)

    +{dim_rows}
    DimensionScoreLevelFindings
    + +

    Priority Remediation Roadmap

    +
      {actions or '
    • All dimensions above threshold — maintain monitoring
      • '}
    + +

    Column Profiles ({len(profiles)} shown)

    +{col_rows}
    ColumnTypeNull%UniqueFlagsSamples
    + +

    Great Expectations Results

    +
    {json.dumps(report.get('checks',{}).get('great_expectations',[]), indent=2)[:8000]}
    + +

    Soda Core Results

    +
    {json.dumps(report.get('checks',{}).get('soda_core',[]), indent=2)[:5000]}
    +
    """ + + +@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+"}, + "soda_core": {"status": "ok", "checks_per_dataset": "10-30+"}, + "pandas": {"status": "ok", "profiling": "column-level"}, + }, + } + + +@app.get("/") +async def root(): + return HTMLResponse(""" +ATC Data Quality

    Use Command Center → Data Quality tab, of /docs

    """) + + +@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 kon geen tekst extraheren uit dit 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), + } + report = { + "id": report_id, + "filename": safe, + "ts": datetime.now(timezone.utc).isoformat(), + "assessment": assessment, + "checks": checks, + "docling": docling_meta, + } + (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["success"])}, + "soda_core": {"total": len(checks["soda_core"]), "warnings": sum(1 for c in checks["soda_core"] if c["outcome"] != "pass")}, + }, + "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"} diff --git a/dq-api/requirements.txt b/dq-api/requirements.txt new file mode 100644 index 0000000..637cc6f --- /dev/null +++ b/dq-api/requirements.txt @@ -0,0 +1,7 @@ +fastapi==0.115.6 +uvicorn[standard]==0.34.0 +httpx==0.28.1 +pandas==2.2.3 +openpyxl==3.1.5 +python-multipart==0.0.20 +pydantic==2.10.4 diff --git a/rag-api/Dockerfile b/rag-api/Dockerfile new file mode 100644 index 0000000..55e2b57 --- /dev/null +++ b/rag-api/Dockerfile @@ -0,0 +1,12 @@ +FROM python:3.12-slim +WORKDIR /app +RUN apt-get update && apt-get install -y --no-install-recommends curl build-essential && rm -rf /var/lib/apt/lists/* +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY main.py . +RUN mkdir -p /data/uploads +ENV RAG_DATA_DIR=/data +ENV CHROMA_HOST=chromadb +ENV CHROMA_PORT=8000 +EXPOSE 5020 +CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "5020"] diff --git a/rag-api/main.py b/rag-api/main.py new file mode 100644 index 0000000..2490962 --- /dev/null +++ b/rag-api/main.py @@ -0,0 +1,548 @@ +"""RAG Knowledge API — LangChain + ChromaDB + Docling + LLM with persistent document registry.""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import httpx +from fastapi import FastAPI, File, Form, UploadFile +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse, JSONResponse +from langchain_community.embeddings import HuggingFaceEmbeddings +from langchain_community.vectorstores import Chroma +from langchain_core.documents import Document +from langchain_core.messages import HumanMessage, SystemMessage +from langchain_openai import ChatOpenAI +from langchain_text_splitters import RecursiveCharacterTextSplitter +import chromadb +from pydantic import BaseModel + +CHROMA_HOST = os.getenv("CHROMA_HOST", "chromadb") +CHROMA_PORT = int(os.getenv("CHROMA_PORT", "8000")) +CHROMA_URL = f"http://{CHROMA_HOST}:{CHROMA_PORT}" +DOCLING_URL = os.getenv("DOCLING_URL", "http://docling-serve:5001").rstrip("/") +LLM_URL = os.getenv("LLM_URL", "http://10.0.20.106:8001/v1") +LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o") +LLM_API_KEY = os.getenv("LLM_API_KEY", "sk-local") +DATA_DIR = Path(os.getenv("RAG_DATA_DIR", "/data")) +UPLOADS_DIR = DATA_DIR / "uploads" +REGISTRY_PATH = DATA_DIR / "document_registry.json" +EMBED_MODEL = os.getenv("EMBED_MODEL", "sentence-transformers/all-MiniLM-L6-v2") + +# Use placeholder images in markdown — embedded base64 destroys RAG quality. +DOCLING_IMAGE_MODE = os.getenv("DOCLING_IMAGE_MODE", "placeholder") + +app = FastAPI(title="ATC RAG Knowledge API", version="1.2.0") +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) + +_splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120) +_embeddings: HuggingFaceEmbeddings | None = None + +_BASE64_BLOB = re.compile(r"[A-Za-z0-9+/]{120,}={0,2}") +_BASE64_IMG = re.compile(r"!\[[^\]]*\]\(data:image/[^)]+\)", re.IGNORECASE) +_IMAGE_REF = re.compile(r"!\[Image\]\([^)]+\)") + + +def _ensure_dirs() -> None: + UPLOADS_DIR.mkdir(parents=True, exist_ok=True) + if not REGISTRY_PATH.exists(): + REGISTRY_PATH.write_text(json.dumps({"documents": []}, indent=2)) + + +def _load_registry() -> list[dict[str, Any]]: + _ensure_dirs() + try: + data = json.loads(REGISTRY_PATH.read_text()) + return data.get("documents", []) + except Exception: + return [] + + +def _save_registry(docs: list[dict[str, Any]]) -> None: + _ensure_dirs() + REGISTRY_PATH.write_text( + json.dumps({"documents": docs, "updated_at": datetime.now(timezone.utc).isoformat()}, indent=2, default=str) + ) + + +def _file_hash(content: bytes) -> str: + return hashlib.sha256(content).hexdigest() + + +def _safe_collection(name: str) -> str: + return "".join(c if c.isalnum() or c in "-_" else "_" for c in name.strip())[:64] or "default" + + +def clean_text_for_rag(text: str) -> str: + """Strip embedded images and base64 blobs that pollute vector search.""" + text = _BASE64_IMG.sub("", text) + text = _IMAGE_REF.sub("", text) + text = _BASE64_BLOB.sub("", text) + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +def is_garbage_chunk(text: str) -> bool: + """Detect chunks that are mostly binary/base64 noise.""" + if not text or len(text) < 20: + return True + if "data:image" in text: + return True + if _BASE64_BLOB.search(text): + return True + alpha = sum(1 for c in text if c.isalpha() or c.isspace()) + if alpha / max(len(text), 1) < 0.35: + return True + return False + + +def get_embeddings() -> HuggingFaceEmbeddings: + global _embeddings + if _embeddings is None: + _embeddings = HuggingFaceEmbeddings(model_name=EMBED_MODEL) + return _embeddings + + +def get_chroma_client() -> chromadb.HttpClient: + return chromadb.HttpClient(host=CHROMA_HOST, port=CHROMA_PORT) + + +def get_vectorstore(collection: str) -> Chroma: + return Chroma( + client=get_chroma_client(), + collection_name=collection, + embedding_function=get_embeddings(), + ) + + +def get_llm(temperature: float = 0.2) -> ChatOpenAI: + return ChatOpenAI( + base_url=LLM_URL, + api_key=LLM_API_KEY, + model=LLM_MODEL, + temperature=temperature, + ) + + +def _delete_doc_vectors(collection: str, doc_id: str) -> None: + try: + col = get_chroma_client().get_collection(collection) + col.delete(where={"doc_id": doc_id}) + except Exception: + pass + + +async def extract_text(content: bytes, filename: str) -> str: + ext = Path(filename).suffix.lower() + if ext in {".txt", ".md", ".csv", ".json"}: + try: + raw = content.decode("utf-8") + except UnicodeDecodeError: + raw = content.decode("latin-1", errors="replace") + return clean_text_for_rag(raw) + + async with httpx.AsyncClient(timeout=300.0) as client: + r = await client.post( + f"{DOCLING_URL}/v1/convert/file", + files={"files": (filename, content, "application/octet-stream")}, + data={ + "to_formats": ["md"], + "image_export_mode": DOCLING_IMAGE_MODE, + "do_ocr": "true", + "table_mode": "accurate", + }, + ) + if r.status_code >= 400: + raise ValueError(f"Docling failed: {r.text[:300]}") + doc = r.json().get("document") or {} + md = doc.get("md_content") or doc.get("text_content") or "" + md = clean_text_for_rag(md) + if len(md) < 50: + raise ValueError("No readable text extracted from document") + return md + + +def _find_duplicate(content_hash: str, collection: str) -> dict[str, Any] | None: + for d in _load_registry(): + if d.get("content_hash") == content_hash and d.get("collection") == collection: + return d + return None + + +def _find_by_id(doc_id: str) -> dict[str, Any] | None: + for d in _load_registry(): + if d.get("id") == doc_id: + return d + return None + + +async def _ingest_bytes( + content: bytes, + filename: str, + collection: str, + source: str = "upload", + *, + force_reindex: bool = False, +) -> dict[str, Any]: + col = _safe_collection(collection) + content_hash = _file_hash(content) + existing = _find_duplicate(content_hash, col) + + if existing and not force_reindex: + return { + "ok": True, + "duplicate": True, + "skipped": True, + "message": f"Document already indexed as '{existing['filename']}' — chat immediately, no re-upload needed.", + **{k: existing[k] for k in ("id", "filename", "collection", "chunks", "characters", "ingested_at") if k in existing}, + } + + if existing and force_reindex: + doc_id = existing["id"] + _delete_doc_vectors(col, doc_id) + stored_path = Path(existing.get("stored_path", "")) + if stored_path.exists(): + stored_path.write_bytes(content) + else: + stored_name = f"{doc_id}_{Path(filename).name}" + stored_path = UPLOADS_DIR / stored_name + stored_path.write_bytes(content) + else: + doc_id = uuid.uuid4().hex[:12] + stored_name = f"{doc_id}_{Path(filename).name}" + stored_path = UPLOADS_DIR / stored_name + stored_path.write_bytes(content) + + text = await extract_text(content, filename) + chunks = [c for c in _splitter.split_text(text) if not is_garbage_chunk(c)] + if not chunks: + raise ValueError("No usable text chunks after cleaning — document may be image-only") + + docs = [ + Document( + page_content=chunk, + metadata={ + "source": filename, + "doc_id": doc_id, + "chunk": i, + "content_hash": content_hash, + "ingested_at": datetime.now(timezone.utc).isoformat(), + }, + ) + for i, chunk in enumerate(chunks) + ] + vs = get_vectorstore(col) + vs.add_documents(docs) + + record = { + "id": doc_id, + "filename": filename, + "collection": col, + "content_hash": content_hash, + "stored_path": str(stored_path), + "chunks": len(chunks), + "characters": len(text), + "bytes": len(content), + "source": source, + "ingested_at": datetime.now(timezone.utc).isoformat(), + } + registry = _load_registry() + registry = [d for d in registry if not (d.get("id") == doc_id and d.get("collection") == col)] + registry.insert(0, record) + _save_registry(registry) + return {"ok": True, "duplicate": False, "reindexed": force_reindex, **record} + + +async def _summarize_text(text: str, filename: str) -> str: + """Summarize document text using LLM with map-reduce for long docs.""" + llm = get_llm(temperature=0.1) + max_chunk = 12000 + if len(text) <= max_chunk: + prompt = ( + f"Summarize this document ({filename}) clearly in English. " + "Include: main topic, key sections, important technologies/products mentioned, and target audience. " + "Use bullet points and short paragraphs.\n\nDocument:\n{text}" + ) + resp = llm.invoke([HumanMessage(content=prompt.format(text=text[:max_chunk]))]) + return resp.content if hasattr(resp, "content") else str(resp) + + # Map-reduce for long documents + parts = [text[i : i + max_chunk] for i in range(0, min(len(text), 60000), max_chunk)] + partials: list[str] = [] + for i, part in enumerate(parts[:5]): + resp = llm.invoke([ + HumanMessage(content=( + f"Summarize part {i + 1}/{min(len(parts), 5)} of '{filename}'. " + f"List key topics, products, and technical points:\n\n{part}" + )) + ]) + partials.append(resp.content if hasattr(resp, "content") else str(resp)) + + combined = "\n\n".join(partials) + final = llm.invoke([ + HumanMessage(content=( + f"Create a clear executive summary of '{filename}' from these section summaries. " + "Structure: Overview, Main Topics, Key Technologies, Audience. Use bullet points.\n\n" + f"{combined}" + )) + ]) + return final.content if hasattr(final, "content") else str(final) + + +class ChatRequest(BaseModel): + message: str + collection: str = "default" + top_k: int = 5 + + +class IngestTextRequest(BaseModel): + text: str + collection: str = "default" + source: str = "manual" + + +class SummarizeRequest(BaseModel): + collection: str = "default" + doc_id: str | None = None + filename: str | None = None + + +@app.get("/health") +async def health(): + chroma_ok = docling_ok = llm_ok = False + doc_count = len(_load_registry()) + try: + async with httpx.AsyncClient(timeout=5.0) as c: + cr = await c.get(f"{CHROMA_URL}/api/v1/heartbeat") + chroma_ok = cr.status_code < 400 + dr = await c.get(f"{DOCLING_URL}/health") + docling_ok = dr.status_code < 400 and dr.json().get("status") == "ok" + lr = await c.get(f"{LLM_URL.rstrip('/')}/models") + llm_ok = lr.status_code < 400 + except Exception: + pass + return { + "ok": chroma_ok, + "chroma": chroma_ok, + "docling": docling_ok, + "llm": llm_ok, + "embed_model": EMBED_MODEL, + "stored_documents": doc_count, + "persistent": True, + "docling_image_mode": DOCLING_IMAGE_MODE, + } + + +@app.get("/documents") +async def list_documents(collection: str | None = None): + docs = _load_registry() + if collection: + col = _safe_collection(collection) + docs = [d for d in docs if d.get("collection") == col] + return {"documents": docs, "total": len(docs)} + + +@app.get("/documents/{doc_id}") +async def get_document(doc_id: str): + doc = _find_by_id(doc_id) + if doc: + return doc + return JSONResponse({"error": "not found"}, status_code=404) + + +@app.get("/documents/{doc_id}/file") +async def download_document(doc_id: str): + doc = _find_by_id(doc_id) + if doc: + path = Path(doc.get("stored_path", "")) + if path.exists(): + return FileResponse(path, filename=doc.get("filename", path.name)) + return JSONResponse({"error": "not found"}, status_code=404) + + +@app.post("/documents/{doc_id}/reindex") +async def reindex_document(doc_id: str): + doc = _find_by_id(doc_id) + if not doc: + return JSONResponse({"error": "not found"}, status_code=404) + path = Path(doc.get("stored_path", "")) + if not path.exists(): + return JSONResponse({"error": "stored file missing"}, status_code=404) + try: + content = path.read_bytes() + result = await _ingest_bytes( + content, + doc["filename"], + doc["collection"], + source=doc.get("source", "reindex"), + force_reindex=True, + ) + return result + except ValueError as exc: + return JSONResponse({"ok": False, "error": str(exc)}, status_code=422) + + +@app.get("/collections") +async def list_collections(): + try: + client = get_chroma_client() + cols = client.list_collections() + registry = _load_registry() + items = [] + for col in cols: + files = {d["filename"] for d in registry if d.get("collection") == col.name} + items.append({ + "name": col.name, + "documents": col.count(), + "files": len(files), + "filenames": sorted(files)[:20], + }) + return {"collections": items} + except Exception as exc: + return JSONResponse({"error": str(exc), "collections": []}, status_code=502) + + +@app.post("/collections") +async def create_collection(name: str = Form(...)): + safe = _safe_collection(name) + get_vectorstore(safe) + return {"ok": True, "collection": safe} + + +@app.post("/ingest") +async def ingest_file( + file: UploadFile = File(...), + collection: str = Form("default"), + force_reindex: bool = Form(False), +): + _ensure_dirs() + safe_name = file.filename or "upload.txt" + content = await file.read() + try: + result = await _ingest_bytes(content, safe_name, collection, force_reindex=force_reindex) + if not result.get("ok"): + return JSONResponse(result, status_code=422) + return result + except ValueError as exc: + return JSONResponse({"ok": False, "error": str(exc)}, status_code=422) + + +@app.post("/ingest/text") +async def ingest_text(body: IngestTextRequest): + col = _safe_collection(body.collection) + content = body.text.encode("utf-8") + filename = f"{body.source}.txt" + return await _ingest_bytes(content, filename, col, source=body.source) + + +@app.post("/summarize") +async def summarize(body: SummarizeRequest): + col = _safe_collection(body.collection) + registry = _load_registry() + doc: dict[str, Any] | None = None + if body.doc_id: + doc = _find_by_id(body.doc_id) + elif body.filename: + for d in registry: + if d.get("filename") == body.filename and d.get("collection") == col: + doc = d + break + else: + docs_in_col = [d for d in registry if d.get("collection") == col] + if len(docs_in_col) == 1: + doc = docs_in_col[0] + + if not doc: + return JSONResponse({"ok": False, "error": "Document not found — specify doc_id or filename"}, status_code=404) + + path = Path(doc.get("stored_path", "")) + if not path.exists(): + return JSONResponse({"ok": False, "error": "Stored file missing"}, status_code=404) + + try: + content = path.read_bytes() + text = await extract_text(content, doc["filename"]) + summary = await _summarize_text(text, doc["filename"]) + return { + "ok": True, + "summary": summary, + "filename": doc["filename"], + "doc_id": doc["id"], + "collection": col, + "characters": len(text), + } + except ValueError as exc: + return JSONResponse({"ok": False, "error": str(exc)}, status_code=422) + except Exception as exc: + return JSONResponse({"ok": False, "error": f"Summarize failed: {exc}"}, status_code=502) + + +@app.post("/chat") +async def chat(body: ChatRequest): + col = _safe_collection(body.collection) + try: + vs = get_vectorstore(col) + count = get_chroma_client().get_collection(col).count() + except Exception as exc: + return JSONResponse({"ok": False, "error": f"Collection unavailable: {exc}"}, status_code=404) + + if count == 0: + stored = [d for d in _load_registry() if d.get("collection") == col] + if stored: + return JSONResponse({ + "ok": False, + "error": "Vectors missing but files exist — click Re-index on the document in the library.", + "stored_documents": len(stored), + }, status_code=400) + return JSONResponse({"ok": False, "error": "Collection is empty — upload documents first."}, status_code=400) + + retriever = vs.as_retriever(search_kwargs={"k": min(body.top_k * 3, 20)}) + raw_docs = retriever.invoke(body.message) + docs = [d for d in raw_docs if not is_garbage_chunk(d.page_content)][: body.top_k] + + if not docs: + return JSONResponse({ + "ok": False, + "error": "Retrieved chunks are corrupted (old base64 index). Click Re-index on the document.", + }, status_code=400) + + context = "\n\n---\n\n".join( + f"[Source: {d.metadata.get('source', '?')} | chunk {d.metadata.get('chunk', '?')}]\n{d.page_content}" + for d in docs + ) + + system = ( + "You are a helpful data assistant for the Dell ATC platform. " + "Answer ONLY based on the provided context. If the context does not contain the answer, say so clearly. " + "Cite sources by filename when relevant. Be concise and technical." + ) + user = f"Context:\n{context}\n\nQuestion: {body.message}" + + try: + llm = get_llm() + resp = llm.invoke([SystemMessage(content=system), HumanMessage(content=user)]) + answer = resp.content if hasattr(resp, "content") else str(resp) + except Exception as exc: + return JSONResponse({"ok": False, "error": f"LLM error: {exc}"}, status_code=502) + + sources = [ + {"source": d.metadata.get("source"), "chunk": d.metadata.get("chunk"), "preview": d.page_content[:200]} + for d in docs + ] + return {"ok": True, "answer": answer, "sources": sources, "collection": col, "context_chunks": len(docs)} + + +@app.get("/") +async def root(): + return { + "service": "ATC RAG Knowledge API", + "persistent_storage": "ChromaDB + document registry on disk", + "endpoints": ["/health", "/documents", "/collections", "/ingest", "/chat", "/summarize", "/docs"], + } diff --git a/rag-api/requirements.txt b/rag-api/requirements.txt new file mode 100644 index 0000000..8ab21b9 --- /dev/null +++ b/rag-api/requirements.txt @@ -0,0 +1,11 @@ +fastapi==0.115.6 +uvicorn[standard]==0.34.0 +httpx==0.28.1 +python-multipart==0.0.20 +pydantic==2.10.4 +langchain==0.3.14 +langchain-community==0.3.14 +langchain-openai==0.2.14 +langchain-text-splitters==0.3.4 +chromadb==0.5.23 +sentence-transformers==3.3.1