zkxw/backend/app/services/oa_post_select.py

"""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_post_item(item: Any) -> bool:
    if not isinstance(item, dict):
        return False
    post_id = item.get("id")
    name = item.get("name")
    if not isinstance(post_id, str) or not post_id.strip():
        return False
    if not isinstance(name, str) or not name.strip():
        return False
    return True


def _validate_post_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_post_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_post_item(value):
        errors.append(f"「{label}」职位字段格式不正确，需包含有效的 id 与 name")
    return errors


def validate_post_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 "") != "post":
            continue
        errors.extend(_validate_post_rule(rule, data))
    return errors


def extract_post_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 post_contains_match(form_value: Any, cond_ids: Any) -> bool:
    form_ids = extract_post_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)