import pandas as pd
 
# Dimension weights — must sum to 1.0
# Adjust these weights based on business requirements
WEIGHTS = {
    'completeness': 0.30,   # missing data is the most common pain point
    'validity':     0.30,   # invalid values corrupt downstream calculations
    'uniqueness':   0.20,   # duplicates cause double-counting
    'consistency':  0.20,   # inconsistent values break JOINs and groupbys
}
 
assert abs(sum(WEIGHTS.values()) - 1.0) < 1e-9, "Weights must sum to 1.0"
 
 
def compute_dimension_scores(
    df: pd.DataFrame,
    id_col: str = None,
    amount_col: str = None,
    date_col: str = None,
    status_col: str = None,
    valid_statuses: set = None,
) -> dict:
    """
    Compute a 0.0-1.0 score for each quality dimension.
    Returns a dict: {dimension_name: score}
    """
    scores = {}
 
    # --- COMPLETENESS: fraction of non-null values across all columns ---
    scores['completeness'] = float(1 - df.isnull().mean().mean())
 
    # --- VALIDITY: check date column for bad/future dates + amount column for negatives ---
    validity_violations = 0
    validity_checks = 0
 
    if date_col and date_col in df.columns:
        dates = pd.to_datetime(df[date_col], errors='coerce')
        validity_violations += int(dates.isna().sum())
        validity_violations += int((dates > pd.Timestamp.today()).sum())
        validity_checks += len(df)
 
    if amount_col and amount_col in df.columns:
        validity_violations += int((df[amount_col] < 0).sum())
        validity_checks += len(df)
 
    if validity_checks > 0:
        scores['validity'] = max(0.0, 1 - (validity_violations / validity_checks))
    else:
        scores['validity'] = 1.0
 
    # --- UNIQUENESS: fraction of non-duplicate rows ---
    if id_col and id_col in df.columns:
        n_dupes = df.duplicated(subset=[id_col]).sum()
    else:
        n_dupes = df.duplicated().sum()
    scores['uniqueness'] = float(1 - (n_dupes / len(df)))
 
    # --- CONSISTENCY: fraction of rows with valid status values ---
    if status_col and status_col in df.columns and valid_statuses:
        normalized = df[status_col].str.lower().str.strip()
        bad = ~normalized.isin(valid_statuses)
        scores['consistency'] = float(1 - (bad.sum() / len(df)))
    else:
        scores['consistency'] = 1.0
 
    return scores
 
 
def compute_overall_score(dimension_scores: dict, weights: dict) -> float:
    """Weighted average of dimension scores."""
    total = sum(dimension_scores.get(dim, 1.0) * w for dim, w in weights.items())
    return round(total, 4)
 
 
# --- Demo ---
df = pd.DataFrame({
    'order_id':   [1001, 1002, 1003, 1002, 1004, 1005, 1006, 1007, 1008, 1009],
    'customer_id': [101, 102, None, 104, 105, None, 107, 108, 109, 110],
    'amount':     [49.99, -5.00, 120.0, 89.50, 0.0, 250.0, 33.0, 88.5, 200.0, 12.5],
    'status':     ['shipped', 'SHIPPED', 'pending', 'delivered', 'shipped',
                   'Pending', 'complete', 'shipped', 'pending', 'delivered'],
    'order_date': ['2024-01-01', '2024-01-02', '2024-01-03', '2099-12-01', '2024-01-05',
                   '2024-01-06', '2024-01-07', 'not-a-date', '2024-01-09', '2024-01-10'],
})
 
scores = compute_dimension_scores(
    df,
    id_col='order_id',
    amount_col='amount',
    date_col='order_date',
    status_col='status',
    valid_statuses={'shipped', 'pending', 'delivered', 'cancelled'},
)
 
overall = compute_overall_score(scores, WEIGHTS)
 
print("Dimension scores:")
for dim, score in scores.items():
    print(f"  {dim:<14}: {score:.4f}  (weight: {WEIGHTS[dim]:.0%})")
print(f"\nOverall score: {overall:.4f}")

import pandas as pd
 
