diff --git a/backend/app/services/oa_engine.py b/backend/app/services/oa_engine.py index 54a9c253..ea82d388 100644 --- a/backend/app/services/oa_engine.py +++ b/backend/app/services/oa_engine.py @@ -1551,11 +1551,16 @@ def start_instance( raise OaEngineError(f"公式计算失败: {e}") from e from app.services.oa_dept_select import validate_dept_form_values + from app.services.oa_money_field import validate_money_form_values dept_errs = validate_dept_form_values(schema, data) if dept_errs: raise OaEngineError(dept_errs[0]) + money_errs = validate_money_form_values(schema, data) + if money_errs: + raise OaEngineError(money_errs[0]) + user = ( db.query(SysUser) .filter(SysUser.id == initiator_user_id, SysUser.del_flag == 0) diff --git a/backend/app/services/oa_money_field.py b/backend/app/services/oa_money_field.py new file mode 100644 index 00000000..1c42bead --- /dev/null +++ b/backend/app/services/oa_money_field.py @@ -0,0 +1,119 @@ +"""OA 表单设计器：金额控件 form_data 校验""" +from __future__ import annotations + +import math +from typing import Any, Dict, List, Optional, Tuple + +MAX_MONEY = 999999999999.99 + + +def _walk_rules(rules: Any, out: List[dict]) -> None: + if not isinstance(rules, list): + return + for r in rules: + if not isinstance(r, dict): + continue + out.append(r) + children = r.get("children") or (r.get("props") or {}).get("rule") or [] + if isinstance(children, list): + _walk_rules(children, out) + + +def _schema_rules(schema: Any) -> List[dict]: + if isinstance(schema, list): + out: List[dict] = [] + _walk_rules(schema, out) + return out + if not isinstance(schema, dict): + return [] + out: List[dict] = [] + _walk_rules(schema.get("rule") or schema.get("rules") or [], out) + return out + + +def _rule_props(rule: dict) -> dict: + props = rule.get("props") + return props if isinstance(props, dict) else {} + + +def _field_label(rule: dict) -> str: + return str(rule.get("title") or rule.get("label") or rule.get("field") or "金额") + + +def _is_required(rule: dict) -> bool: + props = _rule_props(rule) + if rule.get("$required") is True: + return True + if rule.get("required") is True: + return True + if props.get("required") is True: + return True + return False + + +def _is_empty(value: Any) -> bool: + return value is None or value == "" + + +def _parse_money(value: Any) -> Tuple[Optional[float], Optional[str]]: + """解析金额值。成功返回 (number, None)；失败返回 (None, error_key)。""" + if _is_empty(value): + return None, None + if isinstance(value, bool): + return None, "invalid" + if isinstance(value, (int, float)): + n = float(value) + elif isinstance(value, str): + text = value.replace(",", "").strip() + if not text: + return None, None + try: + n = float(text) + except ValueError: + return None, "invalid" + else: + return None, "invalid" + + if not math.isfinite(n): + return None, "invalid" + if n < 0: + return None, "negative" + if n > MAX_MONEY: + return None, "too_large" + return n, None + + +def _validate_money_rule(rule: dict, data: Dict[str, Any]) -> List[str]: + field = str(rule.get("field") or "").strip() + if not field: + return [] + + label = _field_label(rule) + required = _is_required(rule) + value = data.get(field) + errors: List[str] = [] + + if _is_empty(value): + if required: + errors.append(f"「{label}」为必填金额字段") + return errors + + _, err_key = _parse_money(value) + if err_key == "negative": + errors.append(f"「{label}」金额不能为负数") + elif err_key == "too_large": + errors.append("金额过大") + elif err_key == "invalid": + errors.append(f"「{label}」金额格式不正确") + return errors + + +def validate_money_form_values(schema: Any, form_data: Optional[Dict[str, Any]]) -> List[str]: + """校验 form_data 中金额控件的值格式，返回中文错误列表（空表示通过）。""" + data = form_data or {} + errors: List[str] = [] + for rule in _schema_rules(schema): + if str(rule.get("type") or "") != "money": + continue + errors.extend(_validate_money_rule(rule, data)) + return errors diff --git a/backend/tests/test_oa_money_field.py b/backend/tests/test_oa_money_field.py new file mode 100644 index 00000000..032ec22c --- /dev/null +++ b/backend/tests/test_oa_money_field.py @@ -0,0 +1,47 @@ +import unittest + +from app.services.oa_money_field import validate_money_form_values + + +class TestValidateMoneyFormValues(unittest.TestCase): + def test_ok_number(self): + schema = {"rule": [{"type": "money", "field": "amt", "title": "金额", "$required": True}]} + self.assertEqual(validate_money_form_values(schema, {"amt": 1234.56}), []) + + def test_ok_int(self): + schema = {"rule": [{"type": "money", "field": "amt"}]} + self.assertEqual(validate_money_form_values(schema, {"amt": 10}), []) + + def test_ok_null_optional(self): + schema = {"rule": [{"type": "money", "field": "amt", "$required": False}]} + self.assertEqual(validate_money_form_values(schema, {"amt": None}), []) + self.assertEqual(validate_money_form_values(schema, {}), []) + + def test_required_missing(self): + schema = { + "rule": [{"type": "money", "field": "amt", "title": "报销金额", "$required": True}] + } + errs = validate_money_form_values(schema, {}) + self.assertTrue(any("报销金额" in e for e in errs)) + + def test_negative_rejected(self): + schema = {"rule": [{"type": "money", "field": "amt"}]} + errs = validate_money_form_values(schema, {"amt": -1}) + self.assertTrue(len(errs) >= 1) + + def test_string_number_accepted(self): + schema = {"rule": [{"type": "money", "field": "amt"}]} + self.assertEqual(validate_money_form_values(schema, {"amt": "12.5"}), []) + + def test_bool_rejected(self): + schema = {"rule": [{"type": "money", "field": "amt"}]} + errs = validate_money_form_values(schema, {"amt": True}) + self.assertTrue(len(errs) >= 1) + + def test_ignore_other_types(self): + schema = {"rule": [{"type": "input", "field": "x", "$required": True}]} + self.assertEqual(validate_money_form_values(schema, {}), []) + + +if __name__ == "__main__": + unittest.main()