kumar6591/data-quality-env

1from tasks.base import BaseTask2from env.models import AuditReport3 4 5class Task1(BaseTask):6    def get_description(self) -> str:7        return (8            "Audit the 'customers' table. Find: (1) real NULL values in each column, "9            "(2) disguised nulls stored as strings like 'NULL','N/A','-' etc., "10            "(3) exact duplicate rows, and (4) near-duplicate rows (same record, 1-2 fields changed). "11            "Report counts per finding with your confidence (0.0-1.0) in each."12        )13 14    def get_table_names(self) -> list[str]:15        return ["customers"]16 17    def grade(self, report: AuditReport, gold: dict) -> tuple[float, dict]:18        scores: dict[str, float] = {}19        if "email" in report.null_issues:20            fc = report.null_issues["email"]21            acc = self.count_accuracy(int(fc.value), int(gold["null_email_total"]))22            scores["null_email"] = self.brier_adjust(acc, fc.confidence, acc > 0.6)23        else:24            scores["null_email"] = 0.025 26        if "customer_id" in report.null_issues:27            fc = report.null_issues["customer_id"]28            acc = self.count_accuracy(int(fc.value), int(gold["null_customer_id"]))29            scores["null_cid"] = self.brier_adjust(acc, fc.confidence, acc > 0.6)30        else:31            scores["null_cid"] = 0.032 33        fc_dup = report.duplicate_row_count34        dup_acc = self.count_accuracy(int(fc_dup.value), int(gold["exact_duplicate_rows"]))35        scores["exact_dups"] = self.brier_adjust(dup_acc, fc_dup.confidence, dup_acc > 0.6)36 37        near_detected = any("near" in str(v.get("issue_type", "")).lower() for v in report.schema_violations)38        scores["near_dups"] = 0.5 if near_detected else 0.039 40        scores = {k: self.strict_score(v) for k, v in scores.items()}41 42        weights = {"null_email": 0.30, "null_cid": 0.25, "exact_dups": 0.30, "near_dups": 0.15}43        total = sum(scores[k] * weights[k] for k in weights)44        return self.strict_score(round(total, 4)), scores45