# generate_dbt_tests.py
from anthropic import Anthropic
import yaml
from pathlib import Path

client = Anthropic()

def generate_dbt_tests(model_sql: str, model_name: str, sample_data: list[dict]) -> dict:
    """Generate dbt test YAML for a model given its SQL and sample data."""
    sample_json = str(sample_data[:10])

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": f"""Generate comprehensive dbt tests for this model.

Model name: {model_name}
Sample data (10 rows): {sample_json}

Model SQL:
{model_sql}

Generate YAML with:
1. Column-level tests: not_null, unique, accepted_values (based on actual values in sample)
2. At least 2 custom relationship tests if foreign keys are apparent
3. At least 1 custom SQL test for business rule validation
4. Source freshness tests if there's a created_at/updated_at column

Return valid dbt YAML schema format only."""
        }]
    )

    # Parse and validate the YAML
    yaml_text = response.content[0].text
    if yaml_text.startswith("```"):
 yaml_text = "\n".join(yaml_text.split("\n")[1:-1])

 return yaml.safe_load(yaml_text)

# Example usage
model_sql = """
SELECT
 order_id,
 user_id,
 status,
 total_amount,
 currency,
 created_at,
 shipped_at
FROM {{ source('ecommerce', 'orders') }}
WHERE created_at >= '2024-01-01'
"""

sample_data = [
 {"order_id": "ord_001", "user_id": "usr_123", "status": "shipped",
 "total_amount": 4999, "currency": "USD", "created_at": "2026-01-15",
 "shipped_at": "2026-01-17"},
 {"order_id": "ord_002", "user_id": "usr_456", "status": "pending",
 "total_amount": 1299, "currency": "USD", "created_at": "2026-01-16",
 "shipped_at": None},
]

tests = generate_dbt_tests(model_sql, "orders", sample_data)
print(yaml.dump(tests, default_flow_style=False))

# generate_ge_suite.py
import pandas as pd
import json
from anthropic import Anthropic

client = Anthropic()

def profile_dataframe(df: pd.DataFrame) -> dict:
 """Generate a statistical profile for AI to analyze."""
 profile = {}
 for col in df.columns:
 col_profile = {
 "dtype": str(df[col].dtype),
 "null_count": int(df[col].isna().sum()),
 "null_pct": round(df[col].isna().mean() * 100, 1),
 "unique_count": int(df[col].nunique()),
 }

 if df[col].dtype in ["int64", "float64"]:
 col_profile.update({
 "min": float(df[col].min()),
 "max": float(df[col].max()),
 "mean": round(float(df[col].mean()), 2),
 "p5": float(df[col].quantile(0.05)),
 "p95": float(df[col].quantile(0.95)),
 })
 elif df[col].dtype == "object":
 col_profile["sample_values"] = df[col].dropna().unique()[:10].tolist()

 profile[col] = col_profile

 return profile

def generate_ge_expectations(df: pd.DataFrame, dataset_name: str) -> list[dict]:
 """Generate Great Expectations expectations for a DataFrame."""
 profile = profile_dataframe(df)
 row_count = len(df)

 response = client.messages.create(
 model="claude-opus-4-6",
 max_tokens=2500,
 messages=[{
 "role": "user",
 "content": f"""Generate Great Expectations expectations for this dataset.

Dataset: {dataset_name}
Row count: {row_count}
Column profiles:
{json.dumps(profile, indent=2, default=str)}

Return a JSON array of expectation objects using Great Expectations API format.
Include:
- expect_column_values_to_not_be_null for non-null columns
- expect_column_values_to_be_between for numeric columns (use p5/p95 with buffer)
- expect_column_values_to_be_in_set for low-cardinality string columns
- expect_column_to_exist for all columns
- expect_table_row_count_to_be_between (50% to 200% of current count)

Return only valid JSON, no explanation."""
 }]
 )

 text = response.content[0].text
 if text.startswith("```"):
        text = "\n".join(text.split("\n")[1:-1])

    return json.loads(text)

def save_ge_suite(expectations: list[dict], suite_name: str, output_dir: str = "great_expectations/expectations"):
    """Save expectations to a GE suite file."""
    suite = {
        "data_asset_type": None,
        "expectation_suite_name": suite_name,
        "expectations": expectations,
        "meta": {"great_expectations_version": "0.18.0"}
    }
    output_path = f"{output_dir}/{suite_name}.json"
    with open(output_path, "w") as f:
        json.dump(suite, f, indent=2)
    print(f"Saved {len(expectations)} expectations to {output_path}")

# pipeline_output_checker.py
from anthropic import Anthropic
import pandas as pd
import json

client = Anthropic()

def compare_pipeline_runs(
    current_df: pd.DataFrame,
    baseline_df: pd.DataFrame,
    pipeline_name: str
) -> dict:
    """Use Claude to identify significant differences between two pipeline outputs."""
    def summarize_df(df: pd.DataFrame) -> dict:
        return {
            "row_count": len(df),
            "columns": list(df.columns),
            "numeric_stats": df.describe().round(2).to_dict(),
            "null_counts": df.isna().sum().to_dict(),
        }

    current_summary = summarize_df(current_df)
    baseline_summary = summarize_df(baseline_df)

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": f"""Compare these two pipeline run outputs and identify anomalies.

Pipeline: {pipeline_name}

Baseline run:
{json.dumps(baseline_summary, indent=2)}

Current run:
{json.dumps(current_summary, indent=2)}

Identify:
1. ROW_COUNT_CHANGE: Is the change in row count expected or suspicious?
2. DISTRIBUTION_SHIFTS: Are any numeric distributions significantly different?
3. NULL_CHANGES: Have null rates changed significantly (>5% absolute change)?
4. MISSING_COLUMNS: Any columns present in baseline but missing from current?
5. VERDICT: [PASS / WARN / FAIL] — should this pipeline output be trusted?

Be specific with numbers."""
        }]
    )

    text = response.content[0].text

    # Parse verdict
    verdict = "WARN"
    if "VERDICT: PASS" in text:
        verdict = "PASS"
    elif "VERDICT: FAIL" in text:
        verdict = "FAIL"

    return {
        "verdict": verdict,
        "analysis": text,
        "pipeline": pipeline_name
    }

# airflow_dag_with_checks.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import pandas as pd

def run_pipeline_with_check(**context):
    """Run pipeline and validate output before marking success."""
    # Run your actual pipeline
    current_df = run_etl_pipeline()

    # Load baseline from yesterday's run
    baseline_df = load_baseline_from_s3(
        date=context["ds"],
        pipeline="orders_daily"
    )

    result = compare_pipeline_runs(current_df, baseline_df, "orders_daily")

    if result["verdict"] == "FAIL":
        raise ValueError(f"Pipeline output failed quality check:\n{result['analysis']}")
    elif result["verdict"] == "WARN":
        # Log warning but don't fail the DAG
        context["task_instance"].xcom_push(
            key="quality_warning",
            value=result["analysis"]
        )

    return result["verdict"]

with DAG(
    "orders_daily_pipeline",
    default_args={"retries": 1, "retry_delay": timedelta(minutes=5)},
    schedule_interval="0 6 * * *",
    start_date=datetime(2026, 1, 1),
) as dag:

    run_and_check = PythonOperator(
        task_id="run_pipeline_with_quality_check",
        python_callable=run_pipeline_with_check,
        provide_context=True
    )