feat: add Chinese ID card validation expectation

great_expectations/expectations/__init__.py

@@ -32,6 +32,7 @@
     ExpectColumnValuesToBeNull,
     ExpectColumnValuesToBeOfType,
     ExpectColumnValuesToBeUnique,
+    ExpectColumnValuesToBeValidChineseIdCard,
     ExpectColumnValuesToMatchJsonSchema,
     ExpectColumnValuesToMatchLikePattern,
     ExpectColumnValuesToMatchLikePatternList,

great_expectations/expectations/core/__init__.py

@@ -58,6 +58,9 @@
 from .expect_column_values_to_be_null import ExpectColumnValuesToBeNull
 from .expect_column_values_to_be_of_type import ExpectColumnValuesToBeOfType
 from .expect_column_values_to_be_unique import ExpectColumnValuesToBeUnique
+from .expect_column_values_to_be_valid_chinese_id_card import (
+    ExpectColumnValuesToBeValidChineseIdCard,
+)
 from .expect_column_values_to_match_json_schema import (
     ExpectColumnValuesToMatchJsonSchema,
 )

great_expectations/expectations/core/expect_column_values_to_be_valid_chinese_id_card.py

@@ -0,0 +1,69 @@
+from great_expectations.expectations.expectation import ColumnMapExpectation
+from great_expectations.render import RenderedStringTemplateContent
+from great_expectations.render.util import num_to_str, substitute_none_for_missing
+
+
+class ExpectColumnValuesToBeValidChineseIdCard(ColumnMapExpectation):
+    """验证中国身份证号(15位/18位)"""
+
+    examples = [
+        {
+            "data": {
+                "valid_ids": ["110105199003077753", "110105900307775"],
+                "invalid_ids": ["123456", "110105199003077759", "abcdefg"],
+            },
+            "tests": [
+                {
+                    "title": "测试有效身份证",
+                    "exact_match_out": False,
+                    "in": {"column": "valid_ids"},
+                    "out": {"success": True},
+                },
+                {
+                    "title": "测试无效身份证",
+                    "exact_match_out": False,
+                    "in": {"column": "invalid_ids"},
+                    "out": {"success": False},
+                },
+            ],
+        }
+    ]
+
+    map_metric = "column_values.valid_chinese_id_card"
+    success_keys = ("mostly",)
+
+    def _prescriptive_renderer(
+        cls,
+        configuration=None,
+        runtime_configuration=None,
+        **kwargs,
+    ):
+        runtime_configuration = runtime_configuration or {}
+        include_column_name = runtime_configuration.get("include_column_name", True)
+        params = substitute_none_for_missing(
+            configuration.kwargs,
+            ["column", "mostly", "row_condition", "condition_parser"],
+        )
+
+        if params["mostly"] is not None and params["mostly"] < 1.0:
+            params["mostly_pct"] = num_to_str(
+                params["mostly"] * 100, precision=4, no_scientific=True
+            )
+            template_str = "值必须为有效的中国身份证号,至少 $mostly_pct % 的时间满足。"
+        else:
+            template_str = "值必须为有效的中国身份证号。"
+
+        if include_column_name:
+            template_str = "$column " + template_str
+
+        return [
+            RenderedStringTemplateContent(
+                **{
+                    "content_block_type": "string_template",
+                    "string_template": {
+                        "template": template_str,
+                        "params": params,
+                    },
+                }
+            )
+        ]

great_expectations/expectations/metrics/column_map_metrics/__init__.py

@@ -23,4 +23,5 @@
 from .column_values_null import ColumnValuesNull
 from .column_values_of_type import ColumnValuesOfType
 from .column_values_unique import ColumnValuesUnique
+from .column_values_valid_chinese_id_card import ColumnValuesValidChineseIdCard
 from .column_values_z_score import ColumnValuesZScore

