From 330dd9e2ceb79fa1b1e20814d27fa4efab79c9a4 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Thu, 30 Jul 2026 12:12:31 +0800 Subject: [PATCH] feat(backend): validate OA person field and whitelist type Co-authored-by: Cursor --- backend/app/services/oa_engine.py | 5 + backend/app/services/oa_person_select.py | 158 +++++++++++++++++++++++ backend/app/services/oa_publish.py | 2 + backend/tests/test_oa_person_select.py | 74 +++++++++++ 4 files changed, 239 insertions(+) create mode 100644 backend/app/services/oa_person_select.py create mode 100644 backend/tests/test_oa_person_select.py diff --git a/backend/app/services/oa_engine.py b/backend/app/services/oa_engine.py index ea82d388..cdbf016d 100644 --- a/backend/app/services/oa_engine.py +++ b/backend/app/services/oa_engine.py @@ -1552,6 +1552,7 @@ def start_instance( from app.services.oa_dept_select import validate_dept_form_values from app.services.oa_money_field import validate_money_form_values + from app.services.oa_person_select import validate_person_form_values dept_errs = validate_dept_form_values(schema, data) if dept_errs: @@ -1561,6 +1562,10 @@ def start_instance( if money_errs: raise OaEngineError(money_errs[0]) + person_errs = validate_person_form_values(schema, data) + if person_errs: + raise OaEngineError(person_errs[0]) + user = ( db.query(SysUser) .filter(SysUser.id == initiator_user_id, SysUser.del_flag == 0) diff --git a/backend/app/services/oa_person_select.py b/backend/app/services/oa_person_select.py new file mode 100644 index 00000000..d143c7c2 --- /dev/null +++ b/backend/app/services/oa_person_select.py @@ -0,0 +1,158 @@ +"""OA 表单设计器：人员选择控件 form_data 校验""" +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Set + + +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_multiple(rule: dict) -> bool: + props = _rule_props(rule) + multiple = props.get("multiple") + if multiple is None: + return True + return multiple is True + + +def _max_count(rule: dict) -> Optional[int]: + raw = _rule_props(rule).get("max") + if raw is None or raw == "": + return None + try: + value = int(raw) + except (TypeError, ValueError): + return None + if value <= 0: + return None + return value + + +def _is_valid_person_item(item: Any) -> bool: + if not isinstance(item, dict): + return False + person_id = item.get("id") + name = item.get("name") + if not isinstance(person_id, str) or not person_id.strip(): + return False + if not isinstance(name, str) or not name.strip(): + return False + return True + + +def _validate_person_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) + multiple = _is_multiple(rule) + max_count = _max_count(rule) + value = data.get(field) + errors: List[str] = [] + + if multiple: + if value is None: + if required: + errors.append(f"「{label}」为必填人员字段") + return errors + if not isinstance(value, list): + errors.append(f"「{label}」人员字段格式不正确，应为人员列表") + return errors + if required and len(value) == 0: + errors.append(f"「{label}」为必填人员字段") + return errors + for item in value: + if not _is_valid_person_item(item): + errors.append(f"「{label}」人员项格式不正确，需包含有效的 id 与 name") + break + if max_count is not None and len(value) > max_count: + errors.append(f"「{label}」选择人员数量超过最大上限 {max_count} 个") + return errors + + if value is None or value == "": + if required: + errors.append(f"「{label}」为必填人员字段") + return errors + if isinstance(value, list): + errors.append(f"「{label}」人员字段格式不正确，应为单个人员对象") + return errors + if not _is_valid_person_item(value): + errors.append(f"「{label}」人员字段格式不正确，需包含有效的 id 与 name") + return errors + + +def validate_person_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 "") != "person": + continue + errors.extend(_validate_person_rule(rule, data)) + return errors + + +def extract_person_ids(raw: Any) -> Set[str]: + ids: Set[str] = set() + items = raw if isinstance(raw, list) else ([raw] if isinstance(raw, dict) else []) + for item in items: + if not isinstance(item, dict): + continue + pid = str(item.get("id") or "").strip() + if pid: + ids.add(pid) + return ids + + +def person_contains_match(form_value: Any, cond_ids: Any) -> bool: + form_ids = extract_person_ids(form_value) + if isinstance(cond_ids, str): + want = {cond_ids.strip()} if cond_ids.strip() else set() + elif isinstance(cond_ids, list): + want = {str(x).strip() for x in cond_ids if str(x).strip()} + else: + want = set() + return bool(form_ids & want) diff --git a/backend/app/services/oa_publish.py b/backend/app/services/oa_publish.py index 9b63af6b..c6f62cf8 100644 --- a/backend/app/services/oa_publish.py +++ b/backend/app/services/oa_publish.py @@ -29,6 +29,7 @@ FILLABLE_TYPES = frozenset( "upload", "money", "dept", + "person", "region", "formula", "relatedApproval", @@ -80,6 +81,7 @@ CONDITION_ELIGIBLE_TYPES = frozenset( "select", "datePicker", "date", + "person", } ) diff --git a/backend/tests/test_oa_person_select.py b/backend/tests/test_oa_person_select.py new file mode 100644 index 00000000..8492c4ae --- /dev/null +++ b/backend/tests/test_oa_person_select.py @@ -0,0 +1,74 @@ +import unittest + +from app.services.oa_person_select import ( + extract_person_ids, + person_contains_match, + validate_person_form_values, +) + + +class TestOaPersonSelect(unittest.TestCase): + def test_validate_multi_ok(self): + schema = { + "rule": [ + { + "type": "person", + "field": "F1", + "title": "协同人", + "props": {"multiple": True}, + } + ] + } + data = {"F1": [{"id": "u1", "name": "张三"}, {"id": "u2", "name": "李四"}]} + self.assertEqual(validate_person_form_values(schema, data), []) + + def test_validate_required_empty(self): + schema = { + "rule": [ + { + "type": "person", + "field": "F1", + "title": "协同人", + "$required": True, + "props": {"multiple": True}, + } + ] + } + errs = validate_person_form_values(schema, {"F1": []}) + self.assertTrue(any("必填" in e for e in errs)) + + def test_validate_max(self): + schema = { + "rule": [ + { + "type": "person", + "field": "F1", + "title": "协同人", + "props": {"multiple": True, "max": 1}, + } + ] + } + errs = validate_person_form_values( + schema, + {"F1": [{"id": "u1", "name": "A"}, {"id": "u2", "name": "B"}]}, + ) + self.assertTrue(any("上限" in e for e in errs)) + + def test_extract_and_contains(self): + self.assertEqual(extract_person_ids({"id": "u1", "name": "A"}), {"u1"}) + self.assertTrue( + person_contains_match( + [{"id": "u1", "name": "A"}], + ["u1", "u2"], + ) + ) + self.assertFalse( + person_contains_match( + [{"id": "u3", "name": "C"}], + ["u1", "u2"], + ) + ) + + +if __name__ == "__main__": + unittest.main()