Commit 8ea2935f authored by Data Governance Dev's avatar Data Governance Dev

feat(web3): 规则扩展为数值/日期类型(Python 函数 + 安全沙箱 + AI 生成)

需求:规则这里增加数值和日期规则(可以切换),同样可以使用AI生成,
目标结果是一个python函数,测试也是调用后台测试。

DB:
- rule 表新增 rule_type(regex/number/date,默认 regex)/ code(Python 函数源码)
- init_db() 自动 ALTER ADD COLUMN(每个 ADD 都带 DEFAULT,老规则仍按 regex 走)

后端:
- core/rule_runner.py(新增):run_rule(rule_type, regex, code, value)
  - regex 走 re.compile.search
  - number/date 走 _exec_user_function(白名单 exec 沙箱)
  - 白名单 __import__:datetime/time/_strptime/math/decimal/re
    (strptime 内部 lazy-import _strptime,所以不能一刀切屏蔽 __import__)
  - 沙箱预注入 datetime/time/math/decimal/re 到 globals
  - validate_user_code():compile + 必须 def check(value) -> bool
- core/ai_rule.py(新增):gen_rule(desc, rule_type)
  - regex 复用老的 ai_regex.gen_regex(行为完全兼容)
  - number/date 走 LLM,validate_user_code 校验,5 次重试
- models/rule.py:Rule 加 rule_type/code + to_dict + RULE_TYPES 常量
- routers/ai.py:
  - 新增 /api/ai/rule(POST {desc, rule_type} → {ok, code, note})
  - 新增 /api/rule/test(POST {rule_type, code, value} → {ok, pass_, error})
  - 保留老 /api/ai/regex 和 /api/regex/test(薄壳转发,老测试不挂)
- routers/tasks.py:RulePayload/RuleOut 加 rule_type/code,按类型持久化
- routers/queries.py:按 rule_type dispatch,number/date 走沙箱;
  沙箱报错(语法错 / 缺 check)也按不合规处理,把错误原因写进 issues

前端:
- src/api/ai.js:genRule/testRule 统一接口(rule_type 参数化)
- src/components/RuleModal.vue:
  - 规则类型 tab 切换(正则/数值/日期)
  - regex 走单行 input;number/date 走等宽 Python 编辑框
  - AI 按钮文案按类型变(生成正则/数值校验函数/日期校验函数)
  - 测试统一走 /api/rule/test,保证和检测引擎语义一致

测试:
- tests/test_rule_runner.py(新增,20 个用例):
  - regex 简单匹配 / 空正则 / 非法正则
  - number 校验函数 / 非法输入 / 缺 check / 运行时异常
  - date strptime 校验 / 范围校验
  - 沙箱安全:os 拒绝 / datetime 预注入 / open 拒绝 / eval 拒绝 / getattr 拒绝
  - validate_user_code 4 项
- tests/test_rule_types_e2e.py(新增,5 个用例):
  - 持久化三种 rule_type
  - 默认 rule_type=regex(向后兼容)
  - queries/run 按 rule_type dispatch
  - 规则执行失败标记为不合规 + 暴露原因
  - 老 /api/ai/regex 端点兼容

