feat: P0-1 价格脚本引擎(安全公式解析+脚本CRUD+试算+报价集成)

src/huamenchuang/api/quote.py

@@ -16,6 +16,8 @@ class QuoteRequest(BaseModel):
     order_id: int | None = Field(default=None, description="或对已有订单报价")
     prices: dict | None = Field(default=None, description="价格本覆盖 {frame,mullion,sash,glass,hardware}")
     use_price_book: bool = Field(default=False, description="从产品管理价格本取价(按材料类别均价)")
+    use_scripts: bool = Field(default=False, description="启用价格脚本计价(公司启用脚本覆盖默认单价/总价)")
+    scripts: dict | None = Field(default=None, description="临时脚本 {target: formula}, 优先于脚本库")
 
 
 async def _resolve_drawing(db, body: QuoteRequest) -> str | dict | None:
@@ -56,15 +58,25 @@ async def _price_book(db, company_id: int) -> dict | None:
 
 @router.post("")
 async def offer_price(body: QuoteRequest, user: CurrentUser, db: DbSession):
-    """算料报价：型材清单 + 玻璃面积 + 五金 + 总价。"""
+    """算料报价：型材清单 + 玻璃面积 + 五金 + 总价（可选价格脚本计价）。"""
     drawing = await _resolve_drawing(db, body)
     if drawing is None:
         return fail("未找到图纸数据")
     prices = body.prices
     if prices is None and body.use_price_book:
         prices = await _price_book(db, user.company_id)
+    # 脚本解析：显式传入的临时脚本优先，其次取脚本库中启用脚本
+    scripts = body.scripts
+    if scripts is None and body.use_scripts:
+        from sqlalchemy import select as _select
+        from huamenchuang.api.script import load_script_map
+        from huamenchuang.models.business import PriceScript
+        rows = list((await db.execute(
+            _select(PriceScript).where(PriceScript.company_id == user.company_id))).scalars())
+        scripts = load_script_map(rows) or None
     try:
-        result = calc_quote(drawing, prices)
-    except Exception:
-        return fail("图纸数据解析失败")
+        result = calc_quote(drawing, prices, scripts)
+    except Exception as e:  # 公式/图纸解析错误 → 用户可读失败
+        msg = str(e) if "公式" in str(e) or "未知变量" in str(e) else "图纸数据解析失败"
+        return fail(msg)
     return ok(data=result, message="报价成功")

src/huamenchuang/api/router.py

@@ -11,6 +11,7 @@ from huamenchuang.api import (
     platform,
     quote,
     reports,
+    script,
     warehouse,
     windoor,
 )
@@ -23,6 +24,7 @@ api_router.include_router(contracts.router)
 api_router.include_router(windoor.router)
 api_router.include_router(optimize.router)
 api_router.include_router(quote.router)
+api_router.include_router(script.router)
 api_router.include_router(warehouse.router)
 api_router.include_router(business.router)
 api_router.include_router(platform.router)

src/huamenchuang/api/script.py

