zkxw/backend/app/services/oa_publish.py

"""OA 审批应用发布校验与发布落库"""
from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.crud import oa_approval as oa_crud
from app.models.models import SysPost
from app.services.oa_assignee import VALID_ASSIGNEE_TYPES, normalize_cc_assignee_types

# 可填控件（FormCreate / 自建控件字典交集；布局类不计）
FILLABLE_TYPES = frozenset(
    {
        "input",
        "textarea",
        "inputNumber",
        "number",
        "radio",
        "checkbox",
        "select",
        "switch",
        "datePicker",
        "date",
        "dateRange",
        "timePicker",
        "time",
        "phone",
        "image",
        "upload",
        "money",
        "dept",
        "person",
        "ownerDept",
        "ownerPerson",
        "post",
        "receiveAccount",
        "region",
        "formula",
        "relatedApproval",
        "rate",
        "slider",
        "cascader",
        "tripItinerary",
    }
)

LAYOUT_TYPES = frozenset(
    {
        "fcRow",
        "col",
        "column",
        "fcTable",
        "html",
        "divider",
        "card",
        "space",
        "tabs",
        "tabPane",
        "group",
        "object",
        "leaveKit",
        "confirmationKit",
        "overtimeKit",
        "resignationKit",
        "tripKit",
    }
)

VALID_NODE_TYPES = frozenset({"start", "approval", "cc", "condition", "end"})

# 系统条件伪字段（非表单控件）
INITIATOR_CONDITION_FIELD = "__initiator__"

# 发布允许的控件 type（未知 type 拒绝）
ALLOWED_CONTROL_TYPES = FILLABLE_TYPES | LAYOUT_TYPES

CONDITION_ELIGIBLE_TYPES = frozenset(
    {
        "input",
        "inputNumber",
        "number",
        "money",
        "radio",
        "select",
        "datePicker",
        "date",
        "person",
        "ownerPerson",
        "post",
        "dept",
        "ownerDept",
    }
)


def _extract_rules(form_schema: Any) -> List[Dict[str, Any]]:
    if form_schema is None:
        return []
    if isinstance(form_schema, list):
        return [r for r in form_schema if isinstance(r, dict)]
    if isinstance(form_schema, dict):
        rules = form_schema.get("rule") or form_schema.get("rules") or form_schema.get("fields")
        if isinstance(rules, list):
            return [r for r in rules if isinstance(r, dict)]
        # 单字段对象
        if form_schema.get("type") or form_schema.get("field"):
            return [form_schema]
    return []


def _walk_form_rules(rules: List[Dict[str, Any]], out: List[Dict[str, Any]]) -> None:
    for rule in rules:
        out.append(rule)
        children = rule.get("children") or rule.get("props", {}).get("rule") or []
        if isinstance(children, list):
            _walk_form_rules([c for c in children if isinstance(c, dict)], out)


def count_fillable_fields(form_schema: Any) -> int:
    flat: List[Dict[str, Any]] = []
    _walk_form_rules(_extract_rules(form_schema), flat)
    n = 0
    for rule in flat:
        t = (rule.get("type") or "").strip()
        if t in LAYOUT_TYPES:
            continue
        field = rule.get("field") or rule.get("name")
        if t in FILLABLE_TYPES or (field and t and t not in LAYOUT_TYPES):
            n += 1
    return n


def _rule_use_in_condition(rule: Dict[str, Any]) -> bool:
    if rule.get("useInCondition") is True:
        return True
    props = rule.get("props") or {}
    return isinstance(props, dict) and props.get("useInCondition") is True


def index_condition_fields(form_schema: Any) -> Dict[str, Dict[str, Any]]:
    """field -> rule（仅 useInCondition 且 type 合格）"""
    flat: List[Dict[str, Any]] = []
    _walk_form_rules(_extract_rules(form_schema), flat)
    out: Dict[str, Dict[str, Any]] = {}
    for rule in flat:
        t = (rule.get("type") or "").strip()
        field = (rule.get("field") or rule.get("name") or "").strip()
        if not field or t not in CONDITION_ELIGIBLE_TYPES:
            continue
        if _rule_use_in_condition(rule):
            out[field] = rule
    return out


