zkxw/backend/app/services/oa_receive_account.py

"""OA 表单设计器：收款账户控件 form_data 校验 + 账户字段规范化"""
from __future__ import annotations

from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from app.models.models import OaReceiveAccount

ACCOUNT_TYPES = frozenset({"alipay", "personal", "corporate"})

_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 _pick_order_num(body: dict) -> Optional[int]:
    raw = _pick(body, "orderNum", "order_num", None)
    if raw is None:
        return None
    if isinstance(raw, str):
        raw = raw.strip()
        if raw == "":
            return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


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_order_num(body)
    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


def validate_receive_account_ownership(
    db: Session,
    schema: Any,
    form_data: Optional[Dict[str, Any]],
    *,
    tenant_no: str,
    initiator_user_id: str,
) -> List[str]:
    """校验收款账户快照 id 属于发起人。调用前须已通过格式校验。"""
    data = form_data or {}
    owner_id = (initiator_user_id or "").strip()
    for rule in _schema_rules(schema):
        if str(rule.get("type") or "") != "receiveAccount":
            continue
        field = str(rule.get("field") or "").strip()
        if not field:
            continue
        value = data.get(field)
        if not isinstance(value, list):
            continue
        label = _field_label(rule)
        for item in value:
            if not isinstance(item, dict):
                continue
            item_id = str(item.get("id") or "").strip()
            if not item_id:
                continue
            row = (
                db.query(OaReceiveAccount)
                .filter(
                    OaReceiveAccount.id == item_id,
                    OaReceiveAccount.tenant_no == tenant_no,
                )
                .first()
            )
            if row is None or int(row.del_flag or 0) == 1:
                return [f"「{label}」不存在或已删除"]
            if (row.create_by or "").strip() != owner_id:
                return [f"「{label}」不属于当前用户"]
    return []