-- Install the Quack extension
INSTALL quack;
LOAD quack;

-- Generate sample data
CREATE TABLE employees AS
SELECT * FROM quack.generate_series(1, 1000) AS id
CROSS JOIN LATERAL (
    SELECT
        id,
        CASE WHEN RANDOM() > 0.1 THEN 'John' || id ELSE NULL END AS name,
        CASE WHEN RANDOM() > 0.05 THEN floor(RANDOM() * 65) + 20 ELSE NULL END AS age,
        CASE WHEN RANDOM() > 0.15 THEN '2020-' || lpad(floor(RANDOM()*12)::text, 2, '0')
             || '-' || lpad(floor(RANDOM()*28+1)::text, 2, '0') ELSE NULL END AS hire_date,
        CASE WHEN RANDOM() > 0.08 THEN round(RANDOM() * 150000, 2) ELSE NULL END AS salary,
        CASE WHEN RANDOM() > 0.2 THEN ('dept_' || floor(RANDOM()*5+1))::varchar
             ELSE NULL END AS department
);

-- 1. Completeness checks: detect null values and uniqueness constraints
SELECT
    COUNT(*) AS total_rows,
    COUNT(id) AS non_null_id,
    COUNT(name) AS non_null_name,
    COUNT(age) AS non_null_age,
    COUNT(salary) AS non_null_salary,
    COUNT(department) AS non_null_department,
    COUNT(DISTINCT id) AS unique_ids,
    -- Completeness ratios
    1.0 * COUNT(id) / COUNT(*) AS id_completeness,
    1.0 * COUNT(name) / COUNT(*) AS name_completeness
FROM employees;

-- 2. Numeric range checks: ensure salary is within reasonable bounds
SELECT
    COUNT(*) AS violations,
    MIN(salary) AS min_salary,
    MAX(salary) AS max_salary,
    AVG(salary) AS avg_salary
FROM employees
WHERE salary < 0 OR salary > 500000;

-- 3. Date validity checks
SELECT
    COUNT(*) AS invalid_dates,
    MIN(hire_date) AS earliest_hire,
    MAX(hire_date) AS latest_hire
FROM employees
WHERE hire_date IS NOT NULL
  AND (hire_date < DATE '2000-01-01' OR hire_date > CURRENT_DATE);

-- 4. Distribution consistency checks: is department distribution balanced?
SELECT
    department,
    COUNT(*) AS emp_count,
    ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER(), 2) AS pct_of_total
FROM employees
GROUP BY department
ORDER BY emp_count DESC;

-- Quack provides out-of-the-box validators
SELECT
    quack_is_not_null(id, 'ID cannot be null') AS id_check,
    quack_is_not_null(name, 'Name cannot be null') AS name_check,
    quack_between(age, 18, 70, 'Age should be between 18-70') AS age_check,
    quack_between(salary, 0, 500000, 'Salary should be in reasonable range') AS salary_check,
    quack_regex_match(department, '^dept_\d+$', 'Department format is incorrect') AS dept_check
FROM employees
LIMIT 5;

-- rules/completeness.sql
SELECT 'completeness' AS rule_type,
       'employees' AS table_name,
       'id' AS column_name,
       COUNT(*) FILTER (WHERE id IS NULL) AS violation_count,
       COUNT(*) AS total_rows,
       ROUND(100.0 * COUNT(*) FILTER (WHERE id IS NULL) / COUNT(*), 2) AS violation_rate
FROM employees;

-- rules/range.sql
SELECT 'range' AS rule_type,
       'employees' AS table_name,
       'salary' AS column_name,
       COUNT(*) AS violation_count,
       COUNT(*) AS total_rows,
       ROUND(100.0 * COUNT(*) FILTER (WHERE salary < 0 OR salary > 500000) / COUNT(*), 2) AS violation_rate
FROM employees;

-- rules/uniqueness.sql
SELECT 'uniqueness' AS rule_type,
       'employees' AS table_name,
       'id' AS column_name,
       COUNT(*) - COUNT(DISTINCT id) AS violation_count,
       COUNT(*) AS total_rows,
       ROUND(100.0 * (COUNT(*) - COUNT(DISTINCT id)) / COUNT(*), 2) AS violation_rate
FROM employees;

# quality_engine.py
import duckdb
import json
from pathlib import Path
from datetime import datetime

