Commit 0967e522 authored by Data Governance Dev's avatar Data Governance Dev

feat(backend): NUMERIC_TYPES / DATE_TYPES + orchestrator 识别 v2 条件树 payload

- web/core/db_adapter.py
  - 新增 NUMERIC_TYPES / DATE_TYPES frozenset(紧挨 TEXT_TYPES)
  - 覆盖 MySQL / 达梦 / Oracle 三方言的常见数值 + 日期类型

- web/core/orchestrator.py
  - 修 bug:"自动追加 custom_value_check 步骤" 判断只认旧 payload {user_input}
  - v2 payload 用 group 树结构 → 永远 False → main_steps=[] → 0 步跑
  - 新增 _has_effective_leaf(node) 递归判断:新旧 payload 都认
  - 修后:steps=[] + 1 条有效规则 → 自动追加 custom_value_check → 跑 200 行命中
parent 7563ac51
......@@ -232,6 +232,28 @@ TEXT_TYPES = {
}
# ── 数值类型集合(自定义规则 v2 用:> / < / = 比较符) ──
NUMERIC_TYPES = {
# MySQL
"int", "integer", "smallint", "tinyint", "bigint", "mediumint",
"decimal", "numeric", "float", "double", "real", "year",
# 达梦 / Oracle
"number", "decimal", "numeric", "float", "double", "real", "binary_float", "binary_double",
"money", "smallmoney", "dec",
}
# ── 日期/时间类型集合(自定义规则 v2 用:早于 / 晚于 比较符) ──
DATE_TYPES = {
# MySQL
"date", "datetime", "timestamp", "time", "year",
# 达梦 / Oracle / SQL Server
"date", "datetime", "timestamp", "time",
"smalldatetime", "datetime2", "datetimeoffset", "interval",
"timestamp with time zone", "timestamp with local time zone",
}
# ── 连接管理 ──────────────────────────────────────────────
class DBConnection:
"""统一的数据库连接封装。
......
......@@ -716,11 +716,28 @@ async def run_governance_workflow(
log("INFO", f"调度计划: 主步骤={main_steps}")
# ── 自动追加:用户提交了 custom_rules → 追加 custom_value_check 步骤 ──
# 仅当至少有一条非空 user_input 的规则(空规则的项会在 step 内被跳过)
has_effective_rule = any(
(r.get("user_input") or "").strip()
for r in (_CUSTOM_RULES_CTX or [])
)
# 兼容新旧两种 payload:
# - 旧:{rule_type, user_input} —— 用 user_input 非空判断
# - 新:{group: {kind, op, children: [...]}} —— 递归查非空叶子
def _has_effective_leaf(node) -> bool:
if not isinstance(node, dict):
return False
if node.get("kind") == "leaf":
v = node.get("value")
return v is not None and (not isinstance(v, str) or v.strip() != "")
if node.get("kind") == "group":
return any(_has_effective_leaf(c) for c in (node.get("children") or []))
return False
def _rule_has_payload(r: dict) -> bool:
"""一条规则有没有可执行内容(新旧 payload 都认)"""
# 新 payload:group 树里有非空叶子
if _has_effective_leaf(r.get("group")):
return True
# 旧 payload:user_input 非空
return bool((r.get("user_input") or "").strip())
has_effective_rule = any(_rule_has_payload(r) for r in (_CUSTOM_RULES_CTX or []))
if has_effective_rule and "custom_value_check" not in main_steps:
main_steps.append("custom_value_check")
log("INFO", "已自动追加自定义规则步骤 custom_value_check", step="custom_value_check")
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment