diff --git a/backend/app/crud/oa_receive_account.py b/backend/app/crud/oa_receive_account.py new file mode 100644 index 00000000..b7acb02c --- /dev/null +++ b/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 diff --git a/backend/app/services/oa_receive_account.py b/backend/app/services/oa_receive_account.py new file mode 100644 index 00000000..f76c2b72 --- /dev/null +++ b/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 diff --git a/backend/tests/test_oa_receive_account.py b/backend/tests/test_oa_receive_account.py new file mode 100644 index 00000000..2d4b4985 --- /dev/null +++ b/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()