@@ -0,0 +1,111 @@
+"""价格脚本路由：按系列配置定价公式（对齐目标站 script/product-test 语义）。"""
+import json
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+
+from huamenchuang.core.deps import CurrentUser, DbSession
+from huamenchuang.core.response import fail, ok
+from huamenchuang.models.business import PriceScript
+from huamenchuang.services.formula import FormulaError, eval_formula, validate_formula
+
+router = APIRouter(prefix="/scripts", tags=["价格脚本"])
+
+
+class ScriptBody(BaseModel):
+    name: str = Field(min_length=1, max_length=64, description="系列/脚本名")
+    target: str = Field(default="frame",
+                        pattern="^(frame|mullion|sash|glass|hardware|total)$",
+                        description="公式作用项")
+    formula: str = Field(min_length=1, description="定价公式（安全表达式）")
+    note: str = ""
+    enabled: int = Field(default=1, ge=0, le=1)
+
+
+class ScriptPreviewBody(BaseModel):
+    formula: str = Field(min_length=1)
+    target: str = Field(default="frame",
+                        pattern="^(frame|mullion|sash|glass|hardware|total)$",
+                        description="作用项(length 变量按此取值)")
+    w: int = Field(default=1500, description="窗宽 mm")
+    h: int = Field(default=1800, description="窗高 mm")
+    glass_area: float = Field(default=2.3, description="玻璃面积 ㎡")
+    sashes: int = Field(default=2, description="扇数")
+    meters_frame: float = Field(default=6.6)
+    meters_mullion: float = Field(default=2.4)
+    meters_sash: float = Field(default=7.2)
+
+
+def _to_out(s: PriceScript) -> dict:
+    return {"id": s.id, "name": s.name, "target": s.target, "formula": s.formula,
+            "note": s.note, "enabled": s.enabled, "created_at": str(s.created_at)}
+
+
+@router.get("")
+async def list_scripts(user: CurrentUser, db: DbSession):
+    stmt = (select(PriceScript).where(PriceScript.company_id == user.company_id)
+            .order_by(PriceScript.id.desc()))
+    rows = list((await db.execute(stmt)).scalars())
+    return ok(data=[_to_out(r) for r in rows])
+
+
+@router.post("")
+async def create_script(body: ScriptBody, user: CurrentUser, db: DbSession):
+    err = validate_formula(body.formula)
+    if err:
+        return fail(f"公式错误: {err}")
+    s = PriceScript(company_id=user.company_id, **body.model_dump())
+    db.add(s)
+    await db.flush()
+    return ok(data=_to_out(s), message="脚本已创建")
+
+
+@router.put("/{script_id}")
+async def update_script(script_id: int, body: ScriptBody, user: CurrentUser, db: DbSession):
+    s = await db.get(PriceScript, script_id)
+    if s is None or s.company_id != user.company_id:
+        return fail("脚本不存在")
+    err = validate_formula(body.formula)
+    if err:
+        return fail(f"公式错误: {err}")
+    for k, v in body.model_dump().items():
+        setattr(s, k, v)
+    return ok(data=_to_out(s), message="脚本已更新")
+
+
+@router.delete("/{script_id}")
+async def delete_script(script_id: int, user: CurrentUser, db: DbSession):
+    s = await db.get(PriceScript, script_id)
+    if s is None or s.company_id != user.company_id:
+        return fail("脚本不存在")
+    await db.delete(s)
+    return ok(message="脚本已删除")
+
+
+@router.post("/preview")
+async def preview_script(body: ScriptPreviewBody, user: CurrentUser):
+    """脚本试算（对齐目标站 product-test）：以示例窗型变量求值公式。"""
+    length_of = {"frame": body.meters_frame, "mullion": body.meters_mullion,
+                 "sash": body.meters_sash, "glass": body.glass_area,
+                 "hardware": body.sashes, "total": 0}
+    vars_ = {"w": body.w, "h": body.h,
+             "area": round(body.w * body.h / 1_000_000, 3),
+             "perimeter": round((body.w + body.h) * 2 / 1000, 3),
+             "glass_area": body.glass_area, "sashes": body.sashes,
+             "meters_frame": body.meters_frame, "meters_mullion": body.meters_mullion,
+             "meters_sash": body.meters_sash, "length": length_of[body.target]}
+    try:
+        result = eval_formula(body.formula, vars_)
+    except FormulaError as e:
+        return fail(f"公式错误: {e}")
+    return ok(data={"result": round(result, 2), "vars": vars_})
+
+
+def load_script_map(scripts: list[PriceScript]) -> dict:
+    """启用脚本按 target 去重（同名 target 取最新一条），供报价使用。"""
+    mapping: dict = {}
+    for s in sorted(scripts, key=lambda x: x.id):
+        if s.enabled and s.formula:
+            mapping[s.target] = s.formula
+    return mapping

src/huamenchuang/models/__init__.py

