数据质量与校验

# Python 3.10+
import pandas as pd
import numpy as np

# 模拟一份数据
df = pd.DataFrame({
    "user_id": [1, 2, 3, 4, 5],
    "name": ["Alice", "Bob", None, "Diana", "Eve"],
    "age": [25, np.nan, 30, -1, 28],
    "email": ["a@x.com", "b@x.com", "c@x.com", None, "e@x.com"],
})

# 检查每列的空值率
null_ratio = df.isnull().mean()
print("=== 各列空值率 ===")
print(null_ratio)
# user_id    0.0
# name       0.2
# age        0.2
# email      0.2

# 检查必需字段是否完整
required_columns = ["user_id", "name", "email"]
for col in required_columns:
    null_count = df[col].isnull().sum()
    if null_count > 0:
        print(f"[完整性告警] 必需字段 '{col}' 存在 {null_count} 个空值")

# Python 3.10+
import re

# 值域范围校验
age_invalid = df[(df["age"] < 0) | (df["age"] > 150)]
if not age_invalid.empty:
    print(f"[准确性告警] 发现 {len(age_invalid)} 条年龄异常记录:")
    print(age_invalid[["user_id", "age"]])

# 格式规则校验
email_pattern = r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
invalid_emails = df[
    df["email"].notna() & ~df["email"].str.match(email_pattern)
]
if not invalid_emails.empty:
    print(f"[准确性告警] 发现 {len(invalid_emails)} 条邮箱格式异常记录")

# 业务逻辑校验：订单金额应 >= 0
orders = pd.DataFrame({
    "order_id": [101, 102, 103],
    "amount": [99.9, -5.0, 200.0],
})
negative_orders = orders[orders["amount"] < 0]
if not negative_orders.empty:
    print(f"[准确性告警] 发现 {len(negative_orders)} 条负金额订单")

# Python 3.10+
# 订单汇总表
order_summary = pd.DataFrame({
    "order_id": [1, 2, 3],
    "total_amount": [100.0, 200.0, 300.0],
})

# 订单明细表
order_details = pd.DataFrame({
    "order_id": [1, 1, 2, 2, 3, 3],
    "item_amount": [60.0, 40.0, 150.0, 50.0, 300.0, 10.0],  # order_id=3 不一致
})

# 跨表一致性校验
detail_sum = order_details.groupby("order_id")["item_amount"].sum().reset_index()
detail_sum.columns = ["order_id", "detail_total"]

merged = order_summary.merge(detail_sum, on="order_id")
inconsistent = merged[~np.isclose(merged["total_amount"], merged["detail_total"])]
if not inconsistent.empty:
    print("[一致性告警] 汇总金额与明细不一致:")
    print(inconsistent)

# Python 3.10+
from datetime import datetime, timedelta

# 检查数据是否在 SLA 时间窗口内到达
data_loaded_at = datetime(2026, 6, 6, 9, 30, 0)  # 实际加载完成时间
sla_deadline = datetime(2026, 6, 6, 8, 0, 0)      # SLA 要求时间

if data_loaded_at > sla_deadline:
    delay = data_loaded_at - sla_deadline
    print(f"[时效性告警] 数据加载延迟 {delay}，超出 SLA 要求")

# 检查数据时间戳是否合理（不应是未来时间）
df_with_ts = pd.DataFrame({
    "event_id": [1, 2, 3],
    "event_time": [
        datetime(2026, 6, 5, 10, 0),
        datetime(2026, 6, 6, 14, 0),
        datetime(2026, 12, 31, 0, 0),  # 未来时间，异常
    ],
})

now = datetime.now()
future_events = df_with_ts[df_with_ts["event_time"] > now]
if not future_events.empty:
    print(f"[时效性告警] 发现 {len(future_events)} 条未来时间记录")

# Python 3.10+
# 主键唯一性检查
users = pd.DataFrame({
    "user_id": [1, 2, 2, 3, 4],  # user_id=2 重复
    "name": ["Alice", "Bob", "Bob2", "Carol", "Dave"],
})

