469 lines
18 KiB
Python
469 lines
18 KiB
Python
"""配置驱动的报价计算引擎。
|
||
|
||
所有产品差异由 product_pricing_rule 表的 JSON 配置决定，
|
||
引擎本身不包含任何产品特定的 if/else 分支。
|
||
"""
|
||
import json
|
||
import re
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
|
||
|
||
class PricingEngine:
|
||
"""读取 PricingRule 配置，执行公式求值、附加费计算、详情构建。
|
||
|
||
所有产品差异由 product_pricing_rule 表的 JSON 配置驱动，
|
||
引擎本身不包含任何产品特定的 if/else 分支。
|
||
"""
|
||
|
||
# ------------------------------------------------------------------
|
||
# 公开方法
|
||
# ------------------------------------------------------------------
|
||
|
||
def calculate(self, rule, user_inputs: dict, product_attrs: dict | None = None) -> dict:
|
||
"""执行完整报价计算。
|
||
|
||
流程：单位换算 -> 构建公式上下文 -> 公式求值 -> 附加费计算 -> 构建公式说明。
|
||
|
||
Args:
|
||
rule: ProductPricingRule ORM 对象（含 formula_expr 等字段）
|
||
user_inputs: 前端传入 {"length": 2, "width": 1.2, "length_unit": "m", ...}
|
||
product_attrs: 产品属性 {"thickness": 0.08} 用于阈值判断
|
||
|
||
Returns:
|
||
结构化计价结果字典，包含 base_cost、surcharge_items、total_surcharge、
|
||
cost_price、formula_detail、tax 字段，以及 rule_snapshot 用于快照存储。
|
||
|
||
被调用路由: pricing.py - POST /pricing/calculate
|
||
"""
|
||
product_attrs = product_attrs or {}
|
||
|
||
# 1. 单位换算
|
||
normalized = self._normalize_inputs(rule, user_inputs)
|
||
area_sqm = normalized.get("length_m", 0) * normalized.get("width_m", 0)
|
||
|
||
# 2. 构建公式上下文并求值
|
||
formula_constants = self._safe_load_json(rule.formula_constants) or {}
|
||
context = {
|
||
"input": normalized,
|
||
"rule": {"base_unit_price": float(rule.base_unit_price)},
|
||
"const": formula_constants,
|
||
"product": product_attrs,
|
||
}
|
||
base_cost = self._evaluate_formula(rule.formula_expr or "0", context)
|
||
|
||
# 3. 计算附加费
|
||
surcharges = self._calculate_surcharges(
|
||
rule.surcharge_json, user_inputs, normalized, area_sqm, product_attrs,
|
||
)
|
||
total_surcharge = round(sum(s["amount"] for s in surcharges), 2)
|
||
cost_price = round(base_cost + total_surcharge, 2)
|
||
|
||
# 4. 构建公式说明
|
||
formula_detail = self._build_formula_detail(rule, normalized, base_cost, surcharges)
|
||
|
||
# 5. 税计算
|
||
tax_rate = float(getattr(rule, "tax_rate", 0) or 0)
|
||
tax_inclusive = int(getattr(rule, "tax_inclusive", 0) or 0)
|
||
if tax_rate > 0:
|
||
if tax_inclusive:
|
||
price_ex_tax = round(cost_price / (1 + tax_rate / 100), 2)
|
||
tax_amount = round(cost_price - price_ex_tax, 2)
|
||
price_in_tax = cost_price
|
||
else:
|
||
price_ex_tax = cost_price
|
||
tax_amount = round(cost_price * tax_rate / 100, 2)
|
||
price_in_tax = round(cost_price + tax_amount, 2)
|
||
else:
|
||
price_ex_tax = cost_price
|
||
tax_amount = 0
|
||
price_in_tax = cost_price
|
||
|
||
# 6. 构建变量快照（用于审计和回显）
|
||
formula_variables = {}
|
||
for prefix, values in context.items():
|
||
if isinstance(values, dict):
|
||
for k, v in values.items():
|
||
formula_variables[f"{prefix}.{k}"] = v
|
||
|
||
# 7. 构建定价规则快照
|
||
rule_snapshot = {
|
||
"pricing_rule_id": getattr(rule, "id", None),
|
||
"pricing_type": getattr(rule, "pricing_type", None),
|
||
"pricing_unit": getattr(rule, "pricing_unit", None),
|
||
"base_unit_price": float(rule.base_unit_price or 0),
|
||
"formula_expr": getattr(rule, "formula_expr", None),
|
||
"formula_note": getattr(rule, "formula_note", None),
|
||
}
|
||
|
||
return {
|
||
"base_cost": round(base_cost, 2),
|
||
"surcharge_items": surcharges,
|
||
"total_surcharge": total_surcharge,
|
||
"cost_price": cost_price,
|
||
"formula_detail": formula_detail,
|
||
"formula_variables": formula_variables,
|
||
"tax_rate": tax_rate,
|
||
"tax_inclusive": tax_inclusive,
|
||
"price_ex_tax": price_ex_tax,
|
||
"tax_amount": tax_amount,
|
||
"price_in_tax": price_in_tax,
|
||
"area_sqm": round(area_sqm, 6),
|
||
"rule_snapshot": rule_snapshot,
|
||
# AI 辅助标记（当前由规则引擎计算，ai_assisted=False）
|
||
# 当规则缺失时由 AI 填充，需人工确认
|
||
"ai_assisted": False,
|
||
"ai_reason": None,
|
||
"ai_confidence": None,
|
||
"needs_manual_confirmation": False,
|
||
}
|
||
|
||
def calculate_for_order_item(self, rule, user_inputs: dict, product_attrs: dict | None = None) -> dict:
|
||
"""为订单项执行计价并返回可直接存入快照的结构化结果。
|
||
|
||
在 calculate() 基础上，额外返回适合存入 SalesOrderItem 的字段映射。
|
||
|
||
Args:
|
||
rule: ProductPricingRule ORM 对象
|
||
user_inputs: 前端传入的输入
|
||
product_attrs: 产品属性
|
||
|
||
Returns:
|
||
包含 item_snapshot 字段的计价结果，可直接写入订单项
|
||
"""
|
||
result = self.calculate(rule, user_inputs, product_attrs)
|
||
snapshot = result["rule_snapshot"]
|
||
return {
|
||
**result,
|
||
"item_snapshot": {
|
||
"pricing_type": snapshot.get("pricing_type"),
|
||
"pricing_unit": snapshot.get("pricing_unit"),
|
||
"area_sqm": result["area_sqm"],
|
||
"surcharge_detail": json.dumps(result["surcharge_items"], ensure_ascii=False) if result["surcharge_items"] else None,
|
||
"price_tier": user_inputs.get("price_tier"),
|
||
"cost_price": result["cost_price"],
|
||
},
|
||
}
|
||
|
||
def get_available_surcharge_options(self, rule) -> list[dict]:
|
||
"""返回定价规则中可选的附加费配置列表。
|
||
|
||
Args:
|
||
rule: ProductPricingRule ORM 对象
|
||
|
||
Returns:
|
||
附加费选项列表，每项包含 key、name、method、price 等
|
||
|
||
被调用路由: pricing.py - GET /pricing/surcharge-options/{rule_id}
|
||
"""
|
||
surcharge = self._safe_load_json(rule.surcharge_json) or {}
|
||
return [
|
||
{"key": key, "name": cfg.get("name", key), "method": cfg.get("method", ""),
|
||
"price": cfg.get("price", 0), "needs_input": cfg.get("needs_input", False),
|
||
"input_key": cfg.get("input_key", key)}
|
||
for key, cfg in surcharge.items()
|
||
]
|
||
|
||
# ------------------------------------------------------------------
|
||
# 单位换算
|
||
# ------------------------------------------------------------------
|
||
|
||
_UNIT_TO_M = {"m": 1.0, "cm": 0.01, "mm": 0.001}
|
||
|
||
# 重量单位 → 千克
|
||
_UNIT_TO_KG = {"kg": 1.0, "g": 0.001, "t": 1000.0}
|
||
|
||
# 面积单位 → 平方米
|
||
_UNIT_TO_SQM = {"sqm": 1.0, "sqcm": 0.0001, "sqmm": 0.000001}
|
||
|
||
# 体积单位 → 立方米
|
||
_UNIT_TO_CBM = {"cbm": 1.0, "cubic_m": 1.0, "cubic_cm": 0.000001, "liter": 0.001}
|
||
|
||
def _normalize_inputs(self, rule, user_inputs: dict) -> dict:
|
||
"""将前端输入转换为统一的标准单位数值。
|
||
|
||
根据 pricing_inputs 声明解析字段，自动读取 _unit 后缀进行单位换算。
|
||
- 长度类字段统一转为米（_m 后缀）
|
||
- 重量类字段统一转为千克（_kg 后缀）
|
||
- 面积类字段统一转为平方米（_sqm 后缀）
|
||
|
||
Args:
|
||
rule: 定价规则 ORM 对象
|
||
user_inputs: 前端传入的原始输入字典
|
||
|
||
Returns:
|
||
标准化后的输入字典，包含原始值和标准化后缀值
|
||
"""
|
||
normalized = {}
|
||
inputs_decl = self._safe_load_json(rule.pricing_inputs) or []
|
||
|
||
for field in inputs_decl:
|
||
key = field.get("key", "")
|
||
val = user_inputs.get(key)
|
||
if val is None or val == "":
|
||
normalized[key] = 0
|
||
continue
|
||
val = float(val)
|
||
|
||
if field.get("type") == "boolean":
|
||
normalized[key] = val
|
||
continue
|
||
|
||
unit_key = f"{key}_unit"
|
||
unit = user_inputs.get(unit_key, field.get("default_unit", "m"))
|
||
field_type = field.get("unit_type", "length")
|
||
|
||
if field_type == "weight":
|
||
multiplier = self._UNIT_TO_KG.get(unit, 1.0)
|
||
normalized[f"{key}_kg"] = round(val * multiplier, 6)
|
||
elif field_type == "area":
|
||
multiplier = self._UNIT_TO_SQM.get(unit, 1.0)
|
||
normalized[f"{key}_sqm"] = round(val * multiplier, 6)
|
||
elif field_type == "volume":
|
||
multiplier = self._UNIT_TO_CBM.get(unit, 1.0)
|
||
normalized[f"{key}_cbm"] = round(val * multiplier, 6)
|
||
else:
|
||
multiplier = self._UNIT_TO_M.get(unit, 1.0)
|
||
normalized[f"{key}_m"] = round(val * multiplier, 6)
|
||
|
||
normalized[key] = val
|
||
|
||
return normalized
|
||
|
||
# ------------------------------------------------------------------
|
||
# 公式求值
|
||
# ------------------------------------------------------------------
|
||
|
||
def _evaluate_formula(self, expr: str, context: dict) -> float:
|
||
"""替换变量 -> 解析条件表达式 -> 安全求值。
|
||
|
||
支持 $input.xxx、$rule.xxx、$const.xxx、$product.xxx 变量引用，
|
||
以及 ${condition} ? a : b 三元条件表达式。
|
||
|
||
使用正则替换（按变量名长度降序）避免短变量名误匹配长变量名前缀。
|
||
|
||
Args:
|
||
expr: 公式表达式字符串
|
||
context: 变量上下文字典
|
||
|
||
Returns:
|
||
计算结果浮点数，异常时返回 0.0
|
||
"""
|
||
resolved = expr
|
||
|
||
# 收集所有 (完整变量名, 替换值)，按变量名长度降序排列防止前缀误匹配
|
||
replacements = []
|
||
for prefix, values in context.items():
|
||
if not isinstance(values, dict):
|
||
continue
|
||
for k, v in values.items():
|
||
var_name = f"${prefix}.{k}"
|
||
replacements.append((var_name, self._to_str(v)))
|
||
replacements.sort(key=lambda x: len(x[0]), reverse=True)
|
||
|
||
for var_name, replacement in replacements:
|
||
# 用 re.escape 转义变量名中的特殊字符（如 $ 和 .）
|
||
pattern = re.escape(var_name)
|
||
resolved = re.sub(pattern, replacement, resolved)
|
||
|
||
# 解析 ${condition} ? a : b
|
||
resolved = self._resolve_conditionals(resolved)
|
||
|
||
# 安全求值
|
||
try:
|
||
return round(float(eval(resolved, {"__builtins__": {}}, {"round": round})), 4)
|
||
except Exception:
|
||
return 0.0
|
||
|
||
def _resolve_conditionals(self, expr: str) -> str:
|
||
"""处理 ${condition} ? valueA : valueB 三元表达式。
|
||
|
||
循环替换直到表达式中不再包含条件表达式（支持嵌套）。
|
||
|
||
Args:
|
||
expr: 含条件表达式的公式字符串
|
||
|
||
Returns:
|
||
条件表达式已解析的公式字符串
|
||
"""
|
||
pattern = r"\$\{([^}]+)\}\s*\?\s*([^:]+?)\s*:\s*([^,\s\)]+)"
|
||
while re.search(pattern, expr):
|
||
expr = re.sub(pattern, self._eval_one_conditional, expr)
|
||
return expr
|
||
|
||
def _eval_one_conditional(self, match: re.Match) -> str:
|
||
"""求值单个三元条件表达式。
|
||
|
||
Args:
|
||
match: 正则匹配对象，包含条件、真值、假值三个分组
|
||
|
||
Returns:
|
||
条件求值结果对应的值字符串
|
||
"""
|
||
cond = match.group(1).strip()
|
||
val_true = match.group(2).strip()
|
||
val_false = match.group(3).strip()
|
||
try:
|
||
result = eval(cond, {"__builtins__": {}}, {}) # noqa: S307
|
||
return val_true if result else val_false
|
||
except Exception:
|
||
return val_false
|
||
|
||
# ------------------------------------------------------------------
|
||
# 附加费计算
|
||
# ------------------------------------------------------------------
|
||
|
||
def _calculate_surcharges(
|
||
self, surcharge_json, user_inputs: dict, normalized: dict,
|
||
area_sqm: float, product_attrs: dict,
|
||
) -> list[dict]:
|
||
"""计算所有启用的附加费项。
|
||
|
||
支持四种计费方式：按面积(per_sqm)、按数量(per_piece)、
|
||
按长度(per_linear_m)、按面积阈值(per_sqm_threshold)。
|
||
|
||
Args:
|
||
surcharge_json: 附加费配置 JSON 字符串
|
||
user_inputs: 前端原始输入
|
||
normalized: 标准化后的输入
|
||
area_sqm: 面积（平方米）
|
||
product_attrs: 产品属性字典
|
||
|
||
Returns:
|
||
附加费计算结果列表，每项包含 key、name、method、amount
|
||
"""
|
||
surcharge = self._safe_load_json(surcharge_json) or {}
|
||
results = []
|
||
for key, cfg in surcharge.items():
|
||
if not user_inputs.get(f"surcharge_{key}", False):
|
||
continue
|
||
|
||
method = cfg.get("method", "per_sqm")
|
||
amount = 0.0
|
||
|
||
if method == "per_sqm":
|
||
amount = round(area_sqm * cfg.get("price", 0), 2)
|
||
elif method == "per_piece":
|
||
input_key = cfg.get("input_key", key)
|
||
qty = float(user_inputs.get(input_key, 0) or 0)
|
||
amount = round(qty * cfg.get("price", 0), 2)
|
||
elif method == "per_linear_m":
|
||
input_key = cfg.get("input_key", key)
|
||
length = float(user_inputs.get(input_key, 0) or 0)
|
||
amount = round(length * cfg.get("price", 0), 2)
|
||
elif method == "per_sqm_threshold":
|
||
threshold = cfg.get("threshold", {})
|
||
field_val = float(product_attrs.get(threshold.get("field", ""), 0) or 0)
|
||
price = threshold["price_true"] if field_val <= threshold.get("lte", 0) else threshold["price_false"]
|
||
amount = round(area_sqm * price, 2)
|
||
|
||
results.append({
|
||
"key": key, "name": cfg.get("name", key),
|
||
"method": method, "amount": amount,
|
||
})
|
||
return results
|
||
|
||
# ------------------------------------------------------------------
|
||
# 公式说明构建
|
||
# ------------------------------------------------------------------
|
||
|
||
def _build_formula_detail(self, rule, normalized: dict, base_cost: float, surcharges: list) -> str:
|
||
"""构建人类可读的计算过程文本。
|
||
|
||
根据定价类型自动生成对应的计算说明：
|
||
- area: 2m x 1.2m x ¥50/㎡ = ¥120.00
|
||
- kg: 5kg x ¥8/kg = ¥40.00
|
||
- unit/件: 10件 x ¥25/件 = ¥250.00
|
||
- linear_m: 20m x ¥15/m = ¥300.00
|
||
|
||
Args:
|
||
rule: 定价规则 ORM 对象
|
||
normalized: 标准化后的输入
|
||
base_cost: 基础费用
|
||
surcharges: 附加费列表
|
||
|
||
Returns:
|
||
公式说明字符串
|
||
"""
|
||
pricing_type = getattr(rule, "pricing_type", "area") or "area"
|
||
unit_price = float(rule.base_unit_price or 0)
|
||
parts = []
|
||
|
||
if pricing_type == "area":
|
||
length = normalized.get("length_m", 0)
|
||
width = normalized.get("width_m", 0)
|
||
parts.append(f"{length}m × {width}m × ¥{unit_price}/㎡ = ¥{base_cost:.2f}")
|
||
elif pricing_type in ("kg", "weight_g"):
|
||
weight = normalized.get("weight_kg", normalized.get("weight_kg", 0))
|
||
if pricing_type == "weight_g":
|
||
weight_g = weight * 1000 if weight else normalized.get("weight", 0)
|
||
parts.append(f"{weight_g}g × ¥{unit_price}/g = ¥{base_cost:.2f}")
|
||
else:
|
||
parts.append(f"{weight}kg × ¥{unit_price}/kg = ¥{base_cost:.2f}")
|
||
elif pricing_type == "linear_m":
|
||
length = normalized.get("length_m", normalized.get("length", 0))
|
||
parts.append(f"{length}m × ¥{unit_price}/m = ¥{base_cost:.2f}")
|
||
elif pricing_type in ("unit", "件"):
|
||
qty = normalized.get("quantity", normalized.get("qty", 0))
|
||
parts.append(f"{qty}件 × ¥{unit_price}/件 = ¥{base_cost:.2f}")
|
||
elif pricing_type == "volume":
|
||
vol = normalized.get("volume_cbm", normalized.get("volume", 0))
|
||
parts.append(f"{vol}m³ × ¥{unit_price}/m³ = ¥{base_cost:.2f}")
|
||
else:
|
||
# 通用：显示公式结果
|
||
parts.append(f"计算结果 = ¥{base_cost:.2f}")
|
||
|
||
for s in surcharges:
|
||
parts.append(f"+ {s['name']} ¥{s['amount']:.2f}")
|
||
|
||
total = base_cost + sum(s["amount"] for s in surcharges)
|
||
parts.append(f"= ¥{total:.2f}")
|
||
|
||
tax_rate = float(getattr(rule, "tax_rate", 0) or 0)
|
||
if tax_rate > 0:
|
||
tax_inclusive = int(getattr(rule, "tax_inclusive", 0) or 0)
|
||
if tax_inclusive:
|
||
parts.append(f"含税价(税率{tax_rate}%)")
|
||
else:
|
||
parts.append(f"+ 税额(税率{tax_rate}%)")
|
||
|
||
return " → ".join(parts)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 工具方法
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _safe_load_json(text: str | None) -> dict | list | None:
|
||
"""安全解析 JSON 字符串。
|
||
|
||
Args:
|
||
text: JSON 字符串
|
||
|
||
Returns:
|
||
解析后的 dict/list，空值或解析失败时返回 None
|
||
"""
|
||
if not text:
|
||
return None
|
||
try:
|
||
return json.loads(text)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _to_str(v) -> str:
|
||
"""将值转换为字符串，布尔值特殊处理为 True/False。
|
||
|
||
Args:
|
||
v: 任意值
|
||
|
||
Returns:
|
||
字符串表示
|
||
"""
|
||
if isinstance(v, bool):
|
||
return "True" if v else "False"
|
||
return str(v)
|
||
|
||
|
||
# 模块级单例
|
||
pricing_engine = PricingEngine()
|