Commit 7563ac51 authored by Data Governance Dev's avatar Data Governance Dev

feat(backend): 新增 rule_types / like_util / sql_builder + step9 重构支持条件树

自定义规则 v2 后端核心:把单条 LIKE 升级为递归条件树。

- web/core/rule_types.py(新)
  - classify_field(data_type) → text/numeric/date/unknown
  - COMPARATORS_BY_CLASS / DEFAULT_COMPARATOR / INPUT_HINT_BY_CLASS
  - 文本 2 算子(contains / not_contains);数值 3 算子(gt / lt / eq);日期 2 算子(date_before / date_after)

- web/core/like_util.py(新)
  - escape_like(s) + ESCAPE_CHAR = '!'(跨 DB 一致,避开达梦 [CODE:-6106])
  - 从 step9 内嵌函数上提,从此模块 import

- web/core/sql_builder.py(新)
  - build_where(node, bind, column_quoted, db_type) 递归编译 group 树
  - _placeholder(db_type) MySQL %s / 达梦+Oracle ?
  - 顶层 compile_rule(group, column_quoted, db_type) → (clause, params)

- web/core/step_impl/step9_custom_rules.py
  - 主循环重写:每条规则编译 group 树 → 跑 SQL → LIMIT 200 命中
  - _normalize_rule 自动包旧 payload {rule_type, user_input} → 新 group 树,兼容旧调用
  - 摘要字段补齐:total_rules / executed_rules / skipped_empty_keyword / skipped_no_valid_leaf / failed_rules / total_matched_rows / violations / fields_checked