great_expectations/expectations/metrics/column_map_metrics/column_values_valid_chinese_id_card.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime
+
+import pandas as pd
+
+from great_expectations.execution_engine import PandasExecutionEngine
+from great_expectations.expectations.metrics.map_metric_provider import (
+    ColumnMapMetricProvider,
+    column_condition_partial,
+)
+
+
+def validate_chinese_id_card(id_str: object) -> bool:
+    """Return True if *id_str* is a valid 15- or 18-digit Chinese resident ID number."""
+    if id_str is None:
+        return False
+    try:
+        if pd.isna(id_str):
+            return False
+    except TypeError:
+        pass
+    if not isinstance(id_str, str):
+        id_str = str(id_str).strip()
+
+    id_str = id_str.strip()
+
+    if len(id_str) not in (15, 18):
+        return False
+
+    if len(id_str) == 18:
+        if not re.match(r"^\d{17}[\dXx]$", id_str):
+            return False
+    elif not re.match(r"^\d{15}$", id_str):
+        return False
+
+    if len(id_str) == 18:
+        weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
+        check_codes = ["1", "0", "X", "9", "8", "7", "6", "5", "4", "3", "2"]
+        total = sum(int(id_str[i]) * weights[i] for i in range(17))
+        if id_str[-1].upper() != check_codes[total % 11]:
+            return False
+
+    try:
+        if len(id_str) == 18:
+            birth_str = id_str[6:14]
+        else:
+            birth_str = "19" + id_str[6:12]
+
+        if not re.match(r"^\d{8}$", birth_str):
+            return False
+
+        year = int(birth_str[:4])
+        month = int(birth_str[4:6])
+        day = int(birth_str[6:8])
+
+        current_year = datetime.now().year
+        if not (1900 <= year <= current_year) or not (1 <= month <= 12) or not (1 <= day <= 31):
+            return False
+    except Exception:
+        return False
+
+    return True
+
+
+class ColumnValuesValidChineseIdCard(ColumnMapMetricProvider):
+    condition_metric_name = "column_values.valid_chinese_id_card"
+    filter_column_isnull = True
+
+    @column_condition_partial(engine=PandasExecutionEngine)
+    def _pandas(cls, column, **kwargs):
+        return column.apply(validate_chinese_id_card)

pyproject.toml

@@ -663,7 +663,8 @@ filterwarnings = [
     # pyOpenSSL 25.0.1 starting raising this warning only for connecting to snowflake but no other datasources.
     "ignore: Attempting to mutate a Context after a Connection was created. In the future, this will raise an exception",
     "ignore: Benchmarks are automatically disabled because xdist plugin is active.Benchmarks cannot be performed reliably in a parallelized environment.:pytest_benchmark.logger.PytestBenchmarkWarning",
-    "ignore: No data was collected\\. \\(no-data-collected\\):coverage.exceptions.CoverageWarning",
+    # Use built-in Warning so pytest need not import optional `coverage` (minimal venvs / IDE runs).
+    "ignore: No data was collected\\. \\(no-data-collected\\):Warning",
 
     # --------------------------------------- TEMPORARY IGNORES --------------------------------------------------------
 ]

scripts/install_mssql_odbc_driver.sh

@@ -1 +1 @@
-install_sql_server_odbc_driver.sh
+install_sql_server_odbc_driver.sh

tests/expectations/test_expect_column_values_to_be_valid_chinese_id_card.py

@@ -0,0 +1,23 @@
+import re
+
+from great_expectations.expectations.expectation import ColumnMapExpectation
+
+
+class ExpectColumnValuesToBeValidChineseIdCard(ColumnMapExpectation):
+    """验证中国身份证号(15位/18位)"""
+
+    map_metric = "column_values.valid_chinese_id_card"
+    success_keys = ("mostly",)
+
+    def validate_id(self, id_str: str) -> bool:
+        if not isinstance(id_str, str):
+            return False
+        id_str = id_str.strip()
+        if len(id_str) not in (15, 18):
+            return False
+        if len(id_str) == 18:
+            if not re.match(r"^\d{17}[\dXx]$", id_str):
+                return False
+        elif not re.match(r"^\d{15}$", id_str):
+            return False
+        return True