def format_quality_report(
    results: dict,
    dataset_name: str = "dataset",
    n_rows: int = 0,
    n_cols: int = 0,
) -> str:
    """
    Format quality results as a structured text table.
 
    results must have keys:
      'overall'    — float score 0.0-1.0
      'status'     — 'PASS' | 'WARN' | 'FAIL'
      'dimensions' — dict of {dim_name: {'score': float, 'status': str}}
      'blocking'   — bool
 
    Returns a multi-line string suitable for printing or logging.
    """
    lines = []
 
    # ─── Header ──────────────────────────────────────────────────────────────
    lines.append("┌─────────────────────────────────────────┐")
    lines.append("│  DATA QUALITY REPORT                    │")
    lines.append(f"│  Dataset: {dataset_name:<30} │")
    lines.append(f"│  Rows: {n_rows:>6,}  |  Columns: {n_cols:<3}          │")
    lines.append("├──────────────────────┬─────────┬────────┤")
    lines.append("│ Dimension            │  Score  │ Status │")
    lines.append("├──────────────────────┼─────────┼────────┤")
 
    # ─── Dimension rows ───────────────────────────────────────────────────────
    for dim_name, dim_data in results['dimensions'].items():
        score  = dim_data['score']
        status = dim_data['status']
 
        # Color indicators using ASCII (works in any terminal)
        status_display = {
            'PASS': ' PASS ',
            'WARN': ' WARN ',
            'FAIL': ' FAIL ',
        }.get(status, status)
 
        dim_display = dim_name.capitalize()
        lines.append(
            f"│ {dim_display:<20} │  {score:.4f} │{status_display}│"
        )
 
    # ─── Overall row ──────────────────────────────────────────────────────────
    lines.append("├──────────────────────┼─────────┼────────┤")
    overall_status = results['status']
    overall_score  = results['overall']
    status_display = {
        'PASS': ' PASS ',
        'WARN': ' WARN ',
        'FAIL': ' FAIL ',
    }.get(overall_status, overall_status)
 
    lines.append(
        f"│ {'Overall Score':<20} │  {overall_score:.4f} │{status_display}│"
    )
    lines.append("└──────────────────────┴─────────┴────────┘")
 
    # ─── Blocking flag ────────────────────────────────────────────────────────
    if results['blocking']:
        lines.append("")
        lines.append("  *** PIPELINE BLOCKED — one or more dimensions FAILED ***")
        lines.append("  Action: investigate FAIL dimensions before proceeding.")
 
    return "\n".join(lines)
 
 
# --- Demo ---
sample_results = {
    'overall': 0.90,
    'status': 'WARN',
    'blocking': False,
    'dimensions': {
        'completeness': {'score': 0.97, 'status': 'PASS'},
        'validity':     {'score': 0.89, 'status': 'WARN'},
        'uniqueness':   {'score': 1.00, 'status': 'PASS'},
        'consistency':  {'score': 0.72, 'status': 'FAIL'},
    }
}
 
report = format_quality_report(
    sample_results,
    dataset_name="transactions.csv",
    n_rows=1000,
    n_cols=8,
)
print(report)
 
# Output:
# ┌─────────────────────────────────────────┐
# │  DATA QUALITY REPORT                    │
# │  Dataset: transactions.csv              │
# │  Rows:  1,000  |  Columns: 8           │
# ├──────────────────────┬─────────┬────────┤
# │ Dimension            │  Score  │ Status │
# ├──────────────────────┼─────────┼────────┤
# │ Completeness         │  0.9700 │ PASS  │
# │ Validity             │  0.8900 │ WARN  │
# │ Uniqueness           │  1.0000 │ PASS  │
# │ Consistency          │  0.7200 │ FAIL  │
# ├──────────────────────┼─────────┼────────┤
# │ Overall Score        │  0.9000 │ WARN  │
# └──────────────────────┴─────────┴────────┘

import pandas as pd
 
# Configurable thresholds
THRESHOLDS = {
    'pass': 0.95,   # score >= 0.95 → PASS
    'warn': 0.80,   # score >= 0.80 → WARN, else FAIL
}
 
WEIGHTS = {
    'completeness': 0.30,
    'validity':     0.30,
    'uniqueness':   0.20,
    'consistency':  0.20,
}
 
 
def score_to_status(score: float, thresholds: dict) -> str:
    """Convert a 0.0-1.0 score to PASS / WARN / FAIL."""
    if score >= thresholds['pass']:
        return 'PASS'
    elif score >= thresholds['warn']:
        return 'WARN'
    else:
        return 'FAIL'
 
 