def normalize_use_in_condition_required(form_schema: Any) -> Any:
    """就地/深拷贝皆可：打标字段强制 required/$required。发布前调用。"""
    import copy

    schema = copy.deepcopy(form_schema) if form_schema is not None else {"rule": []}

    def walk(list_rules: List[Dict[str, Any]]) -> None:
        for rule in list_rules:
            if _rule_use_in_condition(rule):
                rule["useInCondition"] = True
                rule["$required"] = True
                rule["required"] = True
                props = rule.get("props")
                if isinstance(props, dict):
                    props["useInCondition"] = True
            children = rule.get("children") or (rule.get("props") or {}).get("rule") or []
            if isinstance(children, list):
                walk([c for c in children if isinstance(c, dict)])

    if isinstance(schema, list):
        walk([r for r in schema if isinstance(r, dict)])
    elif isinstance(schema, dict):
        r = schema.get("rule")
        if isinstance(r, list):
            walk([x for x in r if isinstance(x, dict)])
    return schema


def _node_type(node: Dict[str, Any]) -> str:
    return (node.get("type") or node.get("nodeType") or "").strip()


def validate_flow_tree(flow_tree: Any) -> List[str]:
    errors: List[str] = []
    if not flow_tree or not isinstance(flow_tree, dict):
        return ["流程树不能为空"]

    root = flow_tree
    # 兼容 { root: node } 包装
    if "root" in flow_tree and isinstance(flow_tree["root"], dict):
        root = flow_tree["root"]

    if _node_type(root) != "start":
        errors.append("流程树根节点必须是发起人(start)")

    seen_end = False

    def walk(node: Optional[Dict[str, Any]], path: str) -> None:
        nonlocal seen_end
        if not node or not isinstance(node, dict):
            errors.append(f"{path}：节点缺失")
            return
        ntype = _node_type(node)
        if ntype not in VALID_NODE_TYPES:
            errors.append(f"{path}：不支持的节点类型「{ntype or '(空)'}」")
            return

        if ntype == "end":
            seen_end = True
            if node.get("child"):
                errors.append(f"{path}：结束节点后不应再有子节点")
            return

        if ntype == "approval":
            cfg = node.get("approverConfig") or node.get("approver_config") or {}
            if not isinstance(cfg, dict):
                cfg = {}
            # 审批只认单字段 assigneeType（忽略 assigneeTypes）
            at = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
            if not at:
                errors.append(f"{path}：未配置审批人/抄送人策略")
            elif at not in VALID_ASSIGNEE_TYPES:
                errors.append(f"{path}：不支持的审批人类型「{at}」")
            elif at == "users":
                ids = cfg.get("userIds") or cfg.get("user_ids") or []
                if not ids:
                    errors.append(f"{path}：指定成员不能为空")
            elif at == "posts":
                pids = cfg.get("postIds") or cfg.get("post_ids") or []
                if not pids:
                    errors.append(f"{path}：职位不能为空")

        if ntype == "cc":
            cfg = node.get("approverConfig") or node.get("approver_config") or {}
            if not isinstance(cfg, dict):
                cfg = {}
            types = normalize_cc_assignee_types(cfg)
            if not types:
                errors.append(f"{path}：未配置审批人/抄送人策略")
            else:
                for at in types:
                    if at not in VALID_ASSIGNEE_TYPES:
                        errors.append(f"{path}：不支持的审批人类型「{at}」")
                    elif at == "users":
                        ids = cfg.get("userIds") or cfg.get("user_ids") or []
                        if not ids:
                            errors.append(f"{path}：指定成员不能为空")
                    elif at == "posts":
                        pids = cfg.get("postIds") or cfg.get("post_ids") or []
                        if not pids:
                            errors.append(f"{path}：职位不能为空")

        if ntype == "condition":
            groups = node.get("conditionGroups") or node.get("condition_groups") or []
            if not isinstance(groups, list) or len(groups) < 1:
                errors.append(f"{path}：条件分支至少需要一组出线")
            else:
                for i, g in enumerate(groups):
                    if not isinstance(g, dict):
                        errors.append(f"{path}.条件组[{i}]：格式非法")
                        continue
                    is_default = bool(g.get("isDefault"))
                    rules = g.get("rules") if isinstance(g.get("rules"), list) else []
                    when = (g.get("when") or "").strip() if isinstance(g.get("when"), str) else ""
                    valid_rules = [
                        r
                        for r in rules
                        if isinstance(r, dict)
                        and (r.get("field") or "").strip()
                        and (r.get("op") or "").strip()
                        and r.get("value") is not None
                        and str(r.get("value")).strip() != ""
                    ]
                    if is_default:
                        pass
                    elif valid_rules:
                        pass
                    elif when and not valid_rules:
                        errors.append(
                            f"{path}.条件组[{i}]：请用可视化规则重新配置条件（不再支持仅公式）"
                        )
                    else:
                        errors.append(f"{path}.条件组[{i}]：请完善条件分支规则")
                    child = g.get("child")
                    if not child:
                        errors.append(f"{path}.条件组[{i}]：缺少下游节点")
                    else:
                        gname = g.get("name") or f"分支{i + 1}"
                        walk(child if isinstance(child, dict) else None, f"{path}/{gname}")
            # 条件分支汇合后的主链（可选）
            merge = node.get("child")
            if merge is not None:
                if not isinstance(merge, dict):
                    errors.append(f"{path}：汇合后子节点格式非法")
                else:
                    walk(merge, f"{path}/汇合")
            return

        child = node.get("child")
        if child is None:
            if ntype != "end":
                errors.append(f"{path}：须连接到后续节点直至结束")
            return
        if not isinstance(child, dict):
            errors.append(f"{path}：子节点格式非法")
            return
        cname = child.get("name") or _node_type(child) or "next"
        walk(child, f"{path}/{cname}")

    walk(root, root.get("name") or "发起人")
    if not seen_end:
        errors.append("流程树未能到达结束节点")
    return errors