@@ -3,6 +3,7 @@ from huamenchuang.models.announcement import Announcement
 from huamenchuang.models.business import (
     AfterSale,
     MeasureOrder,
+    PriceScript,
     Product,
     ProductCatelog,
     Quotation,
@@ -39,6 +40,7 @@ __all__ = [
     "OrderLog",
     "OrderProcess",
     "PayOrder",
+    "PriceScript",
     "Product",
     "ProductCatelog",
     "Quotation",

src/huamenchuang/models/business.py

@@ -7,6 +7,25 @@ from sqlalchemy.orm import Mapped, mapped_column
 from huamenchuang.core.database import Base
 
 
+class PriceScript(Base):
+    """价格脚本（对齐目标站 script/windoorscript：按系列配置定价公式，formula 为安全表达式）。"""
+
+    __tablename__ = "price_scripts"
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    company_id: Mapped[int] = mapped_column(default=1, index=True)
+    name: Mapped[str] = mapped_column(String(64))               # 系列/脚本名（如"断桥铝-55系列"）
+    # 公式作用项: frame外框/mullion中梃/sash扇框/glass玻璃/hardware五金/total总价直取
+    target: Mapped[str] = mapped_column(String(16), default="frame", index=True)
+    formula: Mapped[str] = mapped_column(Text)                  # 如 "(w+h)*2/1000*length*48"
+    note: Mapped[str] = mapped_column(Text, default="")
+    enabled: Mapped[int] = mapped_column(default=1)
+    created_at: Mapped[datetime.datetime] = mapped_column(DateTime, default=datetime.datetime.utcnow)
+    updated_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow
+    )
+
+
 class Product(Base):
     """产品价格本（型材/玻璃/五金/配件），供报价算料取价。"""
 

src/huamenchuang/scripts/seed.py

@@ -99,6 +99,24 @@ async def seed() -> None:
         await db.commit()
     print("平台公告就绪")
 
+    # 6) 价格脚本示例（幂等：按名称去重）
+    async with SessionLocal() as db:
+        from huamenchuang.models.business import PriceScript
+        SAMPLE_SCRIPTS = [
+            ("断桥铝-55系列", "frame", "(w+h)*2/1000*52"),
+            ("断桥铝-55系列", "mullion", "length*46"),
+            ("断桥铝-55系列", "sash", "length*58"),
+            ("双层中空玻璃", "glass", "max(110, area*135)"),
+            ("标准五金", "hardware", "sashes*195"),
+        ]
+        existing = {n for (n,) in (await db.execute(select(PriceScript.name))).all()}
+        for name, target, formula in SAMPLE_SCRIPTS:
+            if name not in existing:
+                db.add(PriceScript(name=name, target=target, formula=formula,
+                                   note="种子示例脚本，可在产品管理中修改"))
+        await db.commit()
+    print("价格脚本就绪")
+
 
 if __name__ == "__main__":
     asyncio.run(seed())

src/huamenchuang/services/formula.py

@@ -0,0 +1,189 @@
+"""安全价格公式引擎：递归下降解析器，支持变量与数学函数，绝不 eval/exec。
+
+文法:
+    expr   := term (('+'|'-') term)*
+    term   := unary (('*'|'/') unary)*
+    unary  := ('-'|'+') unary | primary
+    primary:= NUMBER | IDENT | IDENT '(' expr (',' expr)* ')' | '(' expr ')'
+函数: min max round abs ceil floor sqrt
+"""
+import math
+
+
+class FormulaError(ValueError):
+    """公式语法/求值错误（message 面向用户可读）。"""
+
+
+_FUNCTIONS: dict = {
+    "min": min,
+    "max": max,
+    "round": lambda v, ndigits=0: round(v, int(ndigits)),
+    "abs": abs,
+    "ceil": math.ceil,
+    "floor": math.floor,
+    "sqrt": math.sqrt,
+}
+
+
+def _tokenize(text: str) -> list[tuple[str, str]]:
+    """词法分析 → [(类型, 值)]。类型: num/ident/op/lp/rp/comma。"""
+    tokens: list[tuple[str, str]] = []
+    i, n = 0, len(text)
+    while i < n:
+        ch = text[i]
+        if ch.isspace():
+            i += 1
+        elif ch.isdigit() or (ch == "." and i + 1 < n and text[i + 1].isdigit()):
+            j = i
+            while j < n and (text[j].isdigit() or text[j] == "."):
+                j += 1
+            lit = text[i:j]
+            try:
+                float(lit)
+            except ValueError:
+                raise FormulaError(f"非法数字: {lit}")
+            tokens.append(("num", lit))
+            i = j
+        elif ch.isalpha() or ch == "_":
+            j = i
+            while j < n and (text[j].isalnum() or text[j] == "_"):
+                j += 1
+            tokens.append(("ident", text[i:j]))
+            i = j
+        elif ch in "+-*/":
+            tokens.append(("op", ch))
+            i += 1
+        elif ch == "(":
+            tokens.append(("lp", ch))
+            i += 1
+        elif ch == ")":
+            tokens.append(("rp", ch))
+            i += 1
+        elif ch == ",":
+            tokens.append(("comma", ch))
+            i += 1
+        else:
+            raise FormulaError(f"公式包含非法字符: {ch!r}")
+    return tokens
+
+
+class _Parser:
+    """递归下降求值器。"""
+
+    def __init__(self, tokens: list[tuple[str, str]], variables: dict) -> None:
+        self.tokens = tokens
+        self.pos = 0
+        self.vars = variables
+
+    def _peek(self) -> tuple[str, str] | None:
+        return self.tokens[self.pos] if self.pos < len(self.tokens) else None
+
+    def _next(self) -> tuple[str, str]:
+        tk = self._peek()
+        if tk is None:
+            raise FormulaError("公式不完整")
+        self.pos += 1
+        return tk
+
+    def parse(self) -> float:
+        v = self._expr()
+        tk = self._peek()
+        if tk is not None:
+            raise FormulaError(f"公式有多余内容: {tk[1]}")
+        return v
+
+    def _expr(self) -> float:
+        v = self._term()
+        while (tk := self._peek()) and tk[0] == "op" and tk[1] in "+-":
+            self._next()
+            rhs = self._term()
+            v = v + rhs if tk[1] == "+" else v - rhs
+        return v
+
+    def _term(self) -> float:
+        v = self._unary()
+        while (tk := self._peek()) and tk[0] == "op" and tk[1] in "*/":
+            self._next()
+            rhs = self._unary()
+            if tk[1] == "/" and rhs == 0:
+                raise FormulaError("除数为 0")
+            v = v * rhs if tk[1] == "*" else v / rhs
+        return v
+
+    def _unary(self) -> float:
+        tk = self._peek()
+        if tk and tk[0] == "op" and tk[1] in "+-":
+            self._next()
+            v = self._unary()
+            return -v if tk[1] == "-" else v
+        return self._primary()
+
+    def _primary(self) -> float:
+        kind, val = self._next()
+        if kind == "num":
+            return float(val)
+        if kind == "lp":
+            v = self._expr()
+            if (tk := self._next())[0] != "rp":
+                raise FormulaError("括号不匹配")
+            return v
+        if kind == "ident":
+            nxt = self._peek()
+            if nxt and nxt[0] == "lp":  # 函数调用
+                self._next()
+                args: list[float] = []
+                if (tk := self._peek()) and tk[0] != "rp":
+                    args.append(self._expr())
+                    while (tk := self._peek()) and tk[0] == "comma":
+                        self._next()
+                        args.append(self._expr())
+                if not (tk := self._next()) or tk[0] != "rp":
+                    raise FormulaError(f"函数 {val} 括号不匹配")
+                fn = _FUNCTIONS.get(val)
+                if fn is None:
+                    raise FormulaError(f"不支持的函数: {val}")
+                try:
+                    return float(fn(*args))
+                except TypeError:
+                    raise FormulaError(f"函数 {val} 参数数量不正确")
+            if val not in self.vars:
+                raise FormulaError(f"未知变量: {val}")
+            return float(self.vars[val])
+        raise FormulaError(f"意外的符号: {val}")
+
+
+def eval_formula(formula: str, variables: dict) -> float:
+    """求值价格公式。formula 为空/None 时抛 FormulaError。"""
+    if not formula or not formula.strip():
+        raise FormulaError("公式为空")
+    tokens = _tokenize(formula)
+    if not tokens:
+        raise FormulaError("公式为空")
+    return _Parser(tokens, variables).parse()
+
+
+# 公式可用变量白名单（与 quote_service 脚本变量表一致）
+ALLOWED_VARS = {
+    "w", "h", "area", "perimeter", "glass_area", "sashes",
+    "meters_frame", "meters_mullion", "meters_sash", "length",
+}
+
+
+def validate_formula(formula: str) -> str | None:
+    """校验公式：变量白名单 + 语法/求值检查。返回错误信息，None=合法。"""
+    if not formula or not formula.strip():
+        return "公式为空"
+    try:
+        tokens = _tokenize(formula)
+    except FormulaError as e:
+        return str(e)
+    idents = {v for k, v in tokens if k == "ident"}
+    unknown = idents - ALLOWED_VARS - set(_FUNCTIONS)
+    if unknown:
+        return f"未知变量: {sorted(unknown)[0]}（可用变量: {' '.join(sorted(ALLOWED_VARS))}）"
+    try:
+        # 全变量=1 试算，专查语法结构（括号/运算符/函数参数）
+        eval_formula(formula, {v: 1 for v in ALLOWED_VARS})
+    except FormulaError as e:
+        return str(e)
+    return None

src/huamenchuang/services/quote_service.py

@@ -26,6 +26,8 @@ class TakeoffResult:
     glass_area_m2: float = 0.0
     width_mm: int = 0
     height_mm: int = 0
+    sash_count: int = 0
+    meters: dict = field(default_factory=dict)  # 各型材类别总米数 {frame, mullion, sash}
 
 
 def _shoelace_area(pts: list[tuple[float, float]]) -> float:
@@ -105,6 +107,13 @@ def takeoff(dd) -> TakeoffResult:
     if xs:
         result.width_mm = round(max(xs) - min(xs))
         result.height_mm = round(max(ys) - min(ys))
+    # 型材米数汇总 + 扇数（供价格脚本变量表）
+    for kind in ("frame", "mullion", "sash"):
+        result.meters[kind] = round(sum(b.length * b.qty for b in result.bars if b.kind == kind) / 1000, 3)
+    result.sash_count = sum(
+        1 for s in dd.get("sm", []) for sub in (s.get("sm") or [])
+        if isinstance(sub, dict) and sub.get("bd") and not sub.get("bd").get("hidden")
+    )
     return result
 
 
@@ -120,37 +129,76 @@ def merge_bars(bars: list[BarItem]) -> list[dict]:
     return sorted(merged.values(), key=lambda x: -x["length"])
 
 
-def quote(drawing: str | dict, prices: dict | None = None) -> dict:
-    """完整报价：算料 + 套价。
+def quote(drawing: str | dict, prices: dict | None = None, scripts: dict | None = None) -> dict:
+    """完整报价：算料 + 套价（价格脚本可覆盖单项单价或总价）。
 
     prices 结构: {frame: 元/米, mullion: 元/米, sash: 元/米, glass: 元/㎡, hardware: 元/扇}
+    scripts 结构: {target: 公式字符串}，target ∈ frame/mullion/sash/glass/hardware/total。
+      公式结果 = 该项单价（元/米、元/㎡、元/扇）；target=total 时结果直接作为总价。
+      公式变量: w h(mm) area(㎡) perimeter(m) glass_area(㎡) sashes(扇)
+                meters_frame meters_mullion meters_sash(m) length(作用项对应量)
     """
+    from huamenchuang.services.formula import FormulaError, eval_formula
+
     p = {"frame": 45, "mullion": 40, "sash": 50, "glass": 120, "hardware": 180}
     if prices:
         p.update(prices)
 
     dd = parse_drawing(drawing)
     t = takeoff(dd)
-    rows, total = [], 0.0
-    unit_kind = {"frame": "frame", "mullion": "mullion", "sash": "sash"}
-    for b in merge_bars(t.bars):
-        amount = b["length"] * b["qty"] / 1000 * p[unit_kind[b["kind"]]]
-        total += amount
-        rows.append({**b, "unit_price": p[unit_kind[b["kind"]]], "unit": "元/米", "amount": round(amount, 2)})
 
-    glass_amount = t.glass_area_m2 * p["glass"]
+    # 价格脚本求值（公式错误直接上抛，由路由层转为用户可读失败）
+    total_override: float | None = None
+    script_unit: dict = {}
+    if scripts:
+        base_vars = {
+            "w": t.width_mm, "h": t.height_mm,
+            "area": round(t.width_mm * t.height_mm / 1_000_000, 3),
+            "perimeter": round((t.width_mm + t.height_mm) * 2 / 1000, 3),
+            "glass_area": round(t.glass_area_m2, 3),
+            "sashes": t.sash_count,
+            "meters_frame": t.meters.get("frame", 0),
+            "meters_mullion": t.meters.get("mullion", 0),
+            "meters_sash": t.meters.get("sash", 0),
+        }
+        length_of = {"frame": t.meters.get("frame", 0), "mullion": t.meters.get("mullion", 0),
+                     "sash": t.meters.get("sash", 0), "glass": round(t.glass_area_m2, 3),
+                     "hardware": t.sash_count, "total": 0}
+        for target, formula in scripts.items():
+            if target not in length_of or not formula or not str(formula).strip():
+                continue
+            vars_ = {**base_vars, "length": length_of[target]}
+            value = eval_formula(str(formula), vars_)
+            if target == "total":
+                total_override = value
+            else:
+                script_unit[target] = value
+
+    rows, total = [], 0.0
+    for b in merge_bars(t.bars):
+        unit = script_unit.get(b["kind"], p[b["kind"]])
+        amount = b["length"] * b["qty"] / 1000 * unit
+        total += amount
+        rows.append({**b, "unit_price": round(unit, 2), "unit": "元/米",
+                     "amount": round(amount, 2), "scripted": b["kind"] in script_unit})
+
+    glass_unit = script_unit.get("glass", p["glass"])
+    glass_amount = t.glass_area_m2 * glass_unit
     total += glass_amount
     rows.append({"kind": "glass", "length": 0, "note": "玻璃", "qty": round(t.glass_area_m2, 3),
-                 "unit_price": p["glass"], "unit": "元/㎡", "amount": round(glass_amount, 2)})
+                 "unit_price": round(glass_unit, 2), "unit": "元/㎡",
+                 "amount": round(glass_amount, 2), "scripted": "glass" in script_unit})
 
     sash_count = sum(
         1 for s in dd.get("sm", []) for sub in (s.get("sm") or [])
         if isinstance(sub, dict) and sub.get("bd") and not sub.get("bd").get("hidden")
     )
-    hw_amount = sash_count * p["hardware"]
+    hw_unit = script_unit.get("hardware", p["hardware"])
+    hw_amount = sash_count * hw_unit
     if sash_count:
         rows.append({"kind": "hardware", "length": 0, "note": "五金", "qty": sash_count,
-                     "unit_price": p["hardware"], "unit": "元/扇", "amount": round(hw_amount, 2)})
+                     "unit_price": round(hw_unit, 2), "unit": "元/扇",
+                     "amount": round(hw_amount, 2), "scripted": "hardware" in script_unit})
         total += hw_amount
 
     return {
@@ -158,5 +206,7 @@ def quote(drawing: str | dict, prices: dict | None = None) -> dict:
         "glass_area_m2": round(t.glass_area_m2, 3),
         "width_mm": t.width_mm,
         "height_mm": t.height_mm,
-        "total": round(total, 2),
+        "total": round(total_override if total_override is not None else total, 2),
+        "script_total": total_override is not None,
+        "scripted": sorted(script_unit.keys()),
     }