parent 02a8c7ab
"""LIKE 模式转义工具(跨 DB 共享)
escape char 用 `!`(单字符),原因:
- MySQL / 达梦 / Oracle 都接受 ESCAPE '<任意单字符>'
- 不需要 SQL 反斜杠转义(避免达梦 [CODE:-6106] 无效的转义字符长度)
- 用户输入里 `!` 极少;万一有,先转义成 `!!`
供 sql_builder / step9 共享,避免再在 step9 内嵌一份。
"""
from __future__ import annotations
ESCAPE_CHAR = "!"
def escape_like(s: str) -> str:
"""对 LIKE 模式里的 ! \\ % _ 做反转义,保证用户输入的字面量不被当通配符。
转义顺序:! 必须最先,否则后面产生的 \\! 会被二次转义成 \\!!。
"""
return (
s.replace("!", "!!") # ! → !! (escape char itself, must first)
.replace("\\", "!\\") # \ → !\
.replace("%", "!%") # % → !%
.replace("_", "!_") # _ → !_
)
\ No newline at end of file
"""字段类型 → 比较符映射(自定义规则 v2)
把 DB 的 data_type 归类到三大类:
- text 字符类:varchar / char / text / clob ...
- numeric 数值类:int / decimal / number ...
- date 日期类:date / datetime / timestamp ...
每类对应一组「叶子比较符」:
text: contains / not_contains
numeric: gt / lt / eq
date: date_before / date_after
前后端共享这套语义:
- 前端 <condition-tree> 组件读 COMPARATORS_BY_CLASS 决定 el-option 列表
- 后端 sql_builder 用 COMPARATOR_TO_SQL 决定 WHERE 片段拼法
"""
from __future__ import annotations
from .db_adapter import DATE_TYPES, NUMERIC_TYPES, TEXT_TYPES
# ── 三大类别 ──────────────────────────────────────────────
TEXT = "text"
NUMERIC = "numeric"
DATE = "date"
UNKNOWN = "unknown"
def classify_field(data_type: str) -> str:
"""把后端的 data_type(varchar / int / datetime 等)归到 text / numeric / date。
优先 numeric(避免 int 同时被识别成 text —— 比如某 DB 把 int 也列在 text 里
的奇怪情况;数值和日期都优先于 text)。
"""
dt = (data_type or "").strip().lower()
if not dt:
return UNKNOWN
if dt in NUMERIC_TYPES:
return NUMERIC
if dt in DATE_TYPES:
return DATE
if dt in TEXT_TYPES:
return TEXT
return UNKNOWN
# ── 比较符列表(前端 el-option 用) ────────────────────────
COMPARATORS_BY_CLASS: dict[str, list[dict]] = {
TEXT: [
{"value": "contains", "label": "包含"},
{"value": "not_contains", "label": "不包含"},
],
NUMERIC: [
{"value": "gt", "label": ">"},
{"value": "lt", "label": "<"},
{"value": "eq", "label": "="},
],
DATE: [
{"value": "date_before", "label": "早于"},
{"value": "date_after", "label": "晚于"},
],
}
# 默认 comparator(每类第一个)
DEFAULT_COMPARATOR: dict[str, str] = {
TEXT: "contains",
NUMERIC: "gt",
DATE: "date_before",
}
# ── 比较符 → SQL 操作符(sql_builder 用) ──────────────────
# contains / not_contains 走 LIKE(特殊)
# gt / lt / eq / date_before / date_after 走标准算术
_COMPARATOR_TO_SQL_OP = {
"gt": ">",
"lt": "<",
"eq": "=",
"date_before": "<",
"date_after": ">",
}
def comparator_to_sql_op(comparator: str) -> str | None:
"""标准算术比较符 → SQL 操作符;LIKE 类(contains / not_contains)返回 None。"""
return _COMPARATOR_TO_SQL_OP.get(comparator)
def is_like_comparator(comparator: str) -> bool:
"""是否是 LIKE 系(contains / not_contains),走 ESCAPE 转义。"""
return comparator in ("contains", "not_contains")
# ── 输入提示(前端 placeholder 用) ────────────────────────
INPUT_HINT_BY_CLASS: dict[str, str] = {
TEXT: "关键字(LIKE %x%)",
NUMERIC: "数值",
DATE: "日期 YYYY-MM-DD",
}
\ No newline at end of file
"""把「条件树」编译成 SQL WHERE 子句 + 绑定参数。
条件树节点(dict)两种形态:
- group: { kind: 'group', op: 'AND' | 'OR', children: [node, ...] }
- leaf: { kind: 'leaf', comparator: 'contains' | 'not_contains' | 'gt' | 'lt'
| 'eq' | 'date_before' | 'date_after',
value: <str|number> }
返回:
build_where(node, bind, column_quoted) -> str
- 递归:group → '(c1 <op> c2 <op> c3)'
- leaf → '<quoted col> <cmp_op> <placeholder>'
bind.params 在递归过程中被 .append() 填充;调用方最后用 tuple(bind.params)
传给 db.fetchall(sql, params) 完成绑定。
跨 DB 占位符:
MySQL → %s (pymysql pyformat)
达梦 / Oracle → ? (dmPython qmark;oracledb thin 自动转 :1)
占位符在每条 leaf 里**单独**取一次(db_type 全局固定),保证整条 SQL 用同一种风格。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from .like_util import ESCAPE_CHAR, escape_like
from .rule_types import comparator_to_sql_op, is_like_comparator
# ── 参数绑定容器 ─────────────────────────────────────────
@dataclass
class Bind:
"""递归过程中累积的绑定参数;调用方最后拿 tuple(bind.params) 给 DB 驱动。"""
params: list = field(default_factory=list)
def _placeholder(db_type: str) -> str:
"""跨方言占位符(每个 leaf 调一次,避免组拼出来混用)。"""
return "%s" if (db_type or "").lower() == "mysql" else "?"
# ── 主入口 ────────────────────────────────────────────────
def build_where(
node: dict | None,
bind: Bind,
column_quoted: str,
db_type: str,
) -> str:
"""递归编译条件树 → WHERE 子句片段(不含 WHERE 关键字本身)。
Args:
node: 根节点(group 或 leaf);None / 空组 → 返回 ''
bind: 共享的 Bind 容器,会被 .append() 填充参数
column_quoted: 已 quote 的列名(带反引号 / 双引号),所有 leaf 共用
db_type: 'mysql' | 'dameng' | 'oracle'(决定占位符)
Returns:
SQL WHERE 子句片段;空组返回 ''(调用方自行决定跳过 / 兜底)
"""
if not node:
return ""
kind = node.get("kind")
if kind == "group":
op = (node.get("op") or "AND").upper()
if op not in ("AND", "OR"):
op = "AND"
children = node.get("children") or []
sub_clauses = [
build_where(c, bind, column_quoted, db_type)
for c in children
]
# 过滤空叶子(value 为空 → 编译出 '')
sub_clauses = [s for s in sub_clauses if s]
if not sub_clauses:
return ""
if len(sub_clauses) == 1:
return sub_clauses[0]
return "(" + f" {op} ".join(sub_clauses) + ")"
# kind == 'leaf'
cmp = node.get("comparator")
val = node.get("value")
# 叶子 value 为空 → 返回 '',让外层 group 过滤掉(避免生成 'col LIKE %s' 但 bind 失败)
if val is None or (isinstance(val, str) and not val.strip()):
return ""
# LIKE 系
if is_like_comparator(cmp):
pattern = f"%{escape_like(str(val))}%"
bind.params.append(pattern)
sql_op = "LIKE" if cmp == "contains" else "NOT LIKE"
return f"{column_quoted} {sql_op} {_placeholder(db_type)} ESCAPE '{ESCAPE_CHAR}'"
# 算术系(> / < / = / date_before / date_after)
sql_op = comparator_to_sql_op(cmp)
if sql_op is None:
# 未知 comparator → 跳过(不报错,让上层兜底报「规则无可执行子条件」)
return ""
bind.params.append(val)
return f"{column_quoted} {sql_op} {_placeholder(db_type)}"
# ── 顶层便捷函数 ──────────────────────────────────────────
def compile_rule(
group_node: dict | None,
column_quoted: str,
db_type: str,
) -> tuple[str, list]:
"""单条规则的树 → (where_clause, params)。
Args:
group_node: 一条规则的顶层 group(可能为空)
column_quoted: quote 过的列名
db_type: 数据库方言
Returns:
(where_clause, params_list)
- where_clause 不含 'WHERE' 关键字,可直接拼到 SELECT ... FROM ... WHERE <clause>
- 若 group 为空 / 无有效叶子,返回 ('', [])
"""
bind = Bind()
clause = build_where(group_node, bind, column_quoted, db_type)
return clause, bind.params
\ No newline at end of file
This diff is collapsed.
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