duplicates = users[users.duplicated(subset=["user_id"], keep=False)]
if not duplicates.empty:
    print(f"[唯一性告警] 发现 {len(duplicates)} 条主键重复记录:")
    print(duplicates)

# 业务去重：同一用户同一日内应只有一条记录
daily_events = pd.DataFrame({
    "user_id": [1, 1, 2, 2, 3],
    "event_date": ["2026-06-05", "2026-06-05", "2026-06-05", "2026-06-06", "2026-06-05"],
    "event_count": [3, 5, 1, 2, 4],
})

dup_business = daily_events[daily_events.duplicated(subset=["user_id", "event_date"], keep=False)]
if not dup_business.empty:
    print(f"[唯一性告警] 发现 {len(dup_business)} 条业务重复记录")

# 安装 Great Expectations
pip install great_expectations

# 初始化项目目录（会创建 gx/ 目录）
great_expectations init

# Python 3.10+
# 文件名: gx_expectation_suite.py
"""定义用户表的 Expectation Suite"""
import great_expectations as gx
from great_expectations.core import ExpectationSuite

# 获取或创建 Expectation Suite
context = gx.get_context()

suite = context.suites.add(ExpectationSuite(name="user_table_suite"))

# 添加期望规则

# 1. 完整性：user_id 列不能有空值
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
)

# 2. 唯一性：user_id 列值唯一
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="user_id")
)

# 3. 准确性：age 列值在 0-150 之间
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="age", min_value=0, max_value=150
    )
)

# 4. 格式校验：email 列匹配邮箱正则
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email", regex=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
    )
)

# 5. 行数检查：表至少有 1 条记录
suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeGreaterThan(min_value=0)
)

# 6. 列存在性检查
suite.add_expectation(
    gx.expectations.ExpectTableColumnsToMatchSet(
        column_set=["user_id", "name", "age", "email"],
        exact_match=True,
    )
)

# 7. 空值率阈值：name 列空值率不超过 5%
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(
        column="name", mostly=0.95
    )
)

print(f"Expectation Suite '{suite.name}' 已创建，共 {len(suite.expectations)} 条规则")

# Python 3.10+
# 文件名: gx_checkpoint.py
"""使用 Checkpoint 执行数据校验并生成报告"""
import great_expectations as gx
import pandas as pd

# 创建上下文
context = gx.get_context()

# 准备测试数据
df = pd.DataFrame({
    "user_id": [1, 2, 3, 4, 5],
    "name": ["Alice", "Bob", None, "Diana", "Eve"],
    "age": [25, -1, 30, 200, 28],  # 包含异常值
    "email": ["a@x.com", "invalid", "c@x.com", "d@x.com", "e@x.com"],
})

# 创建数据源和数据资产
data_source = context.data_sources.add_pandas(name="pandas_source")
data_asset = data_source.add_dataframe_asset(name="user_asset")
batch_definition = data_asset.add_batch_definition_whole_dataframe("user_batch")

# 获取之前定义的 Expectation Suite
suite = context.suites.get("user_table_suite")

# 创建验证定义
validation = gx.ValidationDefinition(
    name="user_validation",
    data=batch_definition,
    suite=suite,
)
validation = context.validation_definitions.add(validation)

# 执行校验
result = validation.run(batch_parameters={"dataframe": df})

# 查看结果
print(f"校验状态: {'通过' if result.success else '失败'}")
print(f"总规则数: {result.results.__len__()}")

for idx, vr in enumerate(result.results, 1):
    status = "PASS" if vr.success else "FAIL"
    expectation_type = vr.expectation_config.type
    print(f"  规则 {idx}: [{status}] {expectation_type}")

# Python 3.10+
# 文件名: gx_data_docs.py
"""配置和生成 Data Docs 数据质量报告"""
import great_expectations as gx

context = gx.get_context()

# 配置 Data Docs 站点（本地 HTML）
context.data_docs.add_site(
    name="local_site",
    class_name="SiteBuilder",
    store_backend={
        "class_name": "FilesystemStoreBackend",
        "base_directory": "uncommitted/data_docs/local_site/",
    },
)

# 构建文档
context.build_data_docs()