def run_quality_check(
    df: pd.DataFrame,
    thresholds: dict,
    weights: dict = None,
    id_col: str = None,
    amount_col: str = None,
    date_col: str = None,
    status_col: str = None,
    valid_statuses: set = None,
) -> dict:
    """
    Run a full quality check on df.
 
    Returns:
      {
        'overall':    float,          # weighted average score
        'status':     str,            # 'PASS' | 'WARN' | 'FAIL'
        'dimensions': {               # per-dimension results
            dim_name: {
                'score':  float,
                'status': str,
                'weight': float,
            }
        },
        'blocking':   bool,           # True if any dimension is FAIL
        'metadata': {
            'n_rows': int,
            'n_cols': int,
        }
      }
    """
    if weights is None:
        weights = WEIGHTS
 
    # --- Compute dimension scores ---
    dimension_scores = {}
 
    # Completeness
    dimension_scores['completeness'] = float(1 - df.isnull().mean().mean())
 
    # Validity
    validity_violations = 0
    validity_checks = 0
    if date_col and date_col in df.columns:
        dates = pd.to_datetime(df[date_col], errors='coerce')
        validity_violations += int(dates.isna().sum())
        validity_violations += int((dates > pd.Timestamp.today()).sum())
        validity_checks += len(df)
    if amount_col and amount_col in df.columns:
        validity_violations += int((df[amount_col] < 0).sum())
        validity_checks += len(df)
    dimension_scores['validity'] = (
        max(0.0, 1 - validity_violations / validity_checks)
        if validity_checks > 0 else 1.0
    )
 
    # Uniqueness
    if id_col and id_col in df.columns:
        n_dupes = df.duplicated(subset=[id_col]).sum()
    else:
        n_dupes = df.duplicated().sum()
    dimension_scores['uniqueness'] = float(1 - (n_dupes / len(df)))
 
    # Consistency
    if status_col and status_col in df.columns and valid_statuses:
        normalized = df[status_col].str.lower().str.strip()
        bad = ~normalized.isin(valid_statuses)
        dimension_scores['consistency'] = float(1 - (bad.sum() / len(df)))
    else:
        dimension_scores['consistency'] = 1.0
 
    # --- Apply thresholds and weights ---
    dimensions = {}
    for dim, score in dimension_scores.items():
        w = weights.get(dim, 0)
        dimensions[dim] = {
            'score':  round(score, 4),
            'status': score_to_status(score, thresholds),
            'weight': w,
        }
 
    # --- Overall score (weighted average) ---
    overall = sum(
        dimensions[dim]['score'] * dimensions[dim]['weight']
        for dim in dimensions
    )
    overall_status = score_to_status(overall, thresholds)
 
    # --- Blocking: True if any dimension is FAIL ---
    blocking = any(d['status'] == 'FAIL' for d in dimensions.values())
 
    return {
        'overall':    round(overall, 4),
        'status':     overall_status,
        'dimensions': dimensions,
        'blocking':   blocking,
        'metadata': {
            'n_rows': len(df),
            'n_cols': len(df.columns),
        },
    }
 
 
# --- Demo ---
df = pd.DataFrame({
    'order_id':   list(range(1001, 1021)),
    'customer_id': [i if i % 5 != 0 else None for i in range(20)],
    'amount':     [round(50 + i * 10, 2) for i in range(20)],
    'status':     ['shipped', 'PENDING', 'delivered', 'cancelled', 'SHIPPED'] * 4,
    'order_date': ['2024-01-' + str(i + 1).zfill(2) for i in range(19)] + ['2099-12-31'],
})
 
result = run_quality_check(
    df,
    thresholds=THRESHOLDS,
    id_col='order_id',
    amount_col='amount',
    date_col='order_date',
    status_col='status',
    valid_statuses={'shipped', 'pending', 'delivered', 'cancelled'},
)
 
print(f"Overall: {result['overall']:.4f} [{result['status']}]")
print(f"Blocking: {result['blocking']}")
for dim, data in result['dimensions'].items():
    print(f"  {dim:<14}: {data['score']:.4f}  [{data['status']}]")