def validate_condition_rules_against_form(form_schema: Any, flow_tree: Any) -> List[str]:
    indexed = index_condition_fields(form_schema)
    errors: List[str] = []
    seen_form: Set[str] = set()

    def collect_fields(node: Optional[Dict[str, Any]]) -> None:
        if not node or not isinstance(node, dict):
            return
        for g in node.get("conditionGroups") or node.get("condition_groups") or []:
            if not isinstance(g, dict) or g.get("isDefault"):
                continue
            rules = g.get("rules") if isinstance(g.get("rules"), list) else []
            for r in rules:
                if not isinstance(r, dict):
                    continue
                field = (r.get("field") or "").strip()
                if not field:
                    continue
                if field == INITIATOR_CONDITION_FIELD:
                    op = (r.get("op") or "").strip()
                    val = r.get("value")
                    if op != "contains":
                        errors.append("发起人条件仅支持「包含」")
                    elif not isinstance(val, list) or not any(
                        str(x).strip() for x in val
                    ):
                        errors.append("发起人包含条件须选择至少一名用户")
                    continue
                if field in seen_form:
                    continue
                seen_form.add(field)
                if field not in indexed:
                    errors.append(f"条件引用了无效字段：{field}")
            child = g.get("child")
            if isinstance(child, dict):
                collect_fields(child)
        child = node.get("child")
        if isinstance(child, dict):
            collect_fields(child)

    root = flow_tree
    if isinstance(flow_tree, dict) and isinstance(flow_tree.get("root"), dict):
        root = flow_tree["root"]
    collect_fields(root if isinstance(root, dict) else None)
    return errors


def validate_posts_referenced(
    db: Session, *, tenant_no: str, flow_tree: Any
) -> List[str]:
    post_ids: Set[str] = set()

    def collect(node: Optional[Dict[str, Any]]) -> None:
        if not node or not isinstance(node, dict):
            return
        cfg = node.get("approverConfig") or {}
        if isinstance(cfg, dict):
            types: List[str] = []
            ntype = _node_type(node)
            if ntype == "cc":
                types = normalize_cc_assignee_types(cfg)
            else:
                at = (cfg.get("assigneeType") or "").strip()
                if at:
                    types = [at]
            if "posts" in types:
                for pid in cfg.get("postIds") or cfg.get("post_ids") or []:
                    if pid:
                        post_ids.add(str(pid))
        child = node.get("child")
        if isinstance(child, dict):
            collect(child)
        for g in node.get("conditionGroups") or []:
            if isinstance(g, dict) and isinstance(g.get("child"), dict):
                collect(g["child"])

    root = flow_tree
    if isinstance(flow_tree, dict) and isinstance(flow_tree.get("root"), dict):
        root = flow_tree["root"]
    collect(root if isinstance(root, dict) else None)

    errors: List[str] = []
    if post_ids:
        rows = (
            db.query(SysPost)
            .filter(
                SysPost.id.in_(list(post_ids)),
                SysPost.tenant_no == tenant_no,
                SysPost.del_flag == 0,
            )
            .all()
        )
        found = {r.id: r for r in rows}
        for pid in post_ids:
            p = found.get(pid)
            if not p:
                errors.append(f"引用的职位不存在：{pid}")
            elif p.status != "1":
                errors.append(f"职位已停用：{p.post_name or pid}")
    return errors