print("Data Docs 已生成，请查看 gx/uncommitted/data_docs/local_site/index.html")

pip install pandera

# Python 3.10+
# 文件名: pandera_basic.py
"""Pandera Schema 基础用法"""
import pandera as pa
from pandera import Column, Check, Index
import pandas as pd

# 定义 Schema
user_schema = pa.DataFrameSchema({
    "user_id": Column(
        int,
        checks=[
            Check.ge(1, error="user_id 必须为正整数"),  # 大于等于 1
            Check(lambda s: s.is_unique, error="user_id 必须唯一"),
        ],
        nullable=False,
        description="用户唯一标识",
    ),
    "name": Column(
        str,
        checks=Check.str_length(min_value=1, error="姓名不能为空字符串"),
        nullable=True,  # 允许空值
        description="用户姓名",
    ),
    "age": Column(
        int,
        checks=Check.in_range(0, 150, error="年龄必须在 0-150 之间"),
        nullable=True,
        description="用户年龄",
    ),
    "email": Column(
        str,
        checks=Check.str_matches(
            r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$",
            error="邮箱格式不合法",
        ),
        nullable=True,
        description="用户邮箱",
    ),
})

# 合法数据
valid_df = pd.DataFrame({
    "user_id": [1, 2, 3],
    "name": ["Alice", "Bob", "Carol"],
    "age": [25, 30, 28],
    "email": ["a@x.com", "b@y.com", "c@z.com"],
})

validated = user_schema.validate(valid_df)
print("校验通过!")

# 非法数据会抛出 SchemaError
invalid_df = pd.DataFrame({
    "user_id": [1, 1],       # 重复
    "name": ["", "Bob"],     # 空字符串
    "age": [25, -5],         # 负数
    "email": ["a@x.com", "invalid"],  # 格式错误
})

try:
    user_schema.validate(invalid_df)
except pa.errors.SchemaErrors as err:
    print("=== 校验失败详情 ===")
    for error in err.schema_errors:
        print(f"  - {error}")

# Python 3.10+
# 文件名: pandera_advanced.py
"""Pandera 高级用法：多列校验与自定义校验器"""
import pandera as pa
from pandera import Column, Check, DataFrameSchema
import pandas as pd

# 自定义校验器：订单金额应大于 0
def check_positive_amount(df: pd.DataFrame) -> pd.Series:
    """检查 amount 列是否全为正数"""
    return df["amount"] > 0

# 多列联合校验：结束时间应晚于开始时间
def check_time_order(df: pd.DataFrame) -> pd.Series:
    """检查 end_time 是否晚于 start_time"""
    return df["end_time"] > df["start_time"]

order_schema = DataFrameSchema(
    columns={
        "order_id": Column(int, checks=Check.ge(1), nullable=False),
        "amount": Column(float, nullable=False),
        "start_time": Column("datetime64[ns]", nullable=False),
        "end_time": Column("datetime64[ns]", nullable=False),
        "status": Column(
            str,
            checks=Check.isin(["pending", "paid", "shipped", "cancelled"]),
            nullable=False,
        ),
    },
    # DataFrame 级别的校验
    checks=[
        Check(check_positive_amount, error="订单金额必须为正数"),
        Check(check_time_order, error="结束时间必须晚于开始时间"),
    ],
    # 严格模式：不允许出现 Schema 中未定义的列
    strict=True,
    # 强制类型转换
    coerce=True,
)

# 测试
valid_orders = pd.DataFrame({
    "order_id": [1, 2, 3],
    "amount": [99.9, 200.0, 50.5],
    "start_time": pd.to_datetime(["2026-06-05 10:00", "2026-06-05 11:00", "2026-06-05 12:00"]),
    "end_time": pd.to_datetime(["2026-06-05 11:00", "2026-06-05 12:00", "2026-06-05 13:00"]),
    "status": ["paid", "pending", "shipped"],
})

validated = order_schema.validate(valid_orders)
print("订单数据校验通过!")

# Python 3.10+
# 文件名: pandera_class_schema.py
"""使用 Class Schema 定义更清晰的校验规则"""
import pandera as pa
from pandera import Column, Check
import pandas as pd
from typing import Optional

