—— coding: utf-8 ——

"““数据质量监控器：自动检测数据的完整性、唯一性和取值范围。””" import pandas as pd class DataQualityMonitor: def init(self, df, config): self.df = df self.config = config # {"rule_type": {"columns": [……], "threshold“: 0.9}} self.report_list = [] # 每一条质量检查的结果 self.total_rules = len(config) def check_completeness(self, cols): ”““完整性规则：字段非空比例是否达标。””" threshold = self.config.get("completeness", {}).get("threshold", 0.9) for col in cols: non_null_ratio = self.df[col].notna().mean() self.report_list.append({ "rule": "completeness", "column": col, "score": round(non_null_ratio * 100, 2), "passed": non_null_ratio >= threshold }) def check_uniqueness(self, cols): "““唯一性规则：主键是否出现重复。””" threshold = self.config.get("uniqueness", {}).get("threshold", 0.0) for col in cols: dup_ratio = self.df[col].duplicated().mean() self.report_list.append({ "rule": "uniqueness", "column": col, "score": round((1 - dup_ratio) * 100, 2), "passed": dup_ratio <= threshold }) def check_value_range(self, cols): "““准确性规则：字段取值是否在业务允许范围内。””" min_val, max_val = self.config.get("value_range", {}).get("range", [0, 100]) threshold = self.config.get("value_range", {}).get("threshold", 0.95) for col in cols: in_range_ratio = self.df[col].between(min_val, max_val).mean() self.report_list.append({ "rule": "value_range", "column": col, "score": round(in_range_ratio * 100, 2), "passed": in_range_ratio >= threshold }) def run_all_checks(self): "““按配置执行全部质量规则。””" for rule_name in self.config: if rule_name == "completeness": self.check_completeness(self.config[rule_name]["columns"]) elif rule_name == "uniqueness": self.check_uniqueness(self.config[rule_name]["columns"]) elif rule_name == "value_range": self.check_value_range(self.config[rule_name]["columns"]) return self.report_list def generate_report(self): "““生成质量汇总报告。””“ if not self.report_list: return “尚未运行任何质量检查，请先调用 run_all_checks()。” passed_count = sum(1 for r in self.report_list if r[”passed“]) total_count = len(self.report_list) overall_rate = passed_count / total_count * 100 lines = [“========== 数据质量检测报告 ==========”] for r in self.report_list: mark = ”[PASS]" if r["passed"] else "[FAIL]“ lines.append(f“{mark} {r['rule']:>12} | {r['column']:>8} | 得分 {r['score']}%”) lines.append(”--------------------------------------“) lines.append(f“整体通过率：{overall_rate:.1f}% ({passed_count}/{total_count})”) return ”\n".join(lines) if name == "main": demo_df = pd.DataFrame({ "user_id": [1001, 1001, 1002, 1003, 1004], "age": [25, 30, 25, None, 150], "city“: [“北京”, “上海”, “北京”, “深圳”, “北京”] }) demo_config = { ”completeness": {"columns": ["age", "city"], "threshold": 0.9}, "uniqueness": {"columns": ["user_id"], "threshold": 0.0}, "value_range": {"columns": ["age"], "range": [0, 120], "threshold": 0.95} } monitor = DataQualityMonitor(demo_df, demo_config) monitor.run_all_checks() print(monitor.generate_report()) ```