class DataQualityEngine:
    def __init__(self, db_path="data_warehouse.duckdb"):
        self.con = duckdb.connect(db_path)
        self.rules_dir = Path("rules")
        self.results_history = []

    def load_rules(self):
        """Dynamically load all quality rules"""
        rules = {}
        for rule_file in self.rules_dir.glob("*.sql"):
            rule_name = rule_file.stem
            with open(rule_file) as f:
                rules[rule_name] = f.read()
        return rules

    def run_all_checks(self, source_table="employees"):
        """Execute all data quality checks"""
        rules = self.load_rules()
        results = []

        for rule_name, sql_template in rules.items():
            # Replace table name
            sql = sql_template.replace("employees", source_table)
            result = self.con.execute(sql).fetchdf()

            if not result.empty:
                row = result.iloc[0]
                results.append({
                    "timestamp": datetime.now().isoformat(),
                    "rule": rule_name,
                    "table": source_table,
                    "column": row.get("column_name", "N/A"),
                    "violation_count": int(row.get("violation_count", 0)),
                    "total_rows": int(row.get("total_rows", 0)),
                    "violation_rate": float(row.get("violation_rate", 0)),
                    "status": "PASS" if row.get("violation_rate", 0) < 1.0 else "FAIL"
                })

        # Save historical results
        self.results_history.extend(results)
        self._save_history()
        return results

    def _save_history(self):
        """Save history results to DuckDB"""
        if self.results_history:
            df = __import__('pandas').DataFrame(self.results_history)
            self.con.execute("DROP TABLE IF EXISTS quality_results")
            self.con.execute("CREATE TABLE quality_results AS SELECT * FROM df")

    def get_quality_score(self, source_table="employees"):
        """Calculate overall quality score (0-100)"""
        results = self.run_all_checks(source_table)
        if not results:
            return 100.0

        # Weighted scoring: different rules have different weights
        weights = {
            "completeness": 0.4,
            "range": 0.3,
            "uniqueness": 0.3
        }

        total_score = 0
        total_weight = 0
        for r in results:
            weight = weights.get(r["rule"], 0.2)
            # Lower violation rate means higher score
            score = max(0, 100 - r["violation_rate"] * 10)
            total_score += score * weight
            total_weight += weight

        return round(total_score / total_weight, 2) if total_weight > 0 else 100

    def generate_report(self, source_table="employees"):
        """Generate HTML quality report"""
        results = self.run_all_checks(source_table)
        score = self.get_quality_score(source_table)

        color = "green" if score >= 90 else "orange" if score >= 70 else "red"
        html = f"""
        <html><head><title>Data Quality Report</title></head><body>
        <h1>📊 Data Quality Assessment Report</h1>
        <h2>Overall Score: <span style="color: {color}">{score}/100</span></h2>
        <table border="1" cellpadding="8">
        <tr><th>Rule</th><th>Table</th><th>Column</th><th>Violations</th><th>Rate%</th><th>Status</th></tr>
        """
        for r in results:
            status_color = "#4caf50" if r["status"] == "PASS" else "#f44336"
            html += f"""<tr>
                <td>{r['rule']}</td>
                <td>{r['table']}</td>
                <td>{r['column']}</td>
                <td>{r['violation_count']}</td>
                <td>{r['violation_rate']:.2f}%</td>
                <td style="color:{status_color};font-weight:bold">{r['status']}</td>
            </tr>"""

        html += "</table></body></html>"
        return html


# Usage example
if __name__ == "__main__":
    engine = DataQualityEngine()
    scores = engine.run_all_checks()
    overall = engine.get_quality_score()
    print(f"Overall data quality score: {overall}/100")
    report_html = engine.generate_report()
    with open("quality_report.html", "w") as f:
        f.write(report_html)
    print("HTML report generated: quality_report.html")

# .github/workflows/data-quality.yml
name: Data Quality Check

on:
  schedule:
    - cron: '0 2 * * *'  # Daily at 2 AM
  workflow_dispatch:       # Manual trigger

jobs:
  quality-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install DuckDB CLI
        run: |
          curl -L https://github.com/duckdb/duckdb/releases/download/v1.5.4/duckdb_cli-linux-amd64.zip -o duckdb.zip
          unzip duckdb.zip
          chmod +x duckdb

      - name: Run Data Quality Checks
        run: |
          ./duckdb data_warehouse.duckdb -c "INSTALL quack; LOAD quack;" \
                                          -c ".read rules/completeness.sql" \
                                          -c ".read rules/range.sql" \
                                          -c ".read rules/uniqueness.sql"

      - name: Generate Report
        run: python quality_engine.py

      - name: Slack Notification
        if: failure()
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            {"text": "⚠️ Data quality check failed! Please review the report."}
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK }}

-- Cross-source quality comparison
ATTACH 'sales.duckdb' AS sales_db;
ATTACH 'inventory.duckdb' AS inv_db;

-- Unified quality scoring
WITH quality_scores AS (
    SELECT 'sales' AS source,
           ROUND(100.0 * COUNT(*) FILTER (WHERE order_id IS NOT NULL) / COUNT(*), 2) AS completeness_score
    FROM sales_db.orders
    UNION ALL
    SELECT 'inventory' AS source,
           ROUND(100.0 * COUNT(*) FILTER (WHERE product_id IS NOT NULL) / COUNT(*), 2) AS completeness_score
    FROM inv_db.products
)
SELECT source,
       completeness_score,
       CASE
           WHEN completeness_score >= 99 THEN '✅ Excellent'
           WHEN completeness_score >= 95 THEN '⚠️ Good'
           WHEN completeness_score >= 90 THEN '🔶 Average'
           ELSE '❌ Needs Improvement'
       END AS quality_level
FROM quality_scores
ORDER BY completeness_score DESC;