class ProductSchema(pa.SchemaModel):
    """产品数据 Schema"""
    product_id: pa.typing.Series[int] = pa.Field(
        ge=1, description="产品唯一标识"
    )
    name: pa.typing.Series[str] = pa.Field(
        str_length_min=1, description="产品名称"
    )
    price: pa.typing.Series[float] = pa.Field(
        ge=0, le=100000, description="产品价格（0-100000）"
    )
    category: pa.typing.Series[str] = pa.Field(
        isin=["电子", "服装", "食品", "家居"], description="产品类别"
    )
    stock: pa.typing.Series[int] = pa.Field(
        ge=0, description="库存数量"
    )

    class Config:
        strict = True
        coerce = True

# 使用
products = pd.DataFrame({
    "product_id": [1, 2, 3],
    "name": ["手机", "T恤", "零食"],
    "price": [4999.0, 99.0, 15.5],
    "category": ["电子", "服装", "食品"],
    "stock": [100, 500, 2000],
})

validated = ProductSchema.validate(products)
print("产品数据校验通过!")

# Python 3.10+
# 文件名: lineage_tracking.py
"""简单的数据血缘追踪实现"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import hashlib
import json


@dataclass
class LineageNode:
    """血缘节点：表示一个数据源或数据产物"""
    name: str
    node_type: str  # "source", "transform", "target"
    schema_info: dict[str, str] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class LineageEdge:
    """血缘边：表示数据流转关系"""
    source: str
    target: str
    transform: str = ""
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())


class LineageTracker:
    """简单的血缘追踪器"""

    def __init__(self) -> None:
        self._nodes: dict[str, LineageNode] = {}
        self._edges: list[LineageEdge] = []

    def add_node(self, node: LineageNode) -> None:
        """注册血缘节点"""
        self._nodes[node.name] = node

    def add_edge(self, edge: LineageEdge) -> None:
        """注册血缘边"""
        self._edges.append(edge)

    def get_upstream(self, node_name: str) -> list[str]:
        """获取指定节点的所有上游"""
        return [e.source for e in self._edges if e.target == node_name]

    def get_downstream(self, node_name: str) -> list[str]:
        """获取指定节点的所有下游"""
        return [e.target for e in self._edges if e.source == node_name]

    def get_full_lineage(self, node_name: str, direction: str = "upstream") -> list[str]:
        """递归获取完整血缘链"""
        visited: set[str] = set()
        result: list[str] = []

        def _traverse(name: str) -> None:
            if name in visited:
                return
            visited.add(name)
            if direction == "upstream":
                parents = self.get_upstream(name)
            else:
                parents = self.get_downstream(name)
            for parent in parents:
                _traverse(parent)
                result.append(parent)

        _traverse(node_name)
        return result

    def to_dict(self) -> dict[str, Any]:
        """导出为字典"""
        return {
            "nodes": [
                {"name": n.name, "type": n.node_type, "schema": n.schema_info}
                for n in self._nodes.values()
            ],
            "edges": [
                {"source": e.source, "target": e.target, "transform": e.transform}
                for e in self._edges
            ],
        }


# 使用示例
tracker = LineageTracker()

# 注册数据源
tracker.add_node(LineageNode("api_orders", "source", {"columns": ["order_id", "amount", "user_id"]}))
tracker.add_node(LineageNode("api_users", "source", {"columns": ["user_id", "name", "region"]}))
tracker.add_node(LineageNode("raw_orders", "transform", {"columns": ["order_id", "amount", "user_id"]}))
tracker.add_node(LineageNode("raw_users", "transform", {"columns": ["user_id", "name", "region"]}))
tracker.add_node(LineageNode("dws_order_user", "transform", {"columns": ["order_id", "amount", "user_name", "region"]}))
tracker.add_node(LineageNode("report_daily_sales", "target", {"columns": ["date", "region", "total_amount"]}))

# 注册血缘关系
tracker.add_edge(LineageEdge("api_orders", "raw_orders", "API 采集"))
tracker.add_edge(LineageEdge("api_users", "raw_users", "API 采集"))
tracker.add_edge(LineageEdge("raw_orders", "dws_order_user", "JOIN + 清洗"))
tracker.add_edge(LineageEdge("raw_users", "dws_order_user", "JOIN + 清洗"))
tracker.add_edge(LineageEdge("dws_order_user", "report_daily_sales", "聚合统计"))

# 查询血缘
print("report_daily_sales 的上游:", tracker.get_upstream("report_daily_sales"))
print("完整上游链:", tracker.get_full_lineage("report_daily_sales"))
print("api_orders 的下游:", tracker.get_downstream("api_orders"))

# Python 3.10+
# 文件名: etl_with_pandera.py
"""使用 Pandera 在 ETL 管道的每个阶段进行数据校验"""
import pandera as pa
from pandera import Column, Check, DataFrameSchema
import pandas as pd
from datetime import datetime


# ========== Schema 定义 ==========

# 输入层：原始订单数据的 Schema
raw_order_schema = DataFrameSchema(
    columns={
        "order_id": Column(int, Check.ge(1), nullable=False),
        "user_id": Column(int, Check.ge(1), nullable=False),
        "product_id": Column(int, Check.ge(1), nullable=False),
        "quantity": Column(int, Check.ge(1), nullable=False),
        "unit_price": Column(float, Check.ge(0), nullable=False),
        "order_time": Column("datetime64[ns]", nullable=False),
    },
    strict=True,
    coerce=True,
)

# 输出层：聚合后销售报表的 Schema
sales_report_schema = DataFrameSchema(
    columns={
        "date": Column(str, Check.str_matches(r"^\d{4}-\d{2}-\d{2}$"), nullable=False),
        "product_id": Column(int, Check.ge(1), nullable=False),
        "total_quantity": Column(int, Check.ge(0), nullable=False),
        "total_revenue": Column(float, Check.ge(0), nullable=False),
        "avg_unit_price": Column(float, Check.ge(0), nullable=False),
        "order_count": Column(int, Check.ge(0), nullable=False),
    },
    strict=True,
    coerce=True,
)


# ========== ETL 管道 ==========

def extract() -> pd.DataFrame:
    """提取：从数据源获取原始数据"""
    return pd.DataFrame({
        "order_id": [1, 2, 3, 4, 5],
        "user_id": [101, 102, 101, 103, 102],
        "product_id": [1, 2, 1, 3, 2],
        "quantity": [2, 1, 3, 1, 5],
        "unit_price": [49.9, 199.0, 49.9, 89.5, 199.0],
        "order_time": pd.to_datetime([
            "2026-06-05 10:00", "2026-06-05 11:00",
            "2026-06-05 14:00", "2026-06-05 16:00",
            "2026-06-05 17:00",
        ]),
    })


def transform(df: pd.DataFrame) -> pd.DataFrame:
    """转换：按日期和产品聚合"""
    df = df.copy()
    df["date"] = df["order_time"].dt.strftime("%Y-%m-%d")
    df["revenue"] = df["quantity"] * df["unit_price"]

    report = df.groupby(["date", "product_id"]).agg(
        total_quantity=("quantity", "sum"),
        total_revenue=("revenue", "sum"),
        avg_unit_price=("unit_price", "mean"),
        order_count=("order_id", "count"),
    ).reset_index()

    return report


def load(df: pd.DataFrame) -> None:
    """加载：写入目标存储"""
    print(f"写入 {len(df)} 条聚合记录到数据仓库")
    print(df.to_string())


def run_etl() -> None:
    """运行 ETL 管道，每个阶段都有校验"""
    # 1. 提取并校验输入
    raw = extract()
    print("[ETL] 提取原始数据，执行输入校验...")
    try:
        raw = raw_order_schema.validate(raw)
        print(f"[ETL] 输入校验通过，共 {len(raw)} 条记录")
    except pa.errors.SchemaErrors as err:
        print(f"[ETL] 输入数据校验失败: {err}")
        raise

    # 2. 转换
    print("[ETL] 执行数据转换...")
    transformed = transform(raw)

    # 3. 校验输出
    print("[ETL] 执行输出校验...")
    try:
        transformed = sales_report_schema.validate(transformed)
        print(f"[ETL] 输出校验通过，共 {len(transformed)} 条记录")
    except pa.errors.SchemaErrors as err:
        print(f"[ETL] 输出数据校验失败: {err}")
        raise

    # 4. 加载
    load(transformed)
    print("[ETL] 管道执行完成")


if __name__ == "__main__":
    run_etl()

# Python 3.10+
# 文件名: gx_quality_report.py
"""使用 Great Expectations 构建完整的数据质量报告"""
import great_expectations as gx
from great_expectations.core import ExpectationSuite
import pandas as pd
import numpy as np


def create_quality_report() -> None:
    """构建数据质量报告的完整流程"""

    # ========== 1. 准备数据 ==========
    np.random.seed(42)
    df = pd.DataFrame({
        "transaction_id": range(1, 1001),
        "user_id": np.random.randint(1, 200, 1000),
        "amount": np.random.exponential(100, 1000).round(2),
        "status": np.random.choice(
            ["completed", "pending", "cancelled", "refunded"],
            1000,
            p=[0.7, 0.15, 0.1, 0.05],
        ),
        "created_at": pd.date_range("2026-06-01", periods=1000, freq="h"),
    })

    # 故意注入一些数据质量问题
    df.loc[10:15, "amount"] = -1 * df.loc[10:15, "amount"]  # 负金额
    df.loc[50:55, "user_id"] = None  # 空值
    df.loc[100, "transaction_id"] = df.loc[99, "transaction_id"]  # 重复 ID

    # ========== 2. 初始化 GX ==========
    context = gx.get_context()

    # ========== 3. 定义 Expectation Suite ==========
    suite = context.suites.add(ExpectationSuite(name="transaction_quality_suite"))

    # 唯一性
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeUnique(column="transaction_id")
    )

    # 完整性
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="transaction_id")
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id", mostly=0.99)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="amount")
    )

    # 准确性
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="amount", min_value=0, max_value=10000
        )
    )

    # 一致性
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="status",
            value_set=["completed", "pending", "cancelled", "refunded"],
        )
    )

    # 行数合理性
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeBetween(min_value=500, max_value=5000)
    )

    # ========== 4. 创建数据源并执行校验 ==========
    data_source = context.data_sources.add_pandas(name="tx_pandas_source")
    data_asset = data_source.add_dataframe_asset(name="tx_asset")
    batch_definition = data_asset.add_batch_definition_whole_dataframe("tx_batch")

    validation = gx.ValidationDefinition(
        name="tx_validation",
        data=batch_definition,
        suite=suite,
    )
    validation = context.validation_definitions.add(validation)

    # ========== 5. 运行校验 ==========
    result = validation.run(batch_parameters={"dataframe": df})

    # ========== 6. 输出报告摘要 ==========
    total = len(result.results)
    passed = sum(1 for r in result.results if r.success)
    failed = total - passed

    print("=" * 60)
    print("        数据质量报告摘要")
    print("=" * 60)
    print(f"  数据集: 交易记录表 (1000 行)")
    print(f"  校验时间: {pd.Timestamp.now().isoformat()}")
    print(f"  总规则数: {total}")
    print(f"  通过: {passed}")
    print(f"  失败: {failed}")
    print(f"  通过率: {passed / total * 100:.1f}%")
    print(f"  整体状态: {'PASS' if result.success else 'FAIL'}")
    print("=" * 60)

    # 失败规则详情
    if failed > 0:
        print("\n=== 失败规则详情 ===")
        for vr in result.results:
            if not vr.success:
                expectation_type = vr.expectation_config.type
                kwargs = vr.expectation_config.kwargs
                print(f"  [FAIL] {expectation_type}")
                print(f"         参数: {kwargs}")
                if hasattr(vr, "result") and vr.result:
                    unexpected = vr.result.get("unexpected_count", "N/A")
                    print(f"         异常数: {unexpected}")


if __name__ == "__main__":
    create_quality_report()