Merge pull request #285 from Integration-Automation/feat/data-profile-batch

JE-Chen · web-flow · commit c225cb794a0c · 2026-06-21T22:09:44.000+08:00
Add data profiling and schema inference
diff --git a/README.md b/README.md
@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [What's new (2026-06-21) — Data Profiling & Schema Inference](#whats-new-2026-06-21--data-profiling--schema-inference)
 - [What's new (2026-06-21) — W3C Trace Context Propagation](#whats-new-2026-06-21--w3c-trace-context-propagation)
 - [What's new (2026-06-21) — HTTP Record & Replay Cassette](#whats-new-2026-06-21--http-record--replay-cassette)
 - [What's new (2026-06-21) — Bulkhead & Rate-Limit Headers](#whats-new-2026-06-21--bulkhead--rate-limit-headers)
@@ -129,6 +130,12 @@
 
 ---
 
+## What's new (2026-06-21) — Data Profiling & Schema Inference
+
+Survey a row-set and propose a validation schema. Full reference: [`docs/source/Eng/doc/new_features/v77_features_doc.rst`](docs/source/Eng/doc/new_features/v77_features_doc.rst).
+
+- **`profile_rows` / `infer_schema`** (`AC_profile_rows`, `AC_infer_schema`): `validate_rows` consumes a hand-written schema and `stats.describe` summarizes one numeric list — nothing surveyed a whole row-set. This profiles each column (null fraction, cardinality, inferred type, top values, numeric min/max/mean) and infers a `validate_rows`-compatible schema (required where non-null, unique where distinct, numeric bounds) — the profiler step that feeds the existing validator. Pure-stdlib, fully deterministic.
+
 ## What's new (2026-06-21) — W3C Trace Context Propagation
 
 Correlate spans and logs across HTTP boundaries. Full reference: [`docs/source/Eng/doc/new_features/v76_features_doc.rst`](docs/source/Eng/doc/new_features/v76_features_doc.rst).
diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md
@@ -12,6 +12,7 @@
 
 ## 目录
 
+- [本次更新 (2026-06-21) — 数据剖析与结构推断](#本次更新-2026-06-21--数据剖析与结构推断)
 - [本次更新 (2026-06-21) — W3C Trace Context 传播](#本次更新-2026-06-21--w3c-trace-context-传播)
 - [本次更新 (2026-06-21) — HTTP 录制与重播卡带](#本次更新-2026-06-21--http-录制与重播卡带)
 - [本次更新 (2026-06-21) — 隔舱与速率限制标头](#本次更新-2026-06-21--隔舱与速率限制标头)
@@ -128,6 +129,12 @@
 
 ---
 
+## 本次更新 (2026-06-21) — 数据剖析与结构推断
+
+扫描数据行集合并提出验证结构。完整参考:[`docs/source/Zh/doc/new_features/v77_features_doc.rst`](../docs/source/Zh/doc/new_features/v77_features_doc.rst)。
+
+- **`profile_rows` / `infer_schema`**(`AC_profile_rows`、`AC_infer_schema`):`validate_rows` 消费手写结构,`stats.describe` 只汇总单一数值列表 —— 没有任何东西扫描整个数据行集合。本功能剖析每列(空值比例、基数、推断类型、最常见值、数值 min/max/mean)并推断出 `validate_rows` 兼容结构(无空值即 required、相异即 unique、数值边界)—— 馈入既有验证器的剖析步骤。纯标准库、完全确定。
+
 ## 本次更新 (2026-06-21) — W3C Trace Context 传播
 
 跨 HTTP 边界关联 span 与日志。完整参考:[`docs/source/Zh/doc/new_features/v76_features_doc.rst`](../docs/source/Zh/doc/new_features/v76_features_doc.rst)。
diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md
@@ -12,6 +12,7 @@
 
 ## 目錄
 
+- [本次更新 (2026-06-21) — 資料剖析與結構推斷](#本次更新-2026-06-21--資料剖析與結構推斷)
 - [本次更新 (2026-06-21) — W3C Trace Context 傳播](#本次更新-2026-06-21--w3c-trace-context-傳播)
 - [本次更新 (2026-06-21) — HTTP 錄製與重播卡帶](#本次更新-2026-06-21--http-錄製與重播卡帶)
 - [本次更新 (2026-06-21) — 隔艙與速率限制標頭](#本次更新-2026-06-21--隔艙與速率限制標頭)
@@ -128,6 +129,12 @@
 
 ---
 
+## 本次更新 (2026-06-21) — 資料剖析與結構推斷
+
+掃描資料列集合並提出驗證結構。完整參考:[`docs/source/Zh/doc/new_features/v77_features_doc.rst`](../docs/source/Zh/doc/new_features/v77_features_doc.rst)。
+
+- **`profile_rows` / `infer_schema`**(`AC_profile_rows`、`AC_infer_schema`):`validate_rows` 消費手寫結構,`stats.describe` 只彙總單一數值清單 —— 沒有任何東西掃描整個資料列集合。本功能剖析每欄(空值比例、基數、推斷型別、最常見值、數值 min/max/mean)並推斷出 `validate_rows` 相容結構(無空值即 required、相異即 unique、數值邊界)—— 餵給既有驗證器的剖析步驟。純標準函式庫、完全具決定性。
+
 ## 本次更新 (2026-06-21) — W3C Trace Context 傳播
 
 跨 HTTP 邊界關聯 span 與日誌。完整參考:[`docs/source/Zh/doc/new_features/v76_features_doc.rst`](../docs/source/Zh/doc/new_features/v76_features_doc.rst)。
diff --git a/docs/source/Eng/doc/new_features/v77_features_doc.rst b/docs/source/Eng/doc/new_features/v77_features_doc.rst
@@ -0,0 +1,44 @@
+Data Profiling & Schema Inference
+=================================
+
+``data_quality.validate_rows`` *consumes* a hand-written schema and
+``stats.describe`` summarises one numeric list — nothing surveyed a whole
+row-set to report per-column null fraction, cardinality, inferred type, value
+ranges, and top values, nor proposed a starting schema. This adds the profiler
+step that feeds the existing validator.
+
+Pure standard library (``collections`` + reuse of ``stats``); imports no
+``PySide6``. Every function is pure (rows in, dict out), so it is fully
+deterministic in CI.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import profile_rows, infer_schema, validate_rows, load_rows
+
+    rows = load_rows("export.csv")
+    profile = profile_rows(rows)
+    # profile["columns"]["age"] -> {count, null_count, null_fraction, distinct,
+    #   unique, inferred_type, top_values, min, max, mean}
+
+    schema = infer_schema(rows)        # validate_rows-compatible
+    report = validate_rows(rows, schema)
+
+``profile_rows`` returns ``{row_count, columns}`` where each column carries its
+count, null count and fraction, distinct count, a uniqueness flag, the inferred
+type (``int`` / ``number`` / ``bool`` / ``str``), the top values with counts,
+and ``min`` / ``max`` / ``mean`` for numeric columns. ``infer_schema`` turns
+that profile into a schema the existing ``validate_rows`` understands: a column
+is ``required`` when it has no nulls, ``unique`` when every non-null value is
+distinct, and carries numeric bounds. Pass an explicit ``columns`` list to
+restrict either function to a subset.
+
+Executor commands
+-----------------
+
+``AC_profile_rows`` profiles a ``rows`` list (optional ``columns`` subset) and
+returns ``{profile}``. ``AC_infer_schema`` returns ``{schema}``. Both are
+exposed as MCP tools (``ac_profile_rows`` / ``ac_infer_schema``) and as Script
+Builder commands under **Data**.
diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst
@@ -99,6 +99,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v74_features_doc
    doc/new_features/v75_features_doc
    doc/new_features/v76_features_doc
+   doc/new_features/v77_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc
diff --git a/docs/source/Zh/doc/new_features/v77_features_doc.rst b/docs/source/Zh/doc/new_features/v77_features_doc.rst
@@ -0,0 +1,37 @@
+資料剖析與結構推斷
+================
+
+``data_quality.validate_rows`` *消費* 一份手寫結構,``stats.describe`` 只彙總單一數值清單 —— 沒有任何
+東西能掃描整個資料列集合,回報每欄的空值比例、基數、推斷型別、值域與最常見值,也無法提出一份起始結構。
+本功能補上這個剖析步驟,並餵給既有的驗證器。
+
+純標準函式庫(``collections`` + 重用 ``stats``);不匯入 ``PySide6``。每個函式皆為純函式(輸入列、輸出
+dict),因此在 CI 中完全具決定性。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import profile_rows, infer_schema, validate_rows, load_rows
+
+    rows = load_rows("export.csv")
+    profile = profile_rows(rows)
+    # profile["columns"]["age"] -> {count, null_count, null_fraction, distinct,
+    #   unique, inferred_type, top_values, min, max, mean}
+
+    schema = infer_schema(rows)        # validate_rows 相容
+    report = validate_rows(rows, schema)
+
+``profile_rows`` 回傳 ``{row_count, columns}``,每欄帶有其筆數、空值數與比例、相異值數、唯一性旗標、
+推斷型別(``int`` / ``number`` / ``bool`` / ``str``)、最常見值與其次數,以及數值欄的 ``min`` /
+``max`` / ``mean``。``infer_schema`` 把該剖析轉成既有 ``validate_rows`` 能理解的結構:無空值的欄位
+標為 ``required``,每個非空值皆相異則標為 ``unique``,並帶上數值邊界。傳入明確的 ``columns`` 清單可將
+兩個函式限制在子集。
+
+執行器命令
+----------
+
+``AC_profile_rows`` 剖析 ``rows`` 清單(可選 ``columns`` 子集)回傳 ``{profile}``。``AC_infer_schema``
+回傳 ``{schema}``。兩者皆以 MCP 工具(``ac_profile_rows`` / ``ac_infer_schema``)以及 Script Builder
+中 **Data** 分類下的命令提供。
diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst
@@ -99,6 +99,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v74_features_doc
    doc/new_features/v75_features_doc
    doc/new_features/v76_features_doc
+   doc/new_features/v77_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc
diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py
@@ -157,6 +157,8 @@
 from je_auto_control.utils.data_quality import (
     extract_fields, mask_rows, validate_rows,
 )
+# Data profiling: per-column stats + schema inference from observed rows
+from je_auto_control.utils.data_profile import infer_schema, profile_rows
 # i18n / l10n testing: pseudo-localize, overflow + catalog checks
 from je_auto_control.utils.i18n_test import (
     check_catalog, check_overflow, pseudo_localize, pseudo_localize_catalog,
@@ -803,6 +805,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "build_sbom", "write_sbom",
     "merge_results", "shard_flows",
     "extract_fields", "mask_rows", "validate_rows",
+    "infer_schema", "profile_rows",
     "check_catalog", "check_overflow", "pseudo_localize",
     "pseudo_localize_catalog",
     "Checkpoint", "CheckpointStore", "run_resumable",
diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py
@@ -1624,6 +1624,26 @@ def _add_resilience_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Extract the W3C trace context from request headers.",
     ))
+    specs.append(CommandSpec(
+        "AC_profile_rows", "Data", "Data Profile: Profile Rows",
+        fields=(
+            FieldSpec("rows", FieldType.STRING,
+                      placeholder='[{"id": 1, "name": "a"}, {"id": 2}]'),
+            FieldSpec("columns", FieldType.STRING, optional=True,
+                      placeholder='["id", "name"] (omit for all)'),
+        ),
+        description="Per-column stats: null fraction, cardinality, type, ranges.",
+    ))
+    specs.append(CommandSpec(
+        "AC_infer_schema", "Data", "Data Profile: Infer Schema",
+        fields=(
+            FieldSpec("rows", FieldType.STRING,
+                      placeholder='[{"id": 1, "name": "a"}, {"id": 2}]'),
+            FieldSpec("columns", FieldType.STRING, optional=True,
+                      placeholder='["id", "name"] (omit for all)'),
+        ),
+        description="Infer a validate_rows-compatible schema from observed rows.",
+    ))
     specs.append(CommandSpec(
         "AC_rate_limit", "Flow", "Rate Limit (Token Bucket)",
         fields=(
diff --git a/je_auto_control/utils/data_profile/__init__.py b/je_auto_control/utils/data_profile/__init__.py
@@ -0,0 +1,6 @@
+"""Data profiling and schema inference for AutoControl row-sets."""
+from je_auto_control.utils.data_profile.data_profile import (
+    infer_schema, profile_rows,
+)
+
+__all__ = ["infer_schema", "profile_rows"]
diff --git a/je_auto_control/utils/data_profile/data_profile.py b/je_auto_control/utils/data_profile/data_profile.py
@@ -0,0 +1,115 @@
+"""Profile a row-set and infer a validation schema from observed data.
+
+``data_quality.validate_rows`` *consumes* a hand-written schema and
+``stats.describe`` summarises one numeric list — nothing surveys a whole
+row-set to report per-column null fraction, cardinality, inferred type, value
+ranges, and top values, nor proposes a starting schema. This is the
+profiler step that feeds the existing validator.
+
+Pure standard library (``collections`` + reuse of ``stats``); imports no
+``PySide6``. All functions are pure (rows in, dict out) so they are fully
+deterministic in CI.
+"""
+from collections import Counter
+from typing import Any, Dict, List, Optional, Sequence
+
+from je_auto_control.utils.stats.stats import describe
+
+_NULLS = (None, "")
+_TOP_N = 5
+
+
+def _is_int(value: Any) -> bool:
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
+def _is_number(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+def _infer_type(non_null: Sequence[Any]) -> str:
+    if not non_null:
+        return "str"
+    if all(_is_int(value) for value in non_null):
+        return "int"
+    if all(_is_number(value) for value in non_null):
+        return "number"
+    if all(isinstance(value, bool) for value in non_null):
+        return "bool"
+    return "str"
+
+
+def _numeric_summary(kind: str, non_null: Sequence[Any]) -> Dict[str, Any]:
+    if kind not in ("int", "number") or not non_null:
+        return {}
+    stats = describe([float(value) for value in non_null])
+    return {"min": stats["min"], "max": stats["max"], "mean": stats["mean"]}
+
+
+def _column_profile(name: str, rows: List[Dict[str, Any]]) -> Dict[str, Any]:
+    values = [row.get(name) for row in rows]
+    non_null = [value for value in values if value not in _NULLS]
+    null_count = len(values) - len(non_null)
+    kind = _infer_type(non_null)
+    distinct = len({_hashable(value) for value in non_null})
+    top = Counter(_hashable(value) for value in non_null).most_common(_TOP_N)
+    profile = {
+        "count": len(values), "null_count": null_count,
+        "null_fraction": (null_count / len(values)) if values else 0.0,
+        "distinct": distinct,
+        "unique": bool(non_null) and distinct == len(non_null),
+        "inferred_type": kind,
+        "top_values": [{"value": value, "count": count}
+                       for value, count in top],
+    }
+    profile.update(_numeric_summary(kind, non_null))
+    return profile
+
+
+def _hashable(value: Any) -> Any:
+    return value if isinstance(value, (str, int, float, bool)) else repr(value)
+
+
+def _column_names(rows: List[Dict[str, Any]],
+                  columns: Optional[Sequence[str]]) -> List[str]:
+    if columns is not None:
+        return list(columns)
+    seen: Dict[str, None] = {}
+    for row in rows:
+        for key in row:
+            seen.setdefault(key, None)
+    return list(seen)
+
+
+def profile_rows(rows: List[Dict[str, Any]],
+                 columns: Optional[Sequence[str]] = None) -> Dict[str, Any]:
+    """Profile ``rows`` into per-column statistics.
+
+    Returns ``{row_count, columns: {name: {count, null_count, null_fraction,
+    distinct, unique, inferred_type, top_values, [min, max, mean]}}}``.
+    """
+    names = _column_names(rows, columns)
+    return {"row_count": len(rows),
+            "columns": {name: _column_profile(name, rows) for name in names}}
+
+
+def infer_schema(rows: List[Dict[str, Any]],
+                 columns: Optional[Sequence[str]] = None) -> Dict[str, Any]:
+    """Infer a ``validate_rows``-compatible schema from observed ``rows``.
+
+    A column is ``required`` when it has no nulls, ``unique`` when every
+    non-null value is distinct, and carries numeric ``min``/``max`` bounds.
+    """
+    profile = profile_rows(rows, columns)
+    schema: Dict[str, Any] = {}
+    for name, column in profile["columns"].items():
+        rule: Dict[str, Any] = {"type": column["inferred_type"]}
+        if column["null_count"] == 0 and column["count"] > 0:
+            rule["required"] = True
+        if column.get("unique"):
+            rule["unique"] = True
+        if "min" in column:
+            rule["min"] = column["min"]
+            rule["max"] = column["max"]
+        schema[name] = rule
+    return schema
diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py
@@ -3000,6 +3000,28 @@ def _trace_extract(headers: Any) -> Dict[str, Any]:
     return {"context": ctx.to_dict() if ctx is not None else None}
 
 
+def _profile_rows(rows: Any, columns: Any = None) -> Dict[str, Any]:
+    """Adapter: profile a row-set into per-column statistics."""
+    import json
+    from je_auto_control.utils.data_profile import profile_rows
+    if isinstance(rows, str):
+        rows = json.loads(rows)
+    if isinstance(columns, str):
+        columns = json.loads(columns)
+    return {"profile": profile_rows(rows, columns)}
+
+
+def _infer_schema(rows: Any, columns: Any = None) -> Dict[str, Any]:
+    """Adapter: infer a validate_rows-compatible schema from rows."""
+    import json
+    from je_auto_control.utils.data_profile import infer_schema
+    if isinstance(rows, str):
+        rows = json.loads(rows)
+    if isinstance(columns, str):
+        columns = json.loads(columns)
+    return {"schema": infer_schema(rows, columns)}
+
+
 def _percentiles(samples: Any, qs: Any = None) -> Dict[str, Any]:
     """Adapter: exact percentiles of a numeric sample list (or JSON string)."""
     import json
@@ -4078,6 +4100,8 @@ def __init__(self):
             "AC_http_replay": _http_replay,
             "AC_trace_inject": _trace_inject,
             "AC_trace_extract": _trace_extract,
+            "AC_profile_rows": _profile_rows,
+            "AC_infer_schema": _infer_schema,
             "AC_unified_diff": _unified_diff,
             "AC_apply_unified": _apply_unified,
             "AC_three_way_merge": _three_way_merge,
diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py
diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py
diff --git a/test/unit_test/headless/test_data_profile_batch.py b/test/unit_test/headless/test_data_profile_batch.py