迁移:老规则行 rule_type='regex'、code=NULL,自动走旧逻辑(不动现有数据)。
parent 23151684
"""AI 生成规则代码(web3 / 2026-08-21)
三种 rule_type 统一入口 `gen_rule()`:
- regex : 走老 ai_regex.gen_regex(),返回单行正则字符串
- number: LLM 生成 def check(value) -> bool 的 Python 函数(数值校验)
- date : 同 number(用户代码里自己用 datetime.strptime 解析)
强制走 LLM(用户决策:让模型持续见真实描述),拿到代码后用 rule_runner.validate_user_code()
真实校验语法 + 必含 check;编译/校验失败 → 把错误反馈给 LLM 让它修复,最多 5 次。
降级策略同 ai_regex:
- LLM 不可用(无 key / 网络挂)→ 直接返回空串 + 提示文案,不抛异常
- 5 次重试全部失败 → 返回空串 + 失败提示
"""
from __future__ import annotations
import re
from typing import Optional
from web3.backend.core.llm import LLMUnavailable, get_llm_client
from web3.backend.core.rule_runner import (
RuleRunError,
run_rule,
validate_user_code,
)
from web3.backend.models.rule import RULE_TYPES
from web3.backend._logging import get_logger
logger = get_logger("backend.ai_rule")
# ── 复用老 ai_regex ─────────────────────────────────────
from web3.backend.core.ai_regex import gen_regex, test_regex # noqa: E402,F401
# ── System Prompt ───────────────────────────────────────
_SYSTEM_PROMPT = "你是 Python 程序员,擅长写简短的数据校验函数。"
# ── User Prompt ─────────────────────────────────────────
_NUMBER_USER_TEMPLATE_INITIAL = """请把以下规则描述转换成一个 Python 校验函数。
规则描述:「{desc}」
要求:
1. 定义函数 def check(value) -> bool:
- value 是从数据库读出来的原始值(str / int / float 都可能,按需转换)
- 返回 True 表示合规,False 表示不合规
2. 函数要短小(一般不超过 10 行),只做这一项校验
3. 不要 import,直接用 value 即可;如果需要数学常量(如 math.pi)才用 math 模块
4. 如果描述里包含具体范围/阈值(如"0 到 150"、"大于 0"),一定要用上
5. 如果 value 不是合法数字,按不合规返回 False(用 try/except 包住转换)
只输出 def check(value) -> bool: 函数体的 Python 代码,不要任何解释、不要 Markdown 代码块。"""
_NUMBER_USER_TEMPLATE_FIX = """上一次生成的代码有问题。
规则描述:「{desc}」
上次生成的代码:
```
{bad_code}
```
运行时错误:`{error_msg}`
请输出修复后的 def check(value) -> bool: 函数。
要求:
1. 函数要短小,只做这一项校验
2. 修掉上面的错误(语法 / 缺 check / 运行时异常)
3. 仍然不要 import;需要数学模块就 math.
只输出修复后的函数代码,不要任何解释、不要 Markdown 代码块。"""
_DATE_USER_TEMPLATE_INITIAL = """请把以下规则描述转换成一个 Python 日期校验函数。
规则描述:「{desc}」
要求:
1. 定义函数 def check(value) -> str -> bool:
- value 是从数据库读出来的原始字符串(如 "2024-01-15"、"2024/01/15"、"20240115")
- 返回 True 表示合规,False 表示不合规
2. 用 datetime.strptime(v, '<format>') 解析日期,解析失败返回 False
3. 函数要短小(一般不超过 10 行),只做这一项校验
4. 不要写 import 语句,datetime 模块已经预注入到命名空间里
5. 如果描述里包含具体格式(如"yyyy-MM-dd"、"yyyyMMdd")或范围(如"晚于 2020-01-01"),一定要用上
6. 日期常用格式:%Y-%m-%d(2024-01-15)、%Y/%m/%d、%Y%m%d、%Y-%m-%d %H:%M:%S
只输出 def check(value) -> bool: 函数体的 Python 代码,不要任何解释、不要 Markdown 代码块。"""
_DATE_USER_TEMPLATE_FIX = """上一次生成的代码有问题。
规则描述:「{desc}」
上次生成的代码:
```
{bad_code}
```
运行时错误:`{error_msg}`
请输出修复后的 def check(value) -> bool: 函数。
要求:
1. 用 datetime.strptime(v, '<format>') 解析日期,解析失败返回 False
2. 修掉上面的错误(语法 / 缺 check / 运行时异常)
3. 仍然不要写 import
只输出修复后的函数代码,不要任何解释、不要 Markdown 代码块。"""
# ── 从 LLM 返回中提取代码 ─────────────────────────────────
# LLM 偶尔会包一层 ```python ... ```,做容错提取
_CODE_FENCE = re.compile(r"^```(?:python)?\s*\n?(.*?)\n?```$", re.DOTALL)
def _extract_code(text: str) -> str:
"""从 LLM 返回中提取 def check(value) ... 代码块。"""
if not text:
return ""
text = text.strip()
m = _CODE_FENCE.match(text)
if m:
return m.group(1).strip()
# 多行:取第一个非空行往下到末尾
lines = [ln for ln in text.splitlines() if ln.strip()]
return "\n".join(lines).strip()
# ── 主入口 ─────────────────────────────────────────────
def gen_rule(desc: str, rule_type: str) -> tuple[str, str]:
"""AI 生成规则代码。
返回 (code, note):
code 非空 = 成功;空 = 失败
note 给前端弹窗用的提示文案
"""
desc = (desc or "").strip()
if not desc:
return "", "规则描述为空"
if rule_type not in RULE_TYPES:
return "", f"不支持的规则类型:{rule_type!r}"
# regex 走老逻辑,保持完全兼容
if rule_type == "regex":
return gen_regex(desc)
client = get_llm_client()
if not client.available:
return "", "LLM 未配置 API Key,请联系管理员配置或手动填写"
# number / date:选 prompt 模板
if rule_type == "number":
tpl_init = _NUMBER_USER_TEMPLATE_INITIAL
tpl_fix = _NUMBER_USER_TEMPLATE_FIX
else: # date
tpl_init = _DATE_USER_TEMPLATE_INITIAL
tpl_fix = _DATE_USER_TEMPLATE_FIX
bad_code = ""
error_msg = ""
last_code = ""
MAX_TRIES = 5
for attempt in range(1, MAX_TRIES + 1):
try:
if attempt == 1:
prompt = tpl_init.format(desc=desc)
else:
prompt = tpl_fix.format(
desc=desc, bad_code=bad_code, error_msg=error_msg
)
text = client.complete(prompt, system=_SYSTEM_PROMPT)
last_code = _extract_code(text)
ok, err = validate_user_code(last_code)
if ok:
logger.info(
f"AI {rule_type} 规则:LLM 第 {attempt} 次成功 "
f"desc={desc!r} → {len(last_code)} chars"
)
return last_code, f"已用 LLM 生成(第 {attempt} 次)"
bad_code = last_code or bad_code
error_msg = err
logger.warning(
f"AI {rule_type} 规则:LLM 第 {attempt}/{MAX_TRIES} 次生成的代码未通过校验 "
f"desc={desc!r} err={err}"
)
except LLMUnavailable as e:
logger.warning(f"AI {rule_type} 规则:LLM 第 {attempt} 次调用失败 desc={desc!r} err={e}")
return "", f"LLM 调用失败:{e}"
return "", f"LLM 重试 {MAX_TRIES} 次仍未生成合法代码(最后错误:{error_msg}),请补充描述或手动填写"
# ── 测试 ─────────────────────────────────────────────
def test_rule(rule_type: str, code_or_regex: str, value: str) -> dict:
"""用 rule_type 对应的引擎校验 value。
返回 dict(与 test_regex 兼容的字段):
- ok: bool 代码/正则本身是否合法
- match: bool 值是否通过
- pass_: 别名,等于 match(前端习惯叫「通过/不通过」)
- error: str 错误信息(成功时空串)
"""
try:
passed = run_rule(rule_type, code_or_regex if rule_type == "regex" else None,
code_or_regex if rule_type in ("number", "date") else None,
value)
return {"ok": True, "match": passed, "pass_": passed, "error": ""}
except RuleRunError as e:
return {"ok": False, "match": False, "pass_": False, "error": str(e)}
except Exception as e:
return {"ok": False, "match": False, "pass_": False,
"error": f"{type(e).__name__}: {e}"}
\ No newline at end of file
"""规则执行引擎(web3)
三种 rule_type 走同一个入口 `run_rule()`,内部按类型 dispatch:
- regex : re.compile(regex).search(value) → 命中 = 通过(True)
- number: 用户提供的 Python 函数 def check(value) → 数值校验
- date : 同 number → 日期校验(函数里自己 strptime)
设计要点:
- 数值 / 日期类规则的代码是用 LLM 生成的,**用户/AI 可能写出危险代码**;
用受限命名空间 exec:白名单 builtins + 预注入 datetime/re/math/decimal 模块,
禁掉 open/eval/exec/__import__/getattr/globals 等危险入口
- 沙箱不杀进程(不搞 signal.alarm,Windows 不支持),只防「合理范围内的破坏」;
真要做超时限制,得用 multiprocessing / subprocess,下个迭代再说
- 任何错误(语法错 / 缺 check / 运行时异常)都包成 RuleRunError,
检测引擎拿 (passed, error_msg) 决定要不要把这一条规则标红
"""
from __future__ import annotations
import builtins as _bi
import datetime
import decimal
import math
import re as _re
import time as _time
from typing import Any, Optional
# 沙箱允许的 import 模块白名单(覆盖 __import__ 实际调用 + `import x` 语句)
# 主要为了让 datetime.datetime.strptime 这种「内部 lazy-import」能跑
_SAFE_IMPORT_NAMES = frozenset({
"datetime", "time", "_strptime", "math", "decimal", "re",
})
def _safe_import(name, globals=None, locals=None, fromlist=(), level=0):
"""白名单版 __import__:只允许 _SAFE_IMPORT_NAMES 里的模块。
datetime.strptime 内部用 `import _strptime` 这种形式 → 会调到 __import__;
不开 __import__ 沙箱里就跑不起来。开了又怕用户 __import__('os') 逃逸,
用这个白名单版本控制可导入的模块名集合。
"""
if level != 0:
# 形如 `from .x import y` 的相对 import 一律拒绝(规则代码不需要)
raise ImportError("Relative imports are not allowed in rule code")
if name not in _SAFE_IMPORT_NAMES:
raise ImportError(
f"Import of {name!r} is not allowed in rule code "
f"(allowed: {sorted(_SAFE_IMPORT_NAMES)})"
)
return _bi.__import__(name, globals, locals, fromlist, level)
# ── 安全沙箱 ────────────────────────────────────────────
# 只放白名单内的内置函数到 __builtins__ 里。
# 故意不放:open / eval / exec / compile / globals / locals /
# vars / getattr / setattr / delattr / breakpoint / input /
# memoryview / exit / quit / help / object / super / property /
# __build_class__(防 type(name, (), {}) 动态建类)
# __import__ 用 _safe_import 替换(白名单版)
_SAFE_BUILTIN_NAMES = {
# 类型 / 构造
'bool', 'int', 'float', 'str', 'complex', 'bytes', 'bytearray',
'list', 'tuple', 'set', 'frozenset', 'dict',
# 集合操作
'len', 'range', 'enumerate', 'zip', 'map', 'filter', 'reversed',
'sorted', 'min', 'max', 'sum', 'any', 'all', 'slice',
# 数学
'abs', 'round', 'pow', 'divmod',
# 布尔 / 类型检查
'isinstance', 'issubclass', 'type', 'callable',
# 数值转换
'hex', 'oct', 'bin', 'ord', 'chr', 'repr',
# 异常类型(用户 catch 用得上)
'Exception', 'ValueError', 'TypeError', 'ArithmeticError',
'ZeroDivisionError', 'IndexError', 'KeyError', 'StopIteration',
'NotImplemented', 'Ellipsis',
# 字面常量
'True', 'False', 'None',
}
_SAFE_BUILTINS: dict[str, Any] = {
name: getattr(_bi, name) for name in _SAFE_BUILTIN_NAMES
}
# 用白名单版 __import__ 替换掉真 __import__
_SAFE_BUILTINS['__import__'] = _safe_import
# 预注入到 exec 命名空间的安全模块(用户代码可以直接用 datetime 等,不需要 import)
_PRELOAD_MODULES: dict[str, Any] = {
"datetime": datetime,
"time": _time,
"math": math,
"decimal": decimal,
"re": _re,
}
class RuleRunError(Exception):
"""规则执行失败(语法错 / 缺 check / 运行时异常),检测引擎按不合规处理并把原因暴露给 issues"""
def _exec_user_function(code: str, value: Any) -> bool:
"""在受限命名空间里执行用户提供的 Python 函数源码,调用 check(value) 返回 bool。
抛出 RuleRunError:
- SyntaxError / IndentationError 等语法错
- 缺少 check 函数定义
- check() 调用时抛出的异常
"""
namespace: dict[str, Any] = {
"__builtins__": _SAFE_BUILTINS,
}
namespace.update(_PRELOAD_MODULES) # 预注入模块
# compile() 先走一遍拿到语法错的具体行号
try:
compiled = compile(code, "<user_rule>", "exec")
except SyntaxError as e:
raise RuleRunError(f"语法错误:{e.msg or e}") from e
try:
exec(compiled, namespace)
except Exception as e:
raise RuleRunError(f"执行错误:{type(e).__name__}: {e}") from e
check = namespace.get("check")
if not callable(check):
raise RuleRunError("代码中必须定义 check(value) 函数")
try:
result = check(value)
except Exception as e:
raise RuleRunError(f"check() 调用异常:{type(e).__name__}: {e}") from e
return bool(result)
# ── 主入口 ──────────────────────────────────────────────
def run_rule(rule_type: str, regex: Optional[str], code: Optional[str], value: Any) -> bool:
"""执行一条规则,返回 True = 通过 / False = 不通过。
参数:
rule_type: 'regex' | 'number' | 'date'
regex: regex 类型用;其他类型传 None
code: number/date 类型用;regex 传 None
value: 数据源该字段的原始值(字符串或 None)
抛出 RuleRunError:
- regex 编译失败
- 用户函数语法错 / 缺 check / 运行时异常
- rule_type 非法
"""
rt = rule_type or "regex"
if rt == "regex":
pattern_src = (regex or "").strip() or r"^.+$" # 空 regex = 非空校验
try:
pat = _re.compile(pattern_src)
except _re.error as e:
raise RuleRunError(f"正则编译失败:{e}") from e
val_str = "" if value is None else str(value)
return pat.search(val_str) is not None
if rt in ("number", "date"):
if not code or not code.strip():
raise RuleRunError("规则代码为空,请填写 check(value) 函数")
return _exec_user_function(code, value)
raise RuleRunError(f"未知的规则类型:{rule_type!r}")
# ── 仅用于 AI 生成阶段 / 测试阶段的「纯语法校验」 ──────────
def validate_user_code(code: str) -> tuple[bool, str]:
"""校验用户代码语法 + 必须含 check(value)。
返回 (ok, error_msg):
ok=True 表示「至少语法 OK + 定义了 check」,但**不保证 check 能跑通具体值**。
"""
if not code or not code.strip():
return False, "代码为空"
try:
compiled = compile(code, "<user_rule>", "exec")
except SyntaxError as e:
return False, f"语法错误:{e.msg or e}"
namespace: dict[str, Any] = {"__builtins__": _SAFE_BUILTINS}
namespace.update(_PRELOAD_MODULES)
try:
exec(compiled, namespace)
except Exception as e:
return False, f"执行错误:{type(e).__name__}: {e}"
if not callable(namespace.get("check")):
return False, "代码中必须定义 check(value) 函数"
return True, ""
\ No newline at end of file
...@@ -92,6 +92,9 @@ def init_db() -> None: ...@@ -92,6 +92,9 @@ def init_db() -> None:
# 解决:drop field(rule 跟着 ON DELETE CASCADE 一起清)→ create_all 重建带 UNIQUE 的版本。 # 解决:drop field(rule 跟着 ON DELETE CASCADE 一起清)→ create_all 重建带 UNIQUE 的版本。
# 重建后字段/规则数据会丢,需要用户重新保存任务。 # 重建后字段/规则数据会丢,需要用户重新保存任务。
_migrate_field_unique() _migrate_field_unique()
# 一次性迁移:rule 表加 rule_type / code 列(2026-08-21 扩展规则类型:数值 / 日期)
# 老数据全部保持 rule_type='regex',code 留空 → 跟旧逻辑完全等价
_migrate_rule_type_and_code()
logger.info(f"[init_db] DB 已存在,已执行增量建表检查:{DB_PATH}") logger.info(f"[init_db] DB 已存在,已执行增量建表检查:{DB_PATH}")
# 灌种子(每个 seed 函数内部判重,可重复调用) # 灌种子(每个 seed 函数内部判重,可重复调用)
seed.seed_task_groups() seed.seed_task_groups()
...@@ -152,4 +155,26 @@ def _migrate_field_unique() -> None: ...@@ -152,4 +155,26 @@ def _migrate_field_unique() -> None:
logger.info("[migrate] field 缺显式 UNIQUE INDEX uq_field_task_key,补建(不再 drop table)") logger.info("[migrate] field 缺显式 UNIQUE INDEX uq_field_task_key,补建(不再 drop table)")
conn.exec_driver_sql( conn.exec_driver_sql(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_field_task_key ON field(task_id, field_key)" "CREATE UNIQUE INDEX IF NOT EXISTS uq_field_task_key ON field(task_id, field_key)"
) )
\ No newline at end of file
def _migrate_rule_type_and_code() -> None:
"""给 rule 表加 rule_type / code 列(2026-08-21 扩展数值/日期规则)。
Base.metadata.create_all 不会给已存在的表加列,SQLite 也不支持「IF NOT EXISTS 列」。
这里走 pragma_table_info 检查 → 缺哪个加哪个,每个 ADD COLUMN 都带 DEFAULT,
保证老规则行(rule_type='regex'、code=NULL)仍然按旧逻辑生效。
"""
with engine.begin() as conn:
cols = {
row[1]: row
for row in conn.exec_driver_sql("PRAGMA table_info(rule)").fetchall()
}
if "rule_type" not in cols:
logger.info("[migrate] rule 缺 rule_type 列,补建(默认 'regex',老规则不受影响)")
conn.exec_driver_sql(
"ALTER TABLE rule ADD COLUMN rule_type TEXT NOT NULL DEFAULT 'regex'"
)
if "code" not in cols:
logger.info("[migrate] rule 缺 code 列,补建(默认 NULL,regex 规则用不上)")
conn.exec_driver_sql("ALTER TABLE rule ADD COLUMN code TEXT")
\ No newline at end of file
...@@ -63,11 +63,19 @@ CREATE INDEX idx_field_task ON field(task_id); ...@@ -63,11 +63,19 @@ CREATE INDEX idx_field_task ON field(task_id);
-- ───────────────────────────────────────────────────────────── -- ─────────────────────────────────────────────────────────────
-- 4) rule · 规则(每条规则挂在一个 field 下) -- 4) rule · 规则(每条规则挂在一个 field 下)
-- ───────────────────────────────────────────────────────────── -- ─────────────────────────────────────────────────────────────
-- rule_type 'regex' / 'number' / 'date' 三类校验规则(2026-08-21 扩展)
-- regex: 仍是 Python re 模式,存在 regex 列(兼容老数据,默认值)
-- number: Python 函数 def check(value) -> bool,存在 code 列
-- date: 同 number,存在 code 列
-- code number/date 类型的 Python 函数源码(必须含 def check(value): 返回 bool)
-- regex regex 类型专用;number/date 留空
CREATE TABLE rule ( CREATE TABLE rule (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
field_id INTEGER NOT NULL REFERENCES field(id) ON DELETE CASCADE, field_id INTEGER NOT NULL REFERENCES field(id) ON DELETE CASCADE,
desc TEXT NOT NULL, desc TEXT NOT NULL,
regex TEXT, regex TEXT,
rule_type TEXT NOT NULL DEFAULT 'regex',
code TEXT,
ord INTEGER NOT NULL DEFAULT 0 ord INTEGER NOT NULL DEFAULT 0
); );
CREATE INDEX idx_rule_field ON rule(field_id); CREATE INDEX idx_rule_field ON rule(field_id);
......
...@@ -2,38 +2,53 @@ ...@@ -2,38 +2,53 @@
对应 schema.sql 里的: 对应 schema.sql 里的:
CREATE TABLE rule ( CREATE TABLE rule (
id, field_id, desc, regex, ord id, field_id, desc, regex, rule_type, code, ord
) )
设计: 设计:
- 每条规则属于一个 field(field_id FK + ON DELETE CASCADE) - 每条规则属于一个 field(field_id FK + ON DELETE CASCADE)
- desc:自然语言规则说明(如「必须是 18 位中国大陆身份证号」) - desc:自然语言规则说明(如「必须是 18 位中国大陆身份证号」)
- regex:Python 正则表达式(可空——「非空」这种规则 regex 留空 + 检测时由后端补 `^.+$`) - rule_type:'regex' / 'number' / 'date'(2026-08-21 扩展)
- regex:传统正则(向后兼容,老数据全部为 regex)
- number:Python 函数 def check(value) -> bool,存 code 列
- date: 同 number,存 code 列(用户代码里自己用 datetime.strptime 解析)
- regex:仅 rule_type='regex' 时使用(可空——「非空」这种规则 regex 留空 + 检测时由后端补 `^.+$`)
- code: 仅 rule_type in ('number', 'date') 时使用,存 Python 函数源码
- ord:规则在 UI 上的显示顺序(保存时按数组顺序写入) - ord:规则在 UI 上的显示顺序(保存时按数组顺序写入)
""" """
from __future__ import annotations from __future__ import annotations
from sqlalchemy import Column, Integer, Text, ForeignKey from sqlalchemy import Column, Integer, Text, String, ForeignKey
from web3.backend.db.database import Base from web3.backend.db.database import Base
# rule_type 的合法值集中在一处常量,ORM / 校验 / 提示文案都引用这一份
RULE_TYPES = ("regex", "number", "date")
DEFAULT_RULE_TYPE = "regex"
class Rule(Base): class Rule(Base):
__tablename__ = "rule" __tablename__ = "rule"
id = Column(Integer, primary_key=True, autoincrement=True) id = Column(Integer, primary_key=True, autoincrement=True)
field_id = Column(Integer, ForeignKey("field.id", ondelete="CASCADE"), nullable=False) field_id = Column(Integer, ForeignKey("field.id", ondelete="CASCADE"), nullable=False)
desc = Column(Text, nullable=False) # 规则说明(自然语言) desc = Column(Text, nullable=False) # 规则说明(自然语言)
regex = Column(Text, nullable=True) # 正则表达式(可空) regex = Column(Text, nullable=True) # Python 正则(regex 类型用)
rule_type = Column( # 校验类型(2026-08-21 扩展)
String, nullable=False, default=DEFAULT_RULE_TYPE,
)
code = Column(Text, nullable=True) # Python 函数源码(number/date 用)
ord = Column(Integer, nullable=False, default=0) ord = Column(Integer, nullable=False, default=0)
def to_dict(self) -> dict: def to_dict(self) -> dict:
return { return {
"id": self.id, "id": self.id,
"desc": self.desc, "desc": self.desc,
"regex": self.regex, "regex": self.regex,
"ord": self.ord, "rule_type": self.rule_type or DEFAULT_RULE_TYPE,
"code": self.code,
"ord": self.ord,
} }
def __repr__(self) -> str: def __repr__(self) -> str:
return f"<Rule id={self.id} field_id={self.field_id} desc={self.desc!r}>" return f"<Rule id={self.id} field_id={self.field_id} rule_type={self.rule_type!r}>"
\ No newline at end of file \ No newline at end of file
"""AI 相关 API(web3) """AI 相关 API(web3)
端点: 端点:
POST /api/ai/regex body: {desc: str} → {ok, regex, note} POST /api/ai/rule body: {desc, rule_type} → {ok, code, note}
POST /api/regex/test body: {regex, value} → {ok, match, error, groups} POST /api/rule/test body: {rule_type, code, value} → {ok, pass_, error}
「测试」端点走 Python re(不依赖前端 JS RegExp),保证 LLM 生成的 Python 正则 兼容老端点(不破坏 2026-08-20 接入的 AI 正则流程):
能被同一种语义验证,避免 JS/Python 正则方言差异导致误判。 POST /api/ai/regex body: {desc} → {ok, regex, note} (rule_type 默认为 regex)
POST /api/regex/test body: {regex, value} → {ok, match, error} (rule_type 默认为 regex)
「测试」端点统一走 Python(regex 走 re.compile、number/date 走 rule_runner._exec_user_function),
保证和检测引擎语义一致。
""" """
from __future__ import annotations from __future__ import annotations
...@@ -13,70 +17,126 @@ from __future__ import annotations ...@@ -13,70 +17,126 @@ from __future__ import annotations
import time import time
from fastapi import APIRouter from fastapi import APIRouter
from pydantic import BaseModel, Field from pydantic import BaseModel, ConfigDict, Field
from web3.backend.core.ai_regex import gen_regex, test_regex from web3.backend.core.ai_rule import gen_rule, test_rule
from web3.backend.core.ai_regex import test_regex as _legacy_test_regex
from web3.backend.models.rule import DEFAULT_RULE_TYPE, RULE_TYPES
from web3.backend._logging import get_logger from web3.backend._logging import get_logger
router = APIRouter(prefix="", tags=["ai"]) router = APIRouter(prefix="", tags=["ai"])
logger = get_logger("backend.routers.ai") logger = get_logger("backend.routers.ai")
# ── /ai/regex ───────────────────────────────────────────── # ── /ai/rule ─────────────────────────────────────────────
class GenRegexRequest(BaseModel): class GenRuleRequest(BaseModel):
desc: str = Field(..., min_length=1, description="规则自然语言描述") desc: str = Field(..., min_length=1, description="规则自然语言描述")
rule_type: str = Field(
DEFAULT_RULE_TYPE,
description=f"规则类型:{ ' / '.join(RULE_TYPES) }",
)
class GenRegexResponse(BaseModel): class GenRuleResponse(BaseModel):
ok: bool ok: bool
regex: str = "" code: str = "" # regex: 正则字符串;number/date: Python 函数源码
note: str = "" note: str = ""
@router.post("/ai/regex", response_model=GenRegexResponse, summary="AI 生成 Python 正则") @router.post("/ai/rule", response_model=GenRuleResponse, summary="AI 生成规则代码(按 rule_type 分流)")
async def ai_gen_regex(req: GenRegexRequest): async def ai_gen_rule(req: GenRuleRequest):
"""前端「规则设置」弹窗的「AI 生成正则」按钮调用。 """前端「规则设置」弹窗的「AI 生成」按钮调用。
返回 ok=true 表示 regex 非空;ok=false 表示降级(无 key / 描述不清 / 5 次重试全败)。 rule_type=regex:返回单行正则;其他:返回 Python 函数源码。
""" """
logger.info("─" * 60) logger.info("─" * 60)
logger.info(f"POST /api/ai/regex desc={req.desc!r}") logger.info(f"POST /api/ai/rule rule_type={req.rule_type!r} desc={req.desc!r}")
started = time.monotonic() started = time.monotonic()
try: try:
regex, note = gen_regex(req.desc) code, note = gen_rule(req.desc, req.rule_type)
except Exception as e: except Exception as e:
elapsed_ms = (time.monotonic() - started) * 1000 elapsed_ms = (time.monotonic() - started) * 1000
logger.exception(f"❌ AI 正则服务抛出异常(耗时 {elapsed_ms:.0f}ms)") logger.exception(f"❌ AI 规则服务抛出异常(耗时 {elapsed_ms:.0f}ms)")
logger.info("─" * 60) logger.info("─" * 60)
return GenRegexResponse(ok=False, regex="", note=f"AI 正则服务异常: {type(e).__name__}: {e}") return GenRuleResponse(ok=False, code="", note=f"AI 规则服务异常: {type(e).__name__}: {e}")
elapsed_ms = (time.monotonic() - started) * 1000 elapsed_ms = (time.monotonic() - started) * 1000
if regex: if code:
logger.info(f"✅ AI 正则生成成功(耗时 {elapsed_ms:.0f}ms)regex={regex!r}") logger.info(
f"✅ AI 规则生成成功(耗时 {elapsed_ms:.0f}ms)"
f"rule_type={req.rule_type!r} code_len={len(code)}"
)
else: else:
logger.warning(f"❌ AI 正则生成失败(耗时 {elapsed_ms:.0f}ms)note={note!r}") logger.warning(f"❌ AI 规则生成失败(耗时 {elapsed_ms:.0f}ms)note={note!r}")
logger.info("─" * 60) logger.info("─" * 60)
return GenRegexResponse(ok=bool(regex), regex=regex, note=note) return GenRuleResponse(ok=bool(code), code=code, note=note)
# ── /regex/test ─────────────────────────────────────────── # ── /rule/test ───────────────────────────────────────────
class TestRegexRequest(BaseModel): class TestRuleRequest(BaseModel):
regex: str = Field(..., min_length=1, description="要测试的正则表达式") rule_type: str = Field(DEFAULT_RULE_TYPE, description=f"规则类型:{ ' / '.join(RULE_TYPES) }")
code: str = Field(..., min_length=1, description="regex 时是正则;number/date 时是 Python 函数源码")
value: str = Field("", description="测试值") value: str = Field("", description="测试值")
class TestRegexResponse(BaseModel): class TestRuleResponse(BaseModel):
ok: bool # 正则编译是否成功 model_config = ConfigDict(populate_by_name=True) # 同时支持 pass_ 和 alias "pass"
match: bool # value 是否匹配
error: str = "" # 编译/匹配错误信息
groups: list[str] = [] # 捕获组内容(无捕获组或未匹配时空)
ok: bool # 代码/正则本身是否合法
pass_: bool = Field(False, alias="pass", description="值是否通过")
error: str = ""
@router.post("/regex/test", response_model=TestRegexResponse, summary="用 Python re 校验 value 是否匹配 regex")
async def ai_test_regex(req: TestRegexRequest): @router.post("/rule/test", response_model=TestRuleResponse, summary="按 rule_type 校验 value")
async def ai_test_rule(req: TestRuleRequest):
"""前端「规则设置」弹窗的「测试」按钮调用。 """前端「规则设置」弹窗的「测试」按钮调用。
用 Python re 校验(不是 JS RegExp),保证和后端检测引擎语义一致。 统一走 rule_runner.run_rule(),保证和检测引擎语义一致。
""" """
result = test_regex(req.regex, req.value) result = test_rule(req.rule_type, req.code, req.value)
return TestRegexResponse(**result) return TestRuleResponse(
\ No newline at end of file ok=result.get("ok", False),
pass_=result.get("pass_", False),
error=result.get("error", ""),
)
# ── 兼容老端点 ─────────────────────────────────────────────
class GenRegexRequest(BaseModel):
desc: str = Field(..., min_length=1, description="规则自然语言描述")
class GenRegexResponse(BaseModel):
ok: bool
regex: str = ""
note: str = ""
@router.post("/ai/regex", response_model=GenRegexResponse, summary="[兼容] AI 生成 Python 正则")
async def ai_gen_regex(req: GenRegexRequest):
"""老端点,转发到 /ai/rule(rule_type='regex')。
保留原因:web3/src/api/ai.js 之前的 genRegex 调用 + 老测试(test_run_query_only_bad_rows
没用到,但 ai.js 单元测试可能用到)继续可用。
"""
logger.info(f"POST /api/ai/regex (legacy) desc={req.desc!r}")
code, note = gen_rule(req.desc, "regex")
return GenRegexResponse(ok=bool(code), regex=code, note=note)
class TestRegexRequest(BaseModel):
regex: str = Field(..., min_length=1)
value: str = Field("", description="测试值")
class TestRegexResponse(BaseModel):
ok: bool
match: bool
error: str = ""
groups: list[str] = []
@router.post("/regex/test", response_model=TestRegexResponse, summary="[兼容] 用 Python re 校验 value")
async def ai_test_regex(req: TestRegexRequest):
"""老端点,转发到 _legacy_test_regex(保持 capture groups 输出)。"""
return TestRegexResponse(**_legacy_test_regex(req.regex, req.value))
\ No newline at end of file
...@@ -10,7 +10,9 @@ ...@@ -10,7 +10,9 @@
SELECT 的列 = 「有规则的字段」∪「show_default 的字段」, SELECT 的列 = 「有规则的字段」∪「show_default 的字段」,
后者没有规则也要拉,否则结果表这一列全空 后者没有规则也要拉,否则结果表这一列全空
- 列名小写归一(达梦默认大写,MySQL 视 collation)→ 与 field_key 对齐 - 列名小写归一(达梦默认大写,MySQL 视 collation)→ 与 field_key 对齐
- 正则校验:rule.regex 为空时回退到 `^.+$`(非空校验) - 校验引擎:rule_runner.run_rule() 按 rule_type dispatch
- regex : re.compile(regex).search(value),空 regex 回退到 `^.+$`(非空校验)
- number/date : 用户提供的 def check(value) -> bool,安全沙箱 exec
判定口径(只返回「不合规」的行): 判定口径(只返回「不合规」的行):
- 字段级:该字段的多条规则是 AND —— 任一条不满足,该字段即不合规, - 字段级:该字段的多条规则是 AND —— 任一条不满足,该字段即不合规,
...@@ -36,6 +38,7 @@ from sqlalchemy.orm import Session ...@@ -36,6 +38,7 @@ from sqlalchemy.orm import Session
from web3.backend._logging import get_logger from web3.backend._logging import get_logger
from web3.backend.core.db_adapter import DBConfig, DBConnection, quote_ident from web3.backend.core.db_adapter import DBConfig, DBConnection, quote_ident
from web3.backend.core.rule_runner import RuleRunError, run_rule
from web3.backend.db.database import get_session from web3.backend.db.database import get_session
from web3.backend.models.field import Field from web3.backend.models.field import Field
from web3.backend.models.rule import Rule from web3.backend.models.rule import Rule
...@@ -105,7 +108,13 @@ async def run_query(req: RunQueryRequest, db: Session = Depends(get_session)): ...@@ -105,7 +108,13 @@ async def run_query(req: RunQueryRequest, db: Session = Depends(get_session)):
"ord": f.ord, "ord": f.ord,
"rules": len(rules_by_field[f.id]), "rules": len(rules_by_field[f.id]),
"rule_list": [ "rule_list": [
{"id": r.id, "desc": r.desc, "regex": r.regex} {
"id": r.id,
"desc": r.desc,
"rule_type": r.rule_type or "regex",
"regex": r.regex,
"code": r.code,
}
for r in rules_by_field[f.id] for r in rules_by_field[f.id]
], ],
} }
...@@ -143,18 +152,26 @@ async def run_query(req: RunQueryRequest, db: Session = Depends(get_session)): ...@@ -143,18 +152,26 @@ async def run_query(req: RunQueryRequest, db: Session = Depends(get_session)):
oracle_client_dir=conn.get("oracleClientDir"), oracle_client_dir=conn.get("oracleClientDir"),
) )
# 5) 编译正则(提前编译,错的话直接返回不要连库) # 5) 校验引擎:按 rule_type 准备执行器
# - regex:提前 re.compile,失败标 None("坏规则")
# - number/date:用户函数源码直接传,run_rule 时走沙箱 + validate_user_code 兜底
compiled: list[tuple[Field, list[tuple[Rule, Optional[re.Pattern]]]]] = [] compiled: list[tuple[Field, list[tuple[Rule, Optional[re.Pattern]]]]] = []
for f, rules in fields_with_rules: for f, rules in fields_with_rules:
rule_patterns: list[tuple[Rule, Optional[re.Pattern]]] = [] rule_patterns: list[tuple[Rule, Optional[re.Pattern]]] = []
for r in rules: for r in rules:
regex_src = (r.regex or "").strip() or r"^.+$" # 空 regex = 非空校验 rt = r.rule_type or "regex"
try: if rt == "regex":
rule_patterns.append((r, re.compile(regex_src))) regex_src = (r.regex or "").strip() or r"^.+$" # 空 regex = 非空校验
except re.error as e: try:
# 单条规则编译失败:记一行 issue,整字段标红 rule_patterns.append((r, re.compile(regex_src)))
logger.warning(f"[queries] 规则 id={r.id} regex 编译失败:{e}") except re.error as e:
rule_patterns.append((r, None)) # None 标记"坏规则" logger.warning(f"[queries] 规则 id={r.id} regex 编译失败:{e}")
rule_patterns.append((r, None)) # None 标记"坏规则"
else:
# number/date:编译阶段不校验代码(每行要 exec 一次太慢),
# 真正跑到这行时再让沙箱兜底;编译阶段先标个非 None 占位,
# 让「坏规则」分支只在执行时触发
rule_patterns.append((r, re.compile(r"^.*$"))) # 占位正则,永远 match
compiled.append((f, rule_patterns)) compiled.append((f, rule_patterns))
# 6) 连库 + 拉数据 # 6) 连库 + 拉数据
...@@ -218,18 +235,33 @@ async def run_query(req: RunQueryRequest, db: Session = Depends(get_session)): ...@@ -218,18 +235,33 @@ async def run_query(req: RunQueryRequest, db: Session = Depends(get_session)):
error_cells: list[str] = [] error_cells: list[str] = []
for f, rule_patterns in compiled: for f, rule_patterns in compiled:
val = raw_lc.get(f.field_key.lower()) val = raw_lc.get(f.field_key.lower())
val_str = "" if val is None else str(val)
for rule, pat in rule_patterns: for rule, pat in rule_patterns:
if pat is None: rt = rule.rule_type or "regex"
# 坏规则:正则本身编译不过,按不合规处理并把原因暴露出来 rule_passed: Optional[bool] = None
issues.append({"field": f.field_key, "desc": f"规则正则编译失败:{rule.desc}"}) fail_desc: Optional[str] = None
elif pat.search(val_str) is None: if rt == "regex":
# re.search 没匹配上 → 这条规则不满足 if pat is None:
issues.append({"field": f.field_key, "desc": rule.desc}) # 坏规则:正则本身编译不过,按不合规处理并把原因暴露出来
rule_passed = False
fail_desc = f"规则正则编译失败:{rule.desc}"
else:
val_str = "" if val is None else str(val)
rule_passed = pat.search(val_str) is not None
if not rule_passed:
fail_desc = rule.desc
else: else:
continue # 这条规则通过,看下一条 # number / date:走沙箱执行用户函数
if f.field_key not in error_cells: try:
error_cells.append(f.field_key) rule_passed = run_rule(rt, None, rule.code, val)
if not rule_passed:
fail_desc = rule.desc
except RuleRunError as e:
rule_passed = False
fail_desc = f"规则执行失败:{e}"
if rule_passed is False:
issues.append({"field": f.field_key, "desc": fail_desc})
if f.field_key not in error_cells:
error_cells.append(f.field_key)
if not issues: if not issues:
continue # 全部规则都通过 → 合规行,不返回 continue # 全部规则都通过 → 合规行,不返回
bad_count += 1 bad_count += 1
......
...@@ -39,7 +39,12 @@ router = APIRouter(prefix="/tasks", tags=["tasks"]) ...@@ -39,7 +39,12 @@ router = APIRouter(prefix="/tasks", tags=["tasks"])
class RulePayload(BaseModel): class RulePayload(BaseModel):
desc: str = PydField(..., min_length=1, description="规则说明(自然语言)") desc: str = PydField(..., min_length=1, description="规则说明(自然语言)")
regex: Optional[str] = PydField(None, description="Python 正则表达式(可空)") rule_type: str = PydField(
"regex",
description="规则类型:regex / number / date(2026-08-21 扩展)",
)
regex: Optional[str] = PydField(None, description="Python 正则表达式(regex 类型专用,可空)")
code: Optional[str] = PydField(None, description="Python 函数源码(number/date 类型专用)")
class FieldPayload(BaseModel): class FieldPayload(BaseModel):
...@@ -87,6 +92,8 @@ class RuleOut(BaseModel): ...@@ -87,6 +92,8 @@ class RuleOut(BaseModel):
id: int id: int
desc: str desc: str
regex: Optional[str] regex: Optional[str]
rule_type: str = "regex"
code: Optional[str] = None
ord: int ord: int
...@@ -159,10 +166,15 @@ def _replace_fields(db: Session, task: Task, fields_payload: list[FieldPayload]) ...@@ -159,10 +166,15 @@ def _replace_fields(db: Session, task: Task, fields_payload: list[FieldPayload])
db.add(field) db.add(field)
db.flush() # 拿到 field.id db.flush() # 拿到 field.id
for r_idx, r in enumerate(f.rules): for r_idx, r in enumerate(f.rules):
rt = (r.rule_type or "regex").strip() or "regex"
regex_val = (r.regex or "").strip() or None if rt == "regex" else None
code_val = (r.code or "").strip() or None if rt in ("number", "date") else None
rule = Rule( rule = Rule(
field_id=field.id, field_id=field.id,
desc=r.desc.strip(), desc=r.desc.strip(),
regex=(r.regex or "").strip() or None, regex=regex_val,
rule_type=rt,
code=code_val,
ord=r_idx, ord=r_idx,
) )
db.add(rule) db.add(rule)
...@@ -187,6 +199,8 @@ def _copy_fields(db: Session, src_task: Task, dst_task: Task) -> None: ...@@ -187,6 +199,8 @@ def _copy_fields(db: Session, src_task: Task, dst_task: Task) -> None:
field_id=new_f.id, field_id=new_f.id,
desc=src_r.desc, desc=src_r.desc,
regex=src_r.regex, regex=src_r.regex,
rule_type=src_r.rule_type or "regex",
code=src_r.code,
ord=r_idx, ord=r_idx,
)) ))
...@@ -212,7 +226,12 @@ def _row_to_out(db: Session, t: Task, *, include_field_list: bool = False) -> Ta ...@@ -212,7 +226,12 @@ def _row_to_out(db: Session, t: Task, *, include_field_list: bool = False) -> Ta
if include_field_list: if include_field_list:
for f in db.query(Field).filter(Field.task_id == t.id).order_by(Field.ord).all(): for f in db.query(Field).filter(Field.task_id == t.id).order_by(Field.ord).all():
rules = [ rules = [
RuleOut(id=r.id, desc=r.desc, regex=r.regex, ord=r.ord) RuleOut(
id=r.id, desc=r.desc, regex=r.regex,
rule_type=r.rule_type or "regex",
code=r.code,
ord=r.ord,
)
for r in db.query(Rule).filter(Rule.field_id == f.id).order_by(Rule.ord).all() for r in db.query(Rule).filter(Rule.field_id == f.id).order_by(Rule.ord).all()
] ]
field_list.append(FieldOut( field_list.append(FieldOut(
......
/** /**
* AI 相关 API(2026-08-20 接入) * AI 相关 API(2026-08-21 扩展规则类型)
* *
* - genRegex(desc) → POST /api/ai/regex 后端走 LLM(强制,无规则库)+ re.compile 校验 * 三类规则共用一对接口:
* - testRegex(...) → POST /api/regex/test 后端用 Python re 校验(不是 JS RegExp) * - genRule(desc, ruleType) → POST /api/ai/rule 后端走 LLM(regex 走老 ai_regex;number/date 走 ai_rule)
* - testRule(ruleType, code, value) → POST /api/rule/test
*
* ruleType: 'regex' | 'number' | 'date'
* - regex: code = 单行正则字符串
* - number: code = Python 函数源码(含 def check(value) -> bool)
* - date: code = 同 number
*
* 老端点 genRegex / testRegex 保留为薄壳(转发到新接口),兼容 2026-08-20 的 AI 正则流程。
*/ */
import { http } from './client' import { http } from './client'
export const RULE_TYPES = ['regex', 'number', 'date']
/** /**
* AI 生成正则表达式 * AI 生成规则代码
* @param {string} desc 规则说明(自然语言) * @param {string} desc 规则说明(自然语言)
* @returns { ok: boolean, regex: string, note: string } * @param {string} ruleType 'regex' | 'number' | 'date'
* - ok=true:regex 非空,可直接填到「正则表达式」输入框 * @returns { ok: boolean, code: string, note: string }
* - ok=false:regex 空字符串;note 是失败原因(前端展示) * - ok=true:code 非空,可直接填到对应输入框
* - ok=false:code 空字符串;note 是失败原因(前端展示)
*/
export function genRule(desc, ruleType = 'regex') {
return http.post('/ai/rule', { desc, rule_type: ruleType })
}
/**
* 用后端引擎校验 value 是否通过
* @param {string} ruleType
* @param {string} code regex → 正则;number/date → Python 函数源码
* @param {string} value 测试值
* @returns { ok: boolean, pass: boolean, error: string }
* - ok false = 代码/正则本身非法;error 是报错信息
* - ok true + pass true/false = 是否通过
*/
export function testRule(ruleType, code, value) {
return http.post('/rule/test', { rule_type: ruleType, code, value })
}
/**
* [兼容] AI 生成正则 —— 转发到 genRule('regex')
*/ */
export function genRegex(desc) { export function genRegex(desc) {
return http.post('/ai/regex', { desc }) return genRule(desc, 'regex').then((r) => ({ ok: r.ok, regex: r.code, note: r.note }))
} }
/** /**
* 用 Python re 校验 value 是否匹配 regex * [兼容] 用 Python re 校验 value —— 转发到 testRule('regex')
* @param {string} regex * 老响应里有 groups 数组(捕获组),新接口不再返回;这里也模拟空数组保持兼容
* @param {string} value
* @returns { ok: boolean, match: boolean, error: string, groups: string[] }
* - ok false = 正则编译失败;error 是 Python 报错(不是 JS 报错)
* - ok true + match true/false = 是否命中
* - groups = 捕获组内容(无捕获组或未匹配时 [])
*/ */
export function testRegex(regex, value) { export function testRegex(regex, value) {
return http.post('/regex/test', { regex, value }) return testRule('regex', regex, value).then((r) => ({
ok: r.ok,
match: r.pass,
error: r.error,
groups: [],
}))
} }
\ No newline at end of file
"""
验证:core/rule_runner.py 的安全沙箱 + 按 rule_type dispatch(2026-08-21 扩展)。
覆盖:
- regex : re.compile,空 regex 回退 `^.+$`,坏正则抛 RuleRunError
- number : 用户 def check(value) -> bool,沙箱内执行
- date : 同 number
- 沙箱安全性:__import__ / open / eval / exec / getattr / globals 都不可用
- 错误处理:缺 check / 语法错 / 运行时异常都包成 RuleRunError
跑:python -m pytest web3/tests/test_rule_runner.py -v
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
import pytest
from web3.backend.core.rule_runner import (
RuleRunError,
run_rule,
validate_user_code,
)
from web3.backend.models.rule import RULE_TYPES
# ── regex ───────────────────────────────────────────────
class TestRegex:
def test_simple_match(self):
assert run_rule("regex", r"^1\d{10}$", None, "13800138000") is True
assert run_rule("regex", r"^1\d{10}$", None, "abc") is False
def test_empty_regex_means_non_empty(self):
# 空 regex 回退 `^.+$` → 非空校验
assert run_rule("regex", "", None, "anything") is True
assert run_rule("regex", None, None, "") is False
assert run_rule("regex", None, None, None) is False
def test_bad_regex_raises(self):
with pytest.raises(RuleRunError, match="正则编译失败"):
run_rule("regex", r"[unclosed", None, "abc")
# ── number ──────────────────────────────────────────────
class TestNumber:
def test_check_returns_true(self):
code = (
"def check(value):\n"
" n = float(value)\n"
" return 0 <= n <= 150\n"
)
assert run_rule("number", None, code, "30") is True
assert run_rule("number", None, code, "200") is False
def test_check_handles_invalid_input(self):
code = (
"def check(value):\n"
" try:\n"
" return float(value) > 0\n"
" except Exception:\n"
" return False\n"
)
assert run_rule("number", None, code, "abc") is False
assert run_rule("number", None, code, "1.5") is True
def test_missing_check_raises(self):
with pytest.raises(RuleRunError, match="必须定义 check"):
run_rule("number", None, "x = 1\n", "5")
def test_syntax_error_raises(self):
with pytest.raises(RuleRunError, match="语法错误"):
run_rule("number", None, "def check(value:\n return True\n", "5")
def test_runtime_error_raises(self):
code = "def check(value):\n return 1 / 0\n"
with pytest.raises(RuleRunError, match="check"):
run_rule("number", None, code, "5")
# ── date ────────────────────────────────────────────────
class TestDate:
def test_strptime_format_check(self):
code = (
"def check(value):\n"
" try:\n"
" datetime.datetime.strptime(value, '%Y-%m-%d')\n"
" return True\n"
" except Exception:\n"
" return False\n"
)
assert run_rule("date", None, code, "2024-01-15") is True
assert run_rule("date", None, code, "2024/01/15") is False
assert run_rule("date", None, code, "not-a-date") is False
# ── 沙箱安全 ────────────────────────────────────────────
class TestSandbox:
def test_import_os_blocked(self):
code = (
"import os\n"
"def check(value):\n"
" return True\n"
)
# exec 应抛错(__import__ 不在白名单)
with pytest.raises(RuleRunError):
run_rule("number", None, code, "1")
def test_import_datetime_works_via_preload(self):
# datetime 是预注入的,用户代码可以直接用,不写 import 也能跑
code = (
"def check(value):\n"
" return datetime.datetime.now().year > 2020\n"
)
assert run_rule("number", None, code, "1") is True
def test_open_blocked(self):
code = (
"def check(value):\n"
" open('/etc/passwd')\n"
" return True\n"
)
# check 内部调 open() 触发 NameError(open 不在白名单)
with pytest.raises(RuleRunError, match="check"):
run_rule("number", None, code, "1")
def test_eval_blocked(self):
code = (
"def check(value):\n"
" return eval('1+1') == 2\n"
)
with pytest.raises(RuleRunError, match="check"):
run_rule("number", None, code, "1")
def test_getattr_blocked(self):
# getattr 不在白名单 → check 内调用 NameError
code = (
"def check(value):\n"
" return getattr(int, 'x', None) is None\n"
)
with pytest.raises(RuleRunError, match="check"):
run_rule("number", None, code, "1")
# ── validate_user_code ──────────────────────────────────
class TestValidateUserCode:
def test_valid_code(self):
ok, err = validate_user_code("def check(value):\n return True\n")
assert ok is True
assert err == ""
def test_empty_code(self):
ok, err = validate_user_code("")
assert ok is False
assert "为空" in err
def test_no_check_defined(self):
ok, err = validate_user_code("x = 1\n")
assert ok is False
assert "check" in err
def test_syntax_error(self):
ok, err = validate_user_code("def check(value:\n return True\n")
assert ok is False
assert "语法" in err
# ── RULE_TYPES 常量 ─────────────────────────────────────
def test_rule_types_constant():
assert RULE_TYPES == ("regex", "number", "date")
# ── 未知类型 ────────────────────────────────────────────
def test_unknown_rule_type_raises():
with pytest.raises(RuleRunError, match="未知的规则类型"):
run_rule("weird", None, None, "x")
\ 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