From e1d63e73893d8d962778a7cd2bb9cd96b6798584 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 3 Nov 2026 20:50:06 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=84=E5=88=99=20DSL=20=E7=BC=96=E8=AF=91?= =?UTF-8?q?=E6=88=90=E5=8F=AA=E8=AF=BB=20SQL=EF=BC=8C=E5=B9=B6=E7=94=A8=20?= =?UTF-8?q?fixture=20=E9=AA=8C=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/ledger.py | 21 +++++ backend/app/data/rule_compile.py | 130 +++++++++++++++++++++++++++++ backend/tests/test_rule_compile.py | 58 +++++++++++++ 3 files changed, 209 insertions(+) create mode 100644 backend/app/data/rule_compile.py create mode 100644 backend/tests/test_rule_compile.py diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index fecb6a8d..f349c1d1 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -286,6 +286,27 @@ async def caliber_descriptor(workflow_id: str, version: int, node_id: str, return made +class RuleCompileIn(BaseModel): + """规则 DSL 编译：表 + 结构化条件 → 只读 SQL；fixtures 可选，用内存 SQLite 跑一遍。""" + + rule: dict[str, Any] + table: str + fixtures: list[dict[str, Any]] = Field(default_factory=list) + + +@router.post("/rules/compile") +async def compile_rule_endpoint(payload: RuleCompileIn) -> dict[str, Any]: + """把结构化规则编译成 SQL，并（给了 fixtures 时）用样例行验证。只生成 SELECT，不碰业务库。""" + from app.data.rule_compile import RuleInvalid, compile_rule, run_fixtures + + try: + sql = compile_rule(payload.rule, payload.table) + checks = run_fixtures(payload.rule, payload.table, payload.fixtures) + except RuleInvalid as e: + raise HTTPException(422, str(e)) from None + return {"sql": sql, "fixtures": checks} + + class TrialIn(BaseModel): """试跑只收入口字段的值；没写的按口径里的缺省值补。""" diff --git a/backend/app/data/rule_compile.py b/backend/app/data/rule_compile.py new file mode 100644 index 00000000..36298947 --- /dev/null +++ b/backend/app/data/rule_compile.py @@ -0,0 +1,130 @@ +"""规则 DSL 的编译与 fixture 校验：结构化条件 → 一条只读 SQL，发布前用样例行跑一遍。 + +和口径卡里的 `kind=rule` 不是同一层：口径卡里的规则在单元格上确定性求值（不需要查库）；这里给作者和发布检查用， +把「表 + 条件 → then / else」编译成 SQL，再用 fixture 验证条件写对了。编译只生成 SELECT，不执行任何写操作。 + +规则形状： + { + "inputs": {"status": "status", "amount": "amount"}, # 名字 → 表里的列 + "all": [{"input": "status", "op": "eq", "value": "paid"}, + {"input": "amount", "op": "gte", "value": 50}], + "then": 1, "else": 0 + } +""" +from __future__ import annotations + +import re +import sqlite3 +from typing import Any, Iterable, Mapping + +OPS = ("eq", "ne", "gt", "gte", "lt", "lte", "in", "not_in") +_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +class RuleInvalid(ValueError): + """规则本身写错了：message 给人看。""" + + +def _ident(value: Any, what: str) -> str: + text = str(value or "").strip() + if not _IDENT.match(text): + raise RuleInvalid(f"{what}「{value}」不是合法的标识符（字母或下划线开头，只含字母、数字、下划线）") + return text + + +def _literal(exp: Any, value: Any) -> Any: + if isinstance(value, bool): + return exp.Boolean(this=value) + if isinstance(value, (int, float)): + return exp.Literal.number(value) + if value is None: + return exp.Null() + return exp.Literal.string(str(value)) + + +def _condition(exp: Any, cond: Mapping[str, Any], inputs: Mapping[str, str]) -> Any: + name = str(cond.get("input") or "") + if name not in inputs: + raise RuleInvalid(f"条件指向了没声明的输入「{name}」") + column = exp.column(_ident(inputs[name], "列名")) + op = str(cond.get("op") or "") + if op not in OPS: + raise RuleInvalid(f"条件运算符只能是 {'、'.join(OPS)}，当前为「{op}」") + if op in ("in", "not_in"): + values = cond.get("values") + if not isinstance(values, list) or not values: + raise RuleInvalid("in / not_in 需要非空的 values 列表") + node = exp.In(this=column, expressions=[_literal(exp, v) for v in values]) + return node if op == "in" else exp.Not(this=node) + if "value" not in cond: + raise RuleInvalid("条件需要 value") + right = _literal(exp, cond["value"]) + return { + "eq": exp.EQ, "ne": exp.NEQ, "gt": exp.GT, "gte": exp.GTE, "lt": exp.LT, "lte": exp.LTE, + }[op](this=column, expression=right) + + +def compile_rule(rule: Mapping[str, Any], table: str, *, dialect: str | None = None) -> str: + """结构化规则 → `SELECT CASE WHEN … THEN … ELSE … END AS value FROM `。""" + try: + import sqlglot + from sqlglot import exp + except ImportError as e: # pragma: no cover - 依赖缺失时给一句人话 + raise RuleInvalid("没有安装 sqlglot，无法编译规则") from e + table_name = _ident(table, "表名") + if not isinstance(rule, Mapping): + raise RuleInvalid("规则需要是一个对象") + inputs = rule.get("inputs") + if not isinstance(inputs, Mapping) or not inputs: + raise RuleInvalid("规则需要 inputs：名字 → 表里的列") + columns = {str(k): _ident(v, "列名") for k, v in inputs.items()} + conditions = rule.get("all") + if not isinstance(conditions, list) or not conditions: + raise RuleInvalid("规则需要 all：至少一条条件") + where = None + for cond in conditions: + node = _condition(exp, cond, columns) + where = node if where is None else exp.And(this=where, expression=node) + case = exp.Case( + ifs=[exp.If(this=where, true=_literal(exp, rule.get("then", 1)))], + default=_literal(exp, rule.get("else", 0)), + ) + query = exp.select(exp.alias_(case, "value")).from_(exp.table_(table_name)) + return query.sql(dialect=dialect or "sqlite") + + +def run_fixtures(rule: Mapping[str, Any], table: str, fixtures: Iterable[Mapping[str, Any]], *, + dialect: str = "sqlite") -> list[dict[str, Any]]: + """用一条只读 SQL 在内存 SQLite 上跑 fixture：每项 {rows, expected}，返回 [{ok, expected, actual}]。 + + fixture 的 rows 是列名 → 值；表结构按第一行推（整数 / 实数 / 文本）。空 fixture 返回空列表。 + """ + items = [dict(f) for f in fixtures] + if not items: + return [] + sql = compile_rule(rule, table, dialect="sqlite" if dialect != "sqlite" else dialect) + db = sqlite3.connect(":memory:") + try: + columns: dict[str, str] = {} + for item in items: + for row in item.get("rows") or []: + for name, value in row.items(): + columns.setdefault(str(name), "INTEGER" if isinstance(value, int) and not isinstance(value, bool) + else "REAL" if isinstance(value, float) else "TEXT") + if not columns: + raise RuleInvalid("fixture 需要至少一行 rows") + db.execute(f"CREATE TABLE {table} (" + ", ".join(f"{name} {kind}" for name, kind in columns.items()) + ")") + out: list[dict[str, Any]] = [] + for item in items: + db.execute(f"DELETE FROM {table}") + for row in item.get("rows") or []: + db.execute(f"INSERT INTO {table} ({', '.join(row)}) VALUES ({', '.join('?' for _ in row)})", + list(row.values())) + actual = db.execute(sql).fetchone()[0] + out.append({"ok": actual == item.get("expected"), "expected": item.get("expected"), "actual": actual}) + return out + finally: + db.close() + + +__all__ = ["OPS", "RuleInvalid", "compile_rule", "run_fixtures"] diff --git a/backend/tests/test_rule_compile.py b/backend/tests/test_rule_compile.py new file mode 100644 index 00000000..ec3257bd --- /dev/null +++ b/backend/tests/test_rule_compile.py @@ -0,0 +1,58 @@ +"""规则 DSL 编译（app/data/rule_compile.py、POST /api/ledger/rules/compile）：结构化条件 → 只读 SQL，fixture 验证。""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.data.rule_compile import RuleInvalid, compile_rule, run_fixtures +from app.main import app + +RULE = {"inputs": {"status": "status", "amount": "amount"}, + "all": [{"input": "status", "op": "eq", "value": "paid"}, + {"input": "amount", "op": "gte", "value": 50}], + "then": 1, "else": 0} + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def test_compile_rule_is_a_select_with_a_case(): + sql = compile_rule(RULE, "orders") + assert sql.lower().startswith("select") and "case when" in sql.lower() + assert "from orders" in sql.lower() + assert "insert" not in sql.lower() and "update" not in sql.lower() and "delete" not in sql.lower() + + +def test_fixtures_run_against_the_compiled_sql(): + checks = run_fixtures(RULE, "orders", [ + {"rows": [{"status": "paid", "amount": 100}], "expected": 1}, + {"rows": [{"status": "paid", "amount": 10}], "expected": 0}, + {"rows": [{"status": "open", "amount": 100}], "expected": 0}, + ]) + assert [c["ok"] for c in checks] == [True, True, True] + + +@pytest.mark.parametrize("rule", [ + {"inputs": {}, "all": [{"input": "x", "op": "eq", "value": 1}]}, + {"inputs": {"x": "x"}, "all": [{"input": "x", "op": "like", "value": 1}]}, + {"inputs": {"x": "x"}, "all": [{"input": "y", "op": "eq", "value": 1}]}, +]) +def test_bad_rules_are_refused(rule): + with pytest.raises(RuleInvalid): + compile_rule(rule, "orders") + + +async def test_the_compile_endpoint_returns_sql_and_fixture_results(client): + res = await client.post("/api/ledger/rules/compile", json={ + "rule": RULE, "table": "orders", + "fixtures": [{"rows": [{"status": "paid", "amount": 100}], "expected": 1}, + {"rows": [{"status": "paid", "amount": 10}], "expected": 0}], + }) + assert res.status_code == 200, res.text + body = res.json() + assert body["sql"].lower().startswith("select") and [c["ok"] for c in body["fixtures"]] == [True, True] + bad = await client.post("/api/ledger/rules/compile", json={"rule": {"inputs": {}}, "table": "orders"}) + assert bad.status_code == 422