feat(backend): add receive account CRUD helpers and validators.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/crud/oa_receive_account.py

@@ -0,0 +1,203 @@
+"""OA 收款账户 CRUD（oa_receive_account）"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.models import OaReceiveAccount
+
+
+def row_to_dict(r: OaReceiveAccount) -> Dict[str, Any]:
+    return {
+        "id": r.id,
+        "accountType": r.account_type,
+        "accountName": r.account_name,
+        "accountNo": r.account_no,
+        "bankName": r.bank_name or "",
+        "bankProvince": r.bank_province or "",
+        "bankCity": r.bank_city or "",
+        "bankBranch": r.bank_branch or "",
+        "tenantNo": r.tenant_no,
+        "orderNum": r.order_num if r.order_num is not None else 0,
+        "remark": r.remark or "",
+        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
+        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
+    }
+
+
+def _filtered_query(
+    db: Session,
+    *,
+    tenant_no: str,
+    keyword: Optional[str] = None,
+    account_type: Optional[str] = None,
+):
+    q = db.query(OaReceiveAccount).filter(
+        OaReceiveAccount.tenant_no == tenant_no,
+        OaReceiveAccount.del_flag == 0,
+    )
+    kw = (keyword or "").strip()
+    if kw:
+        like = f"%{kw}%"
+        q = q.filter(
+            or_(
+                OaReceiveAccount.account_name.like(like),
+                OaReceiveAccount.account_no.like(like),
+                OaReceiveAccount.bank_name.like(like),
+            )
+        )
+    if account_type:
+        q = q.filter(OaReceiveAccount.account_type == account_type.strip())
+    return q
+
+
+def list_accounts(
+    db: Session,
+    *,
+    tenant_no: str,
+    keyword: Optional[str] = None,
+    account_type: Optional[str] = None,
+    page: int = 1,
+    page_size: int = 10,
+) -> Tuple[List[Dict[str, Any]], int]:
+    base = _filtered_query(db, tenant_no=tenant_no, keyword=keyword, account_type=account_type)
+    total = base.order_by(None).with_entities(func.count(OaReceiveAccount.id)).scalar() or 0
+    page = max(1, page)
+    page_size = min(max(1, page_size), 100)
+    rows = (
+        base.order_by(OaReceiveAccount.create_time.desc())
+        .offset((page - 1) * page_size)
+        .limit(page_size)
+        .all()
+    )
+    return [row_to_dict(x) for x in rows], int(total)
+
+
+def list_options(db: Session, *, tenant_no: str) -> List[Dict[str, Any]]:
+    rows = (
+        db.query(OaReceiveAccount)
+        .filter(OaReceiveAccount.tenant_no == tenant_no, OaReceiveAccount.del_flag == 0)
+        .order_by(OaReceiveAccount.create_time.desc())
+        .all()
+    )
+    return [row_to_dict(x) for x in rows]
+
+
+def get_account(db: Session, account_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
+    row = (
+        db.query(OaReceiveAccount)
+        .filter(
+            OaReceiveAccount.id == account_id,
+            OaReceiveAccount.tenant_no == tenant_no,
+            OaReceiveAccount.del_flag == 0,
+        )
+        .first()
+    )
+    return row_to_dict(row) if row else None
+
+
+def create_account(
+    db: Session,
+    *,
+    tenant_no: str,
+    user_id: Optional[str],
+    dept_id: str,
+    fields: Dict[str, Any],
+) -> Dict[str, Any]:
+    now = datetime.now()
+    obj = OaReceiveAccount(
+        id=generate_uuid(),
+        account_type=fields["account_type"],
+        account_name=fields["account_name"],
+        account_no=fields["account_no"],
+        bank_name=fields.get("bank_name") or None,
+        bank_province=fields.get("bank_province") or None,
+        bank_city=fields.get("bank_city") or None,
+        bank_branch=fields.get("bank_branch") or None,
+        tenant_no=tenant_no,
+        order_num=fields.get("order_num") if fields.get("order_num") is not None else 0,
+        remark=(fields.get("remark") or "").strip() or None,
+        create_by=user_id,
+        create_dept_id=dept_id or "",
+        create_time=now,
+        update_by=user_id,
+        update_time=now,
+        del_flag=0,
+    )
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return row_to_dict(obj)
+
+
+def update_account(
+    db: Session,
+    account_id: str,
+    tenant_no: str,
+    user_id: Optional[str],
+    fields: Dict[str, Any],
+) -> Optional[Dict[str, Any]]:
+    obj = (
+        db.query(OaReceiveAccount)
+        .filter(
+            OaReceiveAccount.id == account_id,
+            OaReceiveAccount.tenant_no == tenant_no,
+            OaReceiveAccount.del_flag == 0,
+        )
+        .first()
+    )
+    if not obj:
+        return None
+    obj.account_type = fields["account_type"]
+    obj.account_name = fields["account_name"]
+    obj.account_no = fields["account_no"]
+    obj.bank_name = fields.get("bank_name") or None
+    obj.bank_province = fields.get("bank_province") or None
+    obj.bank_city = fields.get("bank_city") or None
+    obj.bank_branch = fields.get("bank_branch") or None
+    if fields.get("order_num") is not None:
+        obj.order_num = fields["order_num"]
+    if fields.get("remark") is not None:
+        obj.remark = (fields.get("remark") or "").strip() or None
+    obj.update_by = user_id
+    obj.update_time = datetime.now()
+    db.add(obj)
+    db.commit()
+    db.refresh(obj)
+    return row_to_dict(obj)
+
+
+def soft_delete_account(
+    db: Session, account_id: str, tenant_no: str, user_id: Optional[str]
+) -> bool:
+    obj = (
+        db.query(OaReceiveAccount)
+        .filter(
+            OaReceiveAccount.id == account_id,
+            OaReceiveAccount.tenant_no == tenant_no,
+            OaReceiveAccount.del_flag == 0,
+        )
+        .first()
+    )
+    if not obj:
+        return False
+    obj.del_flag = 1
+    obj.update_by = user_id
+    obj.update_time = datetime.now()
+    db.add(obj)
+    db.commit()
+    return True
+
+
+def soft_delete_accounts(
+    db: Session, ids: List[str], tenant_no: str, user_id: Optional[str]
+) -> int:
+    count = 0
+    for account_id in ids:
+        if soft_delete_account(db, account_id, tenant_no, user_id):
+            count += 1
+    return count

backend/app/services/oa_receive_account.py

@@ -0,0 +1,188 @@
+"""OA 表单设计器：收款账户控件 form_data 校验 + 账户字段规范化"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Tuple
+
+ACCOUNT_TYPES = frozenset({"alipay", "personal", "corporate"})
+
+_BANK_FIELDS = ("bank_name", "bank_province", "bank_city", "bank_branch")
+_BANK_FIELD_ALIASES = {
+    "bank_name": ("bankName", "bank_name"),
+    "bank_province": ("bankProvince", "bank_province"),
+    "bank_city": ("bankCity", "bank_city"),
+    "bank_branch": ("bankBranch", "bank_branch"),
+}
+
+
+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 _pick(body: dict, camel: str, snake: str, default: Any = "") -> Any:
+    if camel in body:
+        return body[camel]
+    if snake in body:
+        return body[snake]
+    return default
+
+
+def _pick_str(body: dict, camel: str, snake: str, default: str = "") -> str:
+    raw = _pick(body, camel, snake, default)
+    if raw is None:
+        return default
+    return str(raw).strip()
+
+
+def normalize_account_fields(body: dict) -> Tuple[bool, Optional[str], Dict[str, Any]]:
+    """规范化收款账户写入字段，返回 (ok, error, snake_case dict)。"""
+    if not isinstance(body, dict):
+        return False, "请求体格式不正确", {}
+
+    account_type = _pick_str(body, "accountType", "account_type")
+    account_name = _pick_str(body, "accountName", "account_name")
+    account_no = _pick_str(body, "accountNo", "account_no")
+    bank_name = _pick_str(body, "bankName", "bank_name")
+    bank_province = _pick_str(body, "bankProvince", "bank_province")
+    bank_city = _pick_str(body, "bankCity", "bank_city")
+    bank_branch = _pick_str(body, "bankBranch", "bank_branch")
+    order_num = _pick(body, "orderNum", "order_num")
+    remark = _pick_str(body, "remark", "remark")
+
+    if not account_name:
+        return False, "户名不能为空", {}
+    if not account_no:
+        return False, "账号不能为空", {}
+    if account_type not in ACCOUNT_TYPES:
+        return False, "账户类型无效", {}
+
+    if account_type == "alipay":
+        bank_name = ""
+        bank_province = ""
+        bank_city = ""
+        bank_branch = ""
+    elif not all([bank_name, bank_province, bank_city, bank_branch]):
+        return False, "银行信息不完整", {}
+
+    return True, None, {
+        "account_type": account_type,
+        "account_name": account_name,
+        "account_no": account_no,
+        "bank_name": bank_name,
+        "bank_province": bank_province,
+        "bank_city": bank_city,
+        "bank_branch": bank_branch,
+        "order_num": order_num,
+        "remark": remark,
+    }
+
+
+def _is_non_empty_str(value: Any) -> bool:
+    return isinstance(value, str) and bool(value.strip())
+
+
+def _is_valid_receive_account_item(item: Any) -> bool:
+    if not isinstance(item, dict):
+        return False
+    item_id = item.get("id")
+    account_type = item.get("accountType") or item.get("account_type")
+    account_name = item.get("accountName") or item.get("account_name")
+    account_no = item.get("accountNo") or item.get("account_no")
+    if not _is_non_empty_str(item_id):
+        return False
+    if not _is_non_empty_str(account_type):
+        return False
+    if str(account_type).strip() not in ACCOUNT_TYPES:
+        return False
+    if not _is_non_empty_str(account_name):
+        return False
+    if not _is_non_empty_str(account_no):
+        return False
+    if str(account_type).strip() == "alipay":
+        return True
+    for snake, (camel, _) in _BANK_FIELD_ALIASES.items():
+        val = item.get(camel)
+        if val is None:
+            val = item.get(snake)
+        if not _is_non_empty_str(val):
+            return False
+    return True
+
+
+def _validate_receive_account_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 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_receive_account_item(item):
+            errors.append(f"「{label}」收款账户项格式不正确")
+            break
+    return errors
+
+
+def validate_receive_account_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 "") != "receiveAccount":
+            continue
+        errors.extend(_validate_receive_account_rule(rule, data))
+    return errors