def validate_control_types(form_schema: Any) -> List[str]:
    flat: List[Dict[str, Any]] = []
    _walk_form_rules(_extract_rules(form_schema), flat)
    errors: List[str] = []
    unknown: Set[str] = set()
    for rule in flat:
        t = (rule.get("type") or "").strip()
        if not t:
            continue
        if t not in ALLOWED_CONTROL_TYPES:
            unknown.add(t)
    for t in sorted(unknown):
        errors.append(f"不支持的表单控件类型：{t}")
    return errors


def validate_draft_for_publish(
    db: Session,
    *,
    tenant_no: str,
    form_schema: Any,
    flow_tree: Any,
    basic: Optional[Dict[str, Any]] = None,
) -> List[str]:
    errors: List[str] = []
    app_name = ""
    if basic and isinstance(basic, dict):
        app_name = (basic.get("appName") or "").strip()
    if not app_name:
        # 校验侧也可能只传 schema；名称在 publish 时再查 app
        pass
    if count_fillable_fields(form_schema) < 1:
        errors.append("表单至少需要 1 个可填字段")
    form_schema = normalize_use_in_condition_required(form_schema)
    errors.extend(validate_control_types(form_schema))
    errors.extend(validate_flow_tree(flow_tree))
    errors.extend(validate_condition_rules_against_form(form_schema, flow_tree))
    errors.extend(validate_posts_referenced(db, tenant_no=tenant_no, flow_tree=flow_tree))
    return errors


def publish_app(
    db: Session,
    app_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> Tuple[Dict[str, Any], List[str]]:
    """
    校验并发布当前草稿。
    成功返回 (draft_dict_like_published_info, [])；
    失败返回 ({}, errors)。
    """
    app = oa_crud.get_app(db, app_id, tenant_no)
    if not app:
        return {}, ["应用不存在"]
    if not (app.app_name or "").strip():
        return {}, ["应用名称不能为空"]

    draft_payload = oa_crud.get_or_create_draft(db, app_id, tenant_no=tenant_no, user_id=user_id)
    form_schema = draft_payload.get("formSchemaJson")
    form_schema = normalize_use_in_condition_required(form_schema)
    flow_tree = draft_payload.get("flowTreeJson")
    errors = validate_draft_for_publish(
        db,
        tenant_no=tenant_no,
        form_schema=form_schema,
        flow_tree=flow_tree,
        basic=draft_payload.get("basic"),
    )
    if errors:
        return {}, errors

    draft = oa_crud._find_draft(db, app_id, tenant_no)
    if not draft:
        return {}, ["草稿不存在"]

    draft.form_schema_json = oa_crud._dumps(form_schema)

    from datetime import datetime

    from app.models.models import OaAppVersion

    now = datetime.now()
    olds = (
        db.query(OaAppVersion)
        .filter(
            OaAppVersion.app_id == app_id,
            OaAppVersion.tenant_no == tenant_no,
            OaAppVersion.is_latest_published == "1",
            OaAppVersion.del_flag == 0,
        )
        .all()
    )
    for old in olds:
        old.is_latest_published = "0"
        old.update_by = user_id
        old.update_time = now
        db.add(old)

    draft.status = "published"
    draft.is_latest_published = "1"
    draft.update_by = user_id
    draft.update_time = now
    db.add(draft)

    # 发布后不应再残留其它草稿（并发/重复创建等）；否则列表 hasDraft 仍为 true
    leftover_drafts = (
        db.query(OaAppVersion)
        .filter(
            OaAppVersion.app_id == app_id,
            OaAppVersion.tenant_no == tenant_no,
            OaAppVersion.status == "draft",
            OaAppVersion.del_flag == 0,
            OaAppVersion.id != draft.id,
        )
        .all()
    )
    for leftover in leftover_drafts:
        leftover.del_flag = 1
        leftover.update_by = user_id
        leftover.update_time = now
        db.add(leftover)

    app.update_by = user_id
    app.update_time = now
    db.add(app)
    db.commit()
    db.refresh(draft)

    return {
        "appId": app.id,
        "versionId": draft.id,
        "versionNo": draft.version_no,
        "status": draft.status,
        "isLatestPublished": draft.is_latest_published,
        "appName": app.app_name,
    }, []