backend/tests/test_oa_receive_account.py

@@ -0,0 +1,104 @@
+import unittest
+
+from app.services.oa_receive_account import (
+    normalize_account_fields,
+    validate_receive_account_form_values,
+)
+
+
+class TestNormalizeAccountFields(unittest.TestCase):
+    def test_alipay_clears_bank(self):
+        ok, err, data = normalize_account_fields(
+            {
+                "accountType": "alipay",
+                "accountName": "张三",
+                "accountNo": "13800138000",
+                "bankName": "工行",
+                "bankProvince": "安徽",
+                "bankCity": "合肥",
+                "bankBranch": "xx支行",
+                "remark": "r",
+            }
+        )
+        self.assertTrue(ok)
+        self.assertIsNone(err)
+        self.assertEqual(data["account_type"], "alipay")
+        self.assertEqual(data["account_name"], "张三")
+        self.assertEqual(data["bank_name"], "")
+        self.assertEqual(data["bank_province"], "")
+        self.assertEqual(data["bank_city"], "")
+        self.assertEqual(data["bank_branch"], "")
+
+    def test_personal_requires_bank(self):
+        ok, err, _ = normalize_account_fields(
+            {
+                "accountType": "personal",
+                "accountName": "李四",
+                "accountNo": "6222",
+            }
+        )
+        self.assertFalse(ok)
+        self.assertTrue(err)
+
+    def test_personal_ok(self):
+        ok, err, data = normalize_account_fields(
+            {
+                "accountType": "personal",
+                "accountName": "李四",
+                "accountNo": "6222",
+                "bankName": "工行",
+                "bankProvince": "安徽省",
+                "bankCity": "合肥市",
+                "bankBranch": "长江中路支行",
+            }
+        )
+        self.assertTrue(ok)
+        self.assertEqual(data["bank_name"], "工行")
+
+
+class TestValidateReceiveAccountFormValues(unittest.TestCase):
+    def test_ok_array(self):
+        schema = {
+            "rule": [
+                {
+                    "type": "receiveAccount",
+                    "field": "acc",
+                    "title": "收款账户",
+                    "$required": True,
+                }
+            ]
+        }
+        data = {
+            "acc": [
+                {
+                    "id": "1",
+                    "accountType": "alipay",
+                    "accountName": "张三",
+                    "accountNo": "138",
+                    "bankName": "",
+                    "bankProvince": "",
+                    "bankCity": "",
+                    "bankBranch": "",
+                    "remark": "",
+                }
+            ]
+        }
+        self.assertEqual(validate_receive_account_form_values(schema, data), [])
+
+    def test_required_empty(self):
+        schema = {
+            "rule": [
+                {"type": "receiveAccount", "field": "acc", "title": "收款账户", "$required": True}
+            ]
+        }
+        errs = validate_receive_account_form_values(schema, {"acc": []})
+        self.assertTrue(any("收款账户" in e for e in errs))
+
+    def test_invalid_item(self):
+        schema = {"rule": [{"type": "receiveAccount", "field": "acc"}]}
+        errs = validate_receive_account_form_values(schema, {"acc": [{"id": "1"}]})
+        self.assertTrue(len(errs) >= 1)
+
+
+if __name__ == "__main__":
+    unittest.main()