zkxw/backend/app/services/oa_engine.py

"""OA 审批运行时引擎（钉钉式竖向树）"""
from __future__ import annotations

import json
import re
from collections import OrderedDict
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.crud import oa_approval as oa_crud
from app.models.models import (
    OaApp,
    OaAppVersion,
    OaInstance,
    OaInstanceEvent,
    OaRelation,
    OaRole,
    OaTask,
    SysDept,
    SysUser,
)
from app.services.oa_assignee import AssigneeResolveError, resolve_assignees
from app.services.oa_formula import FormulaError, apply_formulas_and_linkages, eval_formula


class OaEngineError(ValueError):
    """引擎业务错误"""


_AUTO_PASS_COMMENT = "系统自动通过"


def _loads(text: Optional[str], default: Any = None) -> Any:
    if text is None or text == "":
        return default
    if not isinstance(text, str):
        return text
    try:
        return json.loads(text)
    except (TypeError, ValueError):
        return default


def _dumps(data: Any) -> str:
    return json.dumps(data if data is not None else {}, ensure_ascii=False)


def _normalize_initiator_select_by_node(raw: Any) -> Dict[str, List[str]]:
    if not isinstance(raw, dict):
        return {}
    out: Dict[str, List[str]] = {}
    for k, v in raw.items():
        nid = str(k or "").strip()
        if not nid or not isinstance(v, (list, tuple)):
            continue
        ids: List[str] = []
        seen = set()
        for x in v:
            uid = str(x or "").strip()
            if uid and uid not in seen:
                seen.add(uid)
                ids.append(uid)
        out[nid] = ids
    return out


def _ids_for_initiator_select_node(
    select_by_node: Optional[Dict[str, List[str]]], node_id: str
) -> List[str]:
    m = select_by_node or {}
    nid = str(node_id or "").strip()
    if not nid:
        return []
    return list(m.get(nid) or [])


def _load_initiator_select_by_node(inst: Any) -> Dict[str, List[str]]:
    return _normalize_initiator_select_by_node(
        _loads(getattr(inst, "initiator_select_json", None), {}) or {}
    )


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


def _subtree_contains(node: Optional[Dict[str, Any]], node_id: str) -> bool:
    """节点子树（含自身、child、条件分支）是否包含指定 id。"""
    if not isinstance(node, dict):
        return False
    if (node.get("id") or "") == node_id:
        return True
    child = node.get("child")
    if isinstance(child, dict) and _subtree_contains(child, node_id):
        return True
    for g in node.get("conditionGroups") or []:
        if isinstance(g, dict) and _subtree_contains(g.get("child"), node_id):
            return True
    return False


def _merge_after_branch_end(
    root: Dict[str, Any], end_id: str
) -> Optional[Dict[str, Any]]:
    """若 end 落在某条件分支组内，且该条件节点有汇合后 child，则返回汇合节点。"""

    def walk(node: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
        if not isinstance(node, dict):
            return None
        if _node_type(node) == "condition":
            for g in node.get("conditionGroups") or []:
                if not isinstance(g, dict):
                    continue
                if _subtree_contains(g.get("child"), end_id):
                    merge = node.get("child")
                    return merge if isinstance(merge, dict) else None
            merge_child = node.get("child")
            if isinstance(merge_child, dict):
                return walk(merge_child)
            return None
        child = node.get("child")
        if isinstance(child, dict):
            return walk(child)
        return None

    return walk(root)


def _flow_root(flow_tree: Any) -> Dict[str, Any]:
    if not isinstance(flow_tree, dict):
        raise OaEngineError("流程树无效")
    if isinstance(flow_tree.get("root"), dict):
        return flow_tree["root"]
    return flow_tree


def _find_node(root: Dict[str, Any], node_key: str) -> Optional[Dict[str, Any]]:
    if not root or not node_key:
        return None
    if (root.get("id") or "") == node_key:
        return root
    child = root.get("child")
    if isinstance(child, dict):
        found = _find_node(child, node_key)
        if found:
            return found
    for g in root.get("conditionGroups") or []:
        if isinstance(g, dict) and isinstance(g.get("child"), dict):
            found = _find_node(g["child"], node_key)
            if found:
                return found
    return None


def _flow_node_key_order(root: Dict[str, Any]) -> Dict[str, int]:
    """按流程树先序遍历为 node.id 分配稳定序号（用于详情时间轴排序）。"""
    order: Dict[str, int] = {}
    counter = 0

    def walk(node: Any) -> None:
        nonlocal counter
        if not isinstance(node, dict):
            return
        nid = (node.get("id") or "").strip()
        if nid and nid not in order:
            order[nid] = counter
            counter += 1
        if _node_type(node) == "condition":
            for g in node.get("conditionGroups") or []:
                if isinstance(g, dict) and isinstance(g.get("child"), dict):
                    walk(g["child"])
        child = node.get("child")
        if isinstance(child, dict):
            walk(child)

    walk(root)
    return order


def _next_task_order_num(db: Session, instance_id: str) -> int:
    """同一实例内任务展示序号取 max+1（无任务从 1 起）。"""
    rows = (
        db.query(OaTask.order_num)
        .filter(OaTask.instance_id == instance_id, OaTask.del_flag == 0)
        .all()
    )
    vals = [int(r[0]) for r in rows if r[0] is not None]
    return (max(vals) + 1) if vals else 1


def _sort_tasks_for_detail(
    tasks: List[OaTask], *, flow_root: Optional[Dict[str, Any]]
) -> List[OaTask]:
    """详情时间轴：按流程树节点顺序，再 order_num / create_time（兼容旧数据同秒创建）。"""
    node_order = _flow_node_key_order(flow_root) if isinstance(flow_root, dict) else {}
    far = 10**9

    def key(t: OaTask) -> Tuple[int, int, int, datetime, str]:
        nk = (t.node_key or "").strip()
        return (
            int(getattr(t, "round_no", None) or 1),
            node_order.get(nk, far),
            int(t.order_num) if t.order_num is not None else 0,
            t.create_time or datetime.min,
            t.id or "",
        )

    return sorted(tasks, key=key)


def _fmt_dt(dt: Any) -> str:
    if not dt:
        return ""
    if hasattr(dt, "strftime"):
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    return str(dt)


def _terminal_item(status: str, title: str, inst: Any) -> Dict[str, Any]:
    return {
        "kind": "terminal",
        "roundNo": getattr(inst, "current_round", None) or 1,
        "title": title,
        "status": status,
        "assigneeUserId": "",
        "assigneeUserName": "",
        "timestamp": _fmt_dt(getattr(inst, "update_time", None)),
        "commentText": "",
        "nodeKey": "",
        "taskType": "",
    }


def _task_round(t: Any) -> int:
    v = getattr(t, "round_no", None)
    try:
        if v is None:
            return 1
        return int(v)
    except (TypeError, ValueError):
        return 1


def _append_pending_nodes(
    items: List[Dict[str, Any]],
    *,
    flow_root: Dict[str, Any],
    reached_keys: set,
    round_no: int,
) -> None:
    """兜底：按整棵树先序挂未到达节点（无路径解析 / 无人员）。"""
    order = _flow_node_key_order(flow_root)
    for nid, _idx in sorted(order.items(), key=lambda x: x[1]):
        node = _find_node(flow_root, nid)
        if not node:
            continue
        nt = _node_type(node)
        if nt not in ("approval", "cc", "end") or nid in reached_keys:
            continue
        if nt == "end":
            title = node.get("name") or "结束"
        elif nt == "cc":
            title = node.get("name") or "抄送人"
        else:
            title = node.get("name") or "审批人"
        items.append(
            {
                "kind": "pending_node",
                "roundNo": round_no,
                "title": title,
                "status": "not_reached",
                "assigneeUserId": "",
                "assigneeUserName": "",
                "timestamp": "",
                "commentText": "",
                "nodeKey": nid,
                "taskType": nt,
                "signMode": "",
                "people": [],
            }
        )


def _pending_people_for_node(
    db: Session,
    *,
    node: Dict[str, Any],
    tenant_no: str,
    initiator_user_id: str,
    initiator_dept_id: Optional[str],
    advanced: Dict[str, Any],
    initiator_select_by_node: Optional[Dict[str, List[str]]],
) -> Tuple[List[Dict[str, Any]], str, str]:
    """解析未到达节点的展示人员与签批模式。返回 (people, sign_mode, resolve_source)。"""
    ntype = _node_type(node)
    cfg = node.get("approverConfig") or {}
    assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
    sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "").strip()
    if ntype == "cc":
        sign_mode = "cc"
    elif not sign_mode and assignee_type != "initiator_select":
        sign_mode = "or"
    people: List[Dict[str, Any]] = []
    node_select_ids = _ids_for_initiator_select_node(
        initiator_select_by_node, node.get("id") or ""
    )
    if assignee_type == "initiator_select":
        for uid in node_select_ids:
            people.append(
                {
                    "id": uid,
                    "nickName": "",  # 下方统一补名
                    "status": "not_reached",
                    "timestamp": "",
                    "commentText": "",
                    "taskId": "",
                }
            )
    else:
        try:
            assignees, _meta = resolve_assignees(
                db,
                tenant_no=tenant_no,
                assignee_config=cfg,
                initiator_user_id=initiator_user_id,
                initiator_dept_id=initiator_dept_id,
                advanced=advanced,
                initiator_select_user_ids=node_select_ids,
            )
            for uid in assignees:
                people.append(
                    {
                        "id": uid,
                        "nickName": "",
                        "status": "not_reached",
                        "timestamp": "",
                        "commentText": "",
                        "taskId": "",
                    }
                )
        except AssigneeResolveError:
            people = []
    if people:
        name_map = _user_name_map(db, [p["id"] for p in people if p.get("id")])
        for p in people:
            p["nickName"] = name_map.get(p["id"], "") or p["id"]
    return people, sign_mode, assignee_type


def _iter_resolved_path_nodes(
    root: Optional[Dict[str, Any]], form_data: Optional[Dict[str, Any]]
) -> Optional[List[Dict[str, Any]]]:
    """沿条件命中路径收集节点；无法匹配时返回 None。"""
    if not isinstance(root, dict):
        return []
    data = form_data or {}
    out: List[Dict[str, Any]] = []
    cur: Optional[Dict[str, Any]] = root
    while cur and isinstance(cur, dict):
        ntype = _node_type(cur)
        if ntype == "end":
            out.append(cur)
            break
        if ntype == "condition":
            g = _match_condition_group(cur.get("conditionGroups") or [], data)
            if not g:
                return None
            child = g.get("child")
            if isinstance(child, dict):
                sub = _iter_resolved_path_nodes(child, data)
                if sub is None:
                    return None
                out.extend(sub)
            cur = cur.get("child")
            continue
        out.append(cur)
        cur = cur.get("child")
    return out


def _append_pending_nodes_on_path(
    items: List[Dict[str, Any]],
    *,
    db: Session,
    flow_root: Dict[str, Any],
    reached_keys: set,
    round_no: int,
    tenant_no: str,
    initiator_user_id: str,
    initiator_dept_id: Optional[str],
    advanced: Dict[str, Any],
    form_data: Optional[Dict[str, Any]],
    initiator_select_by_node: Optional[Dict[str, List[str]]],
) -> None:
    """按命中路径列出全部后续审批/抄送/结束节点（含人员），对齐发起预览。"""
    path = _iter_resolved_path_nodes(flow_root, form_data)
    if path is None:
        already = {
            (x.get("nodeKey") or "").strip()
            for x in items
            if x.get("kind") in ("task", "pending_node") and (x.get("nodeKey") or "").strip()
        }
        _append_pending_nodes(
            items,
            flow_root=flow_root,
            reached_keys=reached_keys | already,
            round_no=round_no,
        )
        return

    for node in path:
        ntype = _node_type(node)
        if ntype not in ("approval", "cc", "end"):
            continue
        nid = (node.get("id") or "").strip()
        if not nid or nid in reached_keys:
            continue
        if ntype == "end":
            items.append(
                {
                    "kind": "pending_node",
                    "roundNo": round_no,
                    "title": node.get("name") or "结束",
                    "status": "not_reached",
                    "assigneeUserId": "",
                    "assigneeUserName": "",
                    "timestamp": "",
                    "commentText": "",
                    "nodeKey": nid,
                    "taskType": "end",
                    "signMode": "",
                    "people": [],
                }
            )
            continue
        people, sign_mode, _src = _pending_people_for_node(
            db,
            node=node,
            tenant_no=tenant_no,
            initiator_user_id=initiator_user_id,
            initiator_dept_id=initiator_dept_id,
            advanced=advanced,
            initiator_select_by_node=initiator_select_by_node,
        )
        items.append(
            {
                "kind": "pending_node",
                "roundNo": round_no,
                "title": node.get("name") or ("抄送人" if ntype == "cc" else "审批人"),
                "status": "not_reached",
                "assigneeUserId": "",
                "assigneeUserName": "",
                "timestamp": "",
                "commentText": "",
                "nodeKey": nid,
                "taskType": ntype,
                "signMode": sign_mode,
                "people": people,
            }
        )


def _progress_person(t: Any, name_map: Dict[str, str]) -> Dict[str, Any]:
    uid = (getattr(t, "assignee_user_id", None) or "").strip()
    return {
        "id": uid,
        "nickName": name_map.get(uid, "") or uid,
        "status": getattr(t, "status", None) or "",
        "timestamp": _fmt_dt(getattr(t, "update_time", None) or getattr(t, "create_time", None)),
        "commentText": getattr(t, "comment_text", None) or "",
        "taskId": getattr(t, "id", None) or "",
        "transferToUserId": "",
        "transferToUserName": "",
    }


def _enrich_transfer_targets(
    people: List[Dict[str, Any]],
    group: List[Any],
    name_map: Dict[str, str],
) -> None:
    """为已转交人员补齐 transferTo*：同节点 resolve_source=transfer 且 create_by=转交人 的新任务处理人。"""
    by_creator: Dict[str, List[Any]] = {}
    for t in group:
        if (getattr(t, "resolve_source", None) or "").strip() != "transfer":
            continue
        creator = (getattr(t, "create_by", None) or "").strip()
        if not creator:
            continue
        by_creator.setdefault(creator, []).append(t)

    for p in people:
        if (p.get("status") or "") != "transferred":
            continue
        uid = (p.get("id") or "").strip()
        if not uid:
            continue
        candidates = by_creator.get(uid) or []
        if not candidates:
            continue
        # 连续转交时同一人可能只对应一次；若有多条取创建时间最接近转交时间的
        p_ts = (p.get("timestamp") or "").strip()

        def _score(t: Any) -> tuple:
            cts = _fmt_dt(getattr(t, "create_time", None))
            # 优先同秒/最近；无时间时靠后
            if p_ts and cts:
                return (0, abs(_dt_sort_key(cts) - _dt_sort_key(p_ts)), cts)
            return (1, 0, cts or "")

        target_task = sorted(candidates, key=_score)[0]
        tid = (getattr(target_task, "assignee_user_id", None) or "").strip()
        if not tid:
            continue
        p["transferToUserId"] = tid
        p["transferToUserName"] = name_map.get(tid, "") or tid


def _dt_sort_key(s: str) -> int:
    """将 'YYYY-MM-DD HH:MM:SS' 粗转为可比较整数；非法则 0。"""
    raw = (s or "").strip().replace("-", "").replace(":", "").replace(" ", "")
    try:
        return int(raw[:14]) if raw else 0
    except ValueError:
        return 0


def _aggregate_node_status(statuses: List[str]) -> str:
    """同节点多人任务收成卡片状态：拒绝/打回优先，其次待办，再已同意。"""
    sset = [str(s or "").strip() for s in statuses]
    if any(s == "rejected" for s in sset):
        return "rejected"
    if any(s == "returned" for s in sset):
        return "returned"
    if any(s in ("pending", "running") for s in sset):
        return "pending"
    if any(s == "approved" for s in sset):
        return "approved"
    if any(s == "completed" for s in sset):
        return "completed"
    if sset and all(s == "done" for s in sset):
        return "done"
    return sset[0] if sset else "pending"


def _group_tasks_by_node(sorted_tasks: List[Any]) -> List[List[Any]]:
    """按 node_key 合并同节点多人（或签/会签/抄送），无 key 的任务各自独立。"""
    buckets: OrderedDict[str, List[Any]] = OrderedDict()
    for t in sorted_tasks:
        nk = (getattr(t, "node_key", None) or "").strip()
        key = nk if nk else f"__task_{(getattr(t, 'id', None) or id(t))}"
        buckets.setdefault(key, []).append(t)
    return list(buckets.values())


def build_progress_items(
    *,
    inst: Any,
    flow_root: Optional[Dict[str, Any]],
    tasks: List[Any],
    name_map: Dict[str, str],
    events: Optional[List[Any]] = None,
    db: Optional[Session] = None,
    advanced: Optional[Dict[str, Any]] = None,
    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
) -> List[Dict[str, Any]]:
    """投影审批进度：发起人 → 树序 tasks（同节点多人合并）→ 未到达；终态收尾。

    钉钉对齐：打回结果落在审批任务节点上，不另插 kind=return / terminal 终态「已打回」行；不做同单多轮分段。
    或签/会签：同一 nodeKey 只出一张卡片，people 列出各审批人及状态。
    进行中：按命中路径把后续审批/抄送节点全部挂为未到达（尽量解析人员）。
    """
    events = list(events or [])
    sorted_tasks = _sort_tasks_for_detail(list(tasks), flow_root=flow_root)
    # 展示压成单轴（roundNo 统一为 1）；历史多轮任务仍按树序合并展示
    rn = 1
    submit_ts = _fmt_dt(getattr(inst, "create_time", None))
    for e in events:
        if (getattr(e, "event_type", None) or "") == "submit":
            submit_ts = _fmt_dt(getattr(e, "create_time", None)) or submit_ts
            break

    items: List[Dict[str, Any]] = [
        {
            "kind": "submit",
            "roundNo": rn,
            "title": "发起人",
            "status": "submitted",
            "assigneeUserId": inst.initiator_user_id,
            "assigneeUserName": name_map.get(inst.initiator_user_id or "", ""),
            "timestamp": submit_ts,
            "commentText": "",
            "nodeKey": "",
            "taskType": "",
            "signMode": "",
            "people": [],
        }
    ]
    reached_keys = {
        (t.node_key or "").strip() for t in sorted_tasks if (t.node_key or "").strip()
    }
    for group in _group_tasks_by_node(sorted_tasks):
        head = group[0]
        people = [_progress_person(t, name_map) for t in group]
        _enrich_transfer_targets(people, group, name_map)
        statuses = [p["status"] for p in people]
        node_status = _aggregate_node_status(statuses)
        # 时间：优先取已操作人最新时间，否则取组内最新
        acted = [
            p
            for p in people
            if p["status"] in ("approved", "rejected", "returned", "completed", "transferred")
        ]
        ts_src = acted or people
        timestamp = ""
        for p in ts_src:
            if p.get("timestamp") and (not timestamp or p["timestamp"] > timestamp):
                timestamp = p["timestamp"]
        comments = [
            f"{p['nickName']}：{p['commentText']}"
            if len(people) > 1 and p.get("nickName")
            else p["commentText"]
            for p in people
            if (p.get("commentText") or "").strip()
        ]
        sign_mode = (getattr(head, "sign_mode", None) or "").strip()
        primary = next(
            (p for p in people if p["status"] == node_status),
            people[0] if people else None,
        )
        items.append(
            {
                "kind": "task",
                "roundNo": rn,
                "title": head.node_name or head.node_key or "-",
                "status": node_status,
                "assigneeUserId": (primary or {}).get("id", "") if len(people) == 1 else "",
                "assigneeUserName": (primary or {}).get("nickName", "")
                if len(people) == 1
                else "",
                "timestamp": timestamp,
                "commentText": "；".join(comments),
                "nodeKey": head.node_key or "",
                "taskType": head.task_type or "",
                "resolveSource": getattr(head, "resolve_source", None) or "",
                "signMode": sign_mode,
                "taskId": getattr(head, "id", None) or "",
                "people": people,
            }
        )
    if inst.status == "running" and isinstance(flow_root, dict):
        form_data = _loads(getattr(inst, "form_data_json", None), {}) or {}
        if not isinstance(form_data, dict):
            form_data = {}
        # 自选人：优先入参 → 实例 JSON → 旧数据从任务按 node_key 聚合
        select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
        if not select_by_node:
            select_by_node = _load_initiator_select_by_node(inst)
        if not select_by_node:
            for t in sorted_tasks:
                src = (getattr(t, "resolve_source", None) or "").strip()
                uid = (getattr(t, "assignee_user_id", None) or "").strip()
                nk = (getattr(t, "node_key", None) or "").strip()
                if src == "initiator_select" and uid and nk:
                    bucket = select_by_node.setdefault(nk, [])
                    if uid not in bucket:
                        bucket.append(uid)
        if db is not None:
            _append_pending_nodes_on_path(
                items,
                db=db,
                flow_root=flow_root,
                reached_keys=reached_keys,
                round_no=rn,
                tenant_no=getattr(inst, "tenant_no", None) or "",
                initiator_user_id=inst.initiator_user_id or "",
                initiator_dept_id=getattr(inst, "initiator_dept_id", None),
                advanced=advanced or dict(oa_crud.DEFAULT_ADVANCED_SETTINGS),
                form_data=form_data,
                initiator_select_by_node=select_by_node,
            )
        else:
            _append_pending_nodes(
                items, flow_root=flow_root, reached_keys=reached_keys, round_no=rn
            )

    if inst.status == "completed":
        items.append(_terminal_item("done", "流程结束", inst))
    elif inst.status == "rejected":
        items.append(_terminal_item("rejected", "流程结束（已拒绝）", inst))
    elif inst.status == "cancelled":
        items.append(_terminal_item("cancelled", "已撤销", inst))
    # returned：打回结果只落在审批任务节点上，不再追加独立终态「已打回」行
    return items


def _add_instance_event(
    db: Session,
    *,
    inst: OaInstance,
    event_type: str,
    actor_user_id: Optional[str],
    round_no: int,
    comment: Optional[str] = None,
    task_id: Optional[str] = None,
    node_key: Optional[str] = None,
) -> OaInstanceEvent:
    now = datetime.now()
    ev = OaInstanceEvent(
        id=generate_uuid(),
        instance_id=inst.id,
        round_no=round_no,
        event_type=event_type,
        actor_user_id=actor_user_id,
        task_id=task_id,
        node_key=node_key,
        comment_text=(comment or "").strip() or None,
        tenant_no=inst.tenant_no,
        order_num=0,
        remark=None,
        create_by=actor_user_id,
        create_dept_id=inst.initiator_dept_id or "",
        create_time=now,
        update_by=actor_user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(ev)
    return ev


def _advanced(ver: OaAppVersion) -> Dict[str, Any]:
    data = _loads(ver.advanced_settings_json, {}) or {}
    if not isinstance(data, dict):
        return dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
    merged = dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
    merged.update(data)
    return merged


def _check_initiator_scope(app: OaApp, user_id: str) -> None:
    scope = (app.initiator_scope or "all").strip()
    if scope == "all":
        return
    ids = _loads(app.initiator_user_ids, []) or []
    if user_id not in ids:
        raise OaEngineError("您不在该审批的发起范围内")


def _instance_to_dict(inst: OaInstance, *, app_name: str = "") -> Dict[str, Any]:
    return {
        "id": inst.id,
        "appId": inst.app_id,
        "appVersionId": inst.app_version_id,
        "appName": app_name,
        "title": inst.title or "",
        "initiatorUserId": inst.initiator_user_id,
        "initiatorDeptId": inst.initiator_dept_id or "",
        "formDataJson": _loads(inst.form_data_json, {}),
        "status": inst.status,
        "sourceInstanceId": getattr(inst, "source_instance_id", None) or "",
        "currentRound": int(getattr(inst, "current_round", None) or 1),
        "createTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
        "updateTime": inst.update_time.strftime("%Y-%m-%d %H:%M:%S") if inst.update_time else "",
    }


def _task_to_dict(t: OaTask, *, assignee_user_name: str = "") -> Dict[str, Any]:
    return {
        "id": t.id,
        "instanceId": t.instance_id,
        "nodeKey": t.node_key or "",
        "nodeName": t.node_name or "",
        "taskType": t.task_type,
        "assigneeUserId": t.assignee_user_id,
        "assigneeUserName": assignee_user_name or t.assignee_user_id or "",
        "status": t.status,
        "commentText": t.comment_text or "",
        "resolveSource": t.resolve_source or "",
        "signMode": t.sign_mode or "",
        "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
        "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
    }


_INSTANCE_STATUS_NODE_LABEL = {
    "completed": "已完成",
    "rejected": "已拒绝",
    "returned": "已打回",
    "cancelled": "已撤销",
}


def _instance_progress_summary(
    inst: OaInstance,
    *,
    app_name: str,
    initiator_user_name: str,
    pending_approval_tasks: List[OaTask],
    name_map: Dict[str, str],
) -> Dict[str, Any]:
    status = (inst.status or "").strip()
    apply_time = (
        inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else ""
    )
    if status == "running":
        node_names: List[str] = []
        seen_nodes = set()
        approver_names: List[str] = []
        for t in pending_approval_tasks:
            nn = (t.node_name or t.node_key or "").strip()
            if nn and nn not in seen_nodes:
                seen_nodes.add(nn)
                node_names.append(nn)
            uid = (t.assignee_user_id or "").strip()
            if uid:
                approver_names.append(name_map.get(uid, uid))
        current_node = "、".join(node_names) if node_names else "-"
        current_approvers = "、".join(approver_names) if approver_names else "-"
    else:
        current_node = _INSTANCE_STATUS_NODE_LABEL.get(status, status or "-")
        current_approvers = "-"
    return {
        "appName": app_name or (inst.title or ""),
        "initiatorUserId": inst.initiator_user_id or "",
        "initiatorUserName": initiator_user_name
        or name_map.get(inst.initiator_user_id or "", inst.initiator_user_id or ""),
        "applyTime": apply_time,
        "instanceStatus": status,
        "currentNodeName": current_node,
        "currentApproverNames": current_approvers,
    }


def _enrich_list_rows(
    db: Session,
    *,
    tenant_no: str,
    rows: List[Dict[str, Any]],
    instances: List[OaInstance],
) -> List[Dict[str, Any]]:
    if not rows:
        return []
    inst_by_id = {i.id: i for i in instances}
    app_ids = list({i.app_id for i in instances if i.app_id})
    apps = (
        db.query(OaApp)
        .filter(
            OaApp.id.in_(app_ids),
            OaApp.tenant_no == tenant_no,
            OaApp.del_flag == 0,
        )
        .all()
        if app_ids
        else []
    )
    app_name_by_id = {a.id: (a.app_name or "") for a in apps}

    inst_ids = list(inst_by_id.keys())
    pending_tasks = (
        db.query(OaTask)
        .filter(
            OaTask.tenant_no == tenant_no,
            OaTask.del_flag == 0,
            OaTask.instance_id.in_(inst_ids),
            OaTask.task_type == "approval",
            OaTask.status == "pending",
        )
        .order_by(OaTask.create_time.asc())
        .all()
        if inst_ids
        else []
    )
    pending_by_inst: Dict[str, List[OaTask]] = {}
    for t in pending_tasks:
        pending_by_inst.setdefault(t.instance_id, []).append(t)

    user_ids: List[str] = []
    for i in instances:
        if i.initiator_user_id:
            user_ids.append(i.initiator_user_id)
    for t in pending_tasks:
        if t.assignee_user_id:
            user_ids.append(t.assignee_user_id)
    name_map = _user_name_map(db, user_ids)

    out: List[Dict[str, Any]] = []
    for row in rows:
        iid = row.get("instanceId") or row.get("id")
        inst = inst_by_id.get(iid)
        if not inst:
            out.append(row)
            continue
        summary = _instance_progress_summary(
            inst,
            app_name=app_name_by_id.get(inst.app_id, ""),
            initiator_user_name=name_map.get(inst.initiator_user_id or "", ""),
            pending_approval_tasks=pending_by_inst.get(inst.id, []),
            name_map=name_map,
        )
        merged = dict(row)
        merged.update(summary)
        out.append(merged)
    return out


def _user_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
    ids = [uid for uid in {str(u or "").strip() for u in user_ids} if uid]
    if not ids:
        return {}
    rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
    return {r.id: (r.nick_name or r.id) for r in rows}


def _dept_name(db: Session, dept_id: Optional[str], *, tenant_no: str) -> str:
    did = (dept_id or "").strip()
    if not did:
        return ""
    row = (
        db.query(SysDept.dept_name)
        .filter(SysDept.id == did, SysDept.tenant_no == tenant_no, SysDept.del_flag == 0)
        .first()
    )
    if not row:
        return ""
    return (row[0] or "").strip()


def list_templates(db: Session, *, tenant_no: str, user_id: str) -> List[Dict[str, Any]]:
    apps = (
        db.query(OaApp)
        .filter(OaApp.tenant_no == tenant_no, OaApp.del_flag == 0, OaApp.status == "1")
        .order_by(OaApp.order_num.asc(), OaApp.create_time.desc())
        .all()
    )
    out: List[Dict[str, Any]] = []
    for app in apps:
        pub = oa_crud._find_latest_published(db, app.id, tenant_no)
        if not pub:
            continue
        try:
            _check_initiator_scope(app, user_id)
        except OaEngineError:
            continue
        out.append(
            {
                "id": app.id,
                "appName": app.app_name,
                "groupId": app.group_id or "",
                "iconUrl": app.icon_url or "",
                "description": app.description or "",
                "versionId": pub.id,
                "versionNo": pub.version_no,
            }
        )
    return out


def get_template_detail(
    db: Session, *, tenant_no: str, app_id: str, user_id: str
) -> Dict[str, Any]:
    app = oa_crud.get_app(db, app_id, tenant_no)
    if not app or app.status != "1":
        raise OaEngineError("应用不存在或已停用")
    _check_initiator_scope(app, user_id)
    pub = oa_crud._find_latest_published(db, app_id, tenant_no)
    if not pub:
        raise OaEngineError("应用尚未发布")
    return {
        "id": app.id,
        "appName": app.app_name,
        "groupId": app.group_id or "",
        "iconUrl": app.icon_url or "",
        "description": app.description or "",
        "versionId": pub.id,
        "versionNo": pub.version_no,
        "formSchemaJson": _loads(pub.form_schema_json, {}),
        "flowTreeJson": _loads(pub.flow_tree_json, {}),
        "advancedSettingsJson": _advanced(pub),
    }


_ASSIGNEE_TYPE_LABEL = {
    "users": "指定成员",
    "oa_roles": "工作流角色",
    "dept_leader": "发起人部门负责人",
    "direct_supervisor": "发起人的主管",
    "initiator_select": "发起人自选",
}


def _role_name_map(db: Session, tenant_no: str, role_ids: List[str], codes: List[str]) -> List[str]:
    names: List[str] = []
    ids = [str(x).strip() for x in (role_ids or []) if str(x).strip()]
    code_list = [str(c).strip().upper() for c in (codes or []) if str(c).strip()]
    if ids:
        rows = (
            db.query(OaRole)
            .filter(OaRole.tenant_no == tenant_no, OaRole.id.in_(ids), OaRole.del_flag == 0)
            .all()
        )
        names.extend([(r.role_name or r.role_code or r.id) for r in rows])
    if code_list:
        rows = (
            db.query(OaRole)
            .filter(
                OaRole.tenant_no == tenant_no,
                OaRole.role_code.in_(code_list),
                OaRole.del_flag == 0,
            )
            .all()
        )
        for r in rows:
            label = r.role_name or r.role_code or r.id
            if label not in names:
                names.append(label)
    return names


def _assignee_meta(db: Session, tenant_no: str, cfg: Dict[str, Any]) -> str:
    assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
    base = _ASSIGNEE_TYPE_LABEL.get(assignee_type, assignee_type or "未配置审批人")
    if assignee_type == "oa_roles":
        names = _role_name_map(
            db,
            tenant_no,
            cfg.get("oaRoleIds") or cfg.get("oa_role_ids") or [],
            cfg.get("oaRoleCodes") or cfg.get("oa_role_codes") or [],
        )
        if names:
            return f"{base} · {'/'.join(names)}"
    return base


def _preview_people(db: Session, user_ids: List[str]) -> List[Dict[str, str]]:
    name_map = _user_name_map(db, user_ids)
    out: List[Dict[str, str]] = []
    for uid in user_ids:
        out.append({"id": uid, "nickName": name_map.get(uid) or uid})
    return out


_FLOW_PREVIEW_PENDING_MSG = "必填信息填写后，流程将自动显示"
_FLOW_PREVIEW_UNMATCHED_MSG = "流程条件无法匹配，请检查表单或联系管理员"
_FORMULA_FIELD_RE = re.compile(r"\{([^{}]+)\}")


def _form_value_present(form_data: Optional[Dict[str, Any]], field: str) -> bool:
    """条件字段是否已有有效值（0 / false 视为已填）。"""
    if not field:
        return True
    data = form_data or {}
    if field not in data:
        return False
    val = data.get(field)
    if val is None:
        return False
    if isinstance(val, str) and not val.strip():
        return False
    if isinstance(val, (list, tuple, set, dict)) and len(val) == 0:
        return False
    return True


def _collect_condition_fields(node: Optional[Dict[str, Any]]) -> List[str]:
    """收集流程树中非默认条件分支引用的表单字段（去重保序）。"""
    fields: List[str] = []
    seen = set()

    def add(name: Any) -> None:
        key = str(name or "").strip()
        if not key or key in seen:
            return
        seen.add(key)
        fields.append(key)

    def walk(n: Any) -> None:
        if not isinstance(n, dict):
            return
        if _node_type(n) == "condition":
            for g in n.get("conditionGroups") or []:
                if not isinstance(g, dict):
                    continue
                rules = g.get("rules")
                when = g.get("when")
                if g.get("isDefault") or (not rules and not when):
                    child = g.get("child")
                    if isinstance(child, dict):
                        walk(child)
                    continue
                if isinstance(rules, list) and rules:
                    for r in rules:
                        if isinstance(r, dict) and r.get("field") is not None:
                            add(r.get("field"))
                elif isinstance(rules, str) and rules.strip():
                    for m in _FORMULA_FIELD_RE.findall(rules):
                        add(m)
                elif when:
                    for m in _FORMULA_FIELD_RE.findall(str(when)):
                        add(m)
                child = g.get("child")
                if isinstance(child, dict):
                    walk(child)
            join_child = n.get("child")
            if isinstance(join_child, dict):
                walk(join_child)
            return
        child = n.get("child")
        if isinstance(child, dict):
            walk(child)
        for g in n.get("conditionGroups") or []:
            if isinstance(g, dict) and isinstance(g.get("child"), dict):
                walk(g.get("child"))

    walk(node)
    return fields


def _condition_fields_ready(
    root: Optional[Dict[str, Any]], form_data: Optional[Dict[str, Any]]
) -> bool:
    fields = _collect_condition_fields(root)
    if not fields:
        return True
    data = form_data or {}
    return all(_form_value_present(data, f) for f in fields)


def _walk_preview_nodes(
    db: Session,
    *,
    node: Optional[Dict[str, Any]],
    tenant_no: str,
    initiator_user_id: str,
    initiator_dept_id: Optional[str],
    advanced: Dict[str, Any],
    initiator_select_by_node: Optional[Dict[str, List[str]]],
    form_data: Optional[Dict[str, Any]],
    steps: List[Dict[str, Any]],
) -> bool:
    """沿最终路径写入 steps。条件无法匹配时返回 False。"""
    cur = node
    data = form_data or {}
    while cur and isinstance(cur, dict):
        ntype = _node_type(cur)
        if ntype == "start":
            people = _preview_people(db, [initiator_user_id] if initiator_user_id else [])
            steps.append(
                {
                    "id": cur.get("id") or "",
                    "type": "start",
                    "name": cur.get("name") or "发起人",
                    "signMode": "",
                    "assigneeType": "",
                    "meta": "",
                    "people": people,
                    "resolveError": "",
                }
            )
            cur = cur.get("child")
            continue

        if ntype == "end":
            steps.append(
                {
                    "id": cur.get("id") or "",
                    "type": "end",
                    "name": cur.get("name") or "结束",
                    "signMode": "",
                    "assigneeType": "",
                    "meta": "",
                    "people": [],
                    "resolveError": "",
                }
            )
            break

        if ntype == "condition":
            groups = cur.get("conditionGroups") or []
            g = _match_condition_group(groups, data)
            if not g:
                return False
            child = g.get("child")
            if isinstance(child, dict):
                ok = _walk_preview_nodes(
                    db,
                    node=child,
                    tenant_no=tenant_no,
                    initiator_user_id=initiator_user_id,
                    initiator_dept_id=initiator_dept_id,
                    advanced=advanced,
                    initiator_select_by_node=initiator_select_by_node,
                    form_data=data,
                    steps=steps,
                )
                if not ok:
                    return False
            cur = cur.get("child")
            continue

        if ntype in ("approval", "cc"):
            cfg = cur.get("approverConfig") or {}
            assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
            sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "").strip()
            if ntype == "cc":
                sign_mode = "cc"
            elif not sign_mode and assignee_type != "initiator_select":
                sign_mode = "or"
            meta = _assignee_meta(db, tenant_no, cfg)
            people: List[Dict[str, str]] = []
            resolve_error = ""
            node_select_ids = _ids_for_initiator_select_node(
                initiator_select_by_node, cur.get("id") or ""
            )
            if assignee_type == "initiator_select":
                people = _preview_people(db, [str(x) for x in node_select_ids if str(x).strip()])
                if not people:
                    meta = meta or "发起人自选"
            else:
                try:
                    assignees, _meta = resolve_assignees(
                        db,
                        tenant_no=tenant_no,
                        assignee_config=cfg,
                        initiator_user_id=initiator_user_id,
                        initiator_dept_id=initiator_dept_id,
                        advanced=advanced,
                        initiator_select_user_ids=node_select_ids,
                    )
                    people = _preview_people(db, assignees)
                except AssigneeResolveError as e:
                    resolve_error = str(e)
            steps.append(
                {
                    "id": cur.get("id") or "",
                    "type": ntype,
                    "name": cur.get("name") or ("抄送人" if ntype == "cc" else "审批人"),
                    "signMode": sign_mode,
                    "assigneeType": assignee_type,
                    "meta": meta,
                    "people": people,
                    "resolveError": resolve_error,
                }
            )
            cur = cur.get("child")
            continue

        cur = cur.get("child")
    return True


def preview_start_flow(
    db: Session,
    *,
    tenant_no: str,
    app_id: str,
    initiator_user_id: str,
    initiator_dept_id: Optional[str] = None,
    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
    form_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """发起前预览审批链：条件字段齐后只走命中路径并解析人员。"""
    app = oa_crud.get_app(db, app_id, tenant_no)
    if not app or app.status != "1":
        raise OaEngineError("应用不存在或已停用")
    _check_initiator_scope(app, initiator_user_id)
    pub = oa_crud._find_latest_published(db, app_id, tenant_no)
    if not pub:
        raise OaEngineError("应用尚未发布")

    initiator = (
        db.query(SysUser)
        .filter(
            SysUser.id == initiator_user_id,
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
        )
        .first()
    )
    dept_id = (initiator_dept_id or "").strip() or (
        (initiator.dept_id or "") if initiator else ""
    )

    flow = _loads(pub.flow_tree_json, {}) or {}
    root = flow.get("root") if isinstance(flow, dict) else None
    if not isinstance(root, dict):
        root = flow if isinstance(flow, dict) else {}

    data = dict(form_data or {})
    if not _condition_fields_ready(root, data):
        return {
            "resolved": False,
            "message": _FLOW_PREVIEW_PENDING_MSG,
            "steps": [],
        }

    steps: List[Dict[str, Any]] = []
    select_by_node = _normalize_initiator_select_by_node(initiator_select_by_node)
    ok = _walk_preview_nodes(
        db,
        node=root,
        tenant_no=tenant_no,
        initiator_user_id=initiator_user_id,
        initiator_dept_id=dept_id or None,
        advanced=_advanced(pub),
        initiator_select_by_node=select_by_node,
        form_data=data,
        steps=steps,
    )
    if not ok:
        return {
            "resolved": False,
            "message": _FLOW_PREVIEW_UNMATCHED_MSG,
            "steps": [],
        }
    return {"resolved": True, "message": "", "steps": steps}


def _create_tasks_for_node(
    db: Session,
    *,
    inst: OaInstance,
    node: Dict[str, Any],
    ver: OaAppVersion,
    user_id: Optional[str],
    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
    order_num_start: Optional[int] = None,
) -> List[OaTask]:
    ntype = _node_type(node)
    now = datetime.now()
    created: List[OaTask] = []
    if ntype == "cc":
        task_type = "cc"
    elif ntype == "approval":
        task_type = "approval"
    else:
        return created

    cfg = node.get("approverConfig") or {}
    assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
    node_select_ids = _ids_for_initiator_select_node(
        initiator_select_by_node, node.get("id") or ""
    )
    if assignee_type == "initiator_select":
        if ntype == "cc" and not node_select_ids:
            return created
        if ntype == "approval" and not node_select_ids:
            raise OaEngineError("发起人未选择审批人")

    advanced = _advanced(ver)
    try:
        assignees, meta = resolve_assignees(
            db,
            tenant_no=inst.tenant_no,
            assignee_config=cfg,
            initiator_user_id=inst.initiator_user_id,
            initiator_dept_id=inst.initiator_dept_id,
            advanced=advanced,
            initiator_select_user_ids=node_select_ids,
        )
    except AssigneeResolveError as e:
        raise OaEngineError(str(e)) from e

    sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "or").strip() or "or"
    resolve_source = meta.get("resolve_source")
    status = "done" if task_type == "cc" else "pending"
    seq = order_num_start if order_num_start is not None else _next_task_order_num(db, inst.id)

    for uid in assignees:
        t = OaTask(
            id=generate_uuid(),
            instance_id=inst.id,
            node_key=node.get("id") or "",
            node_name=node.get("name") or "",
            task_type=task_type,
            assignee_user_id=uid,
            status=status,
            comment_text=None,
            resolve_source=resolve_source,
            sign_mode=sign_mode if task_type == "approval" else None,
            round_no=int(getattr(inst, "current_round", None) or 1),
            tenant_no=inst.tenant_no,
            order_num=seq,
            remark=None,
            create_by=user_id,
            create_dept_id=inst.initiator_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        seq += 1
        db.add(t)
        created.append(t)
    return created


def _match_condition_group(groups: List[Any], form_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    default_g = None
    for g in groups:
        if not isinstance(g, dict):
            continue
        rules = g.get("rules")
        when = g.get("when")
        # 空规则 = 默认分支
        if (not rules and not when) or g.get("isDefault"):
            default_g = g
            continue
        expr = when
        if not expr and isinstance(rules, str):
            expr = rules
        if not expr and isinstance(rules, list) and rules:
            # 简化：每条 {field, op, value} 用 AND
            parts = []
            for r in rules:
                if not isinstance(r, dict):
                    continue
                field = r.get("field")
                op = r.get("op") or "=="
                val = r.get("value")
                if field is None:
                    continue
                if isinstance(val, str):
                    parts.append(f"{{{field}}} {op} '{val}'")
                else:
                    parts.append(f"{{{field}}} {op} {val}")
            expr = " and ".join(parts) if parts else None
        if not expr:
            default_g = g
            continue
        try:
            if eval_formula(str(expr), form_data):
                return g
        except FormulaError:
            continue
    return default_g


def advance_from(
    db: Session,
    *,
    inst: OaInstance,
    ver: OaAppVersion,
    from_node: Optional[Dict[str, Any]],
    user_id: Optional[str],
    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
) -> None:
    """从某节点的下一跳开始推进；from_node 为 None 表示从 start 进入。"""
    root = _flow_root(_loads(ver.flow_tree_json, {}))
    form_data = _loads(inst.form_data_json, {}) or {}

    def next_node(node: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        ntype = _node_type(node)
        if ntype == "condition":
            groups = node.get("conditionGroups") or []
            g = _match_condition_group(groups, form_data)
            if not g or not isinstance(g.get("child"), dict):
                raise OaEngineError("条件分支无法匹配出线")
            return g["child"]
        child = node.get("child")
        return child if isinstance(child, dict) else None

    # 起点
    if from_node is None:
        cur = next_node(root) if _node_type(root) == "start" else root
    else:
        cur = next_node(from_node)

    # 同一次推进里可能连续落抄送+审批：共用递增序号，避免 create_time 同秒时乱序
    next_order = _next_task_order_num(db, inst.id)
    safety = 0
    while cur and safety < 50:
        safety += 1
        ntype = _node_type(cur)
        if ntype == "end":
            merge = _merge_after_branch_end(root, cur.get("id") or "")
            if merge is not None:
                cur = merge
                continue
            inst.status = "completed"
            inst.update_time = datetime.now()
            db.add(inst)
            return
        if ntype == "condition":
            cur = next_node(cur)
            continue
        if ntype in ("approval", "cc"):
            tasks = _create_tasks_for_node(
                db,
                inst=inst,
                node=cur,
                ver=ver,
                user_id=user_id,
                initiator_select_by_node=initiator_select_by_node,
                order_num_start=next_order,
            )
            next_order += max(len(tasks), 1)
            if ntype == "cc":
                # 抄送不阻断，继续往下
                cur = next_node(cur)
                continue
            if not tasks:
                raise OaEngineError("审批节点未生成任务")
            _maybe_auto_pass_initiator(
                db,
                inst=inst,
                ver=ver,
                tasks=tasks,
                user_id=user_id,
            )
            return
        # 未知类型跳过子节点
        cur = next_node(cur)

    # 无结束也无待办：异常结束
    if inst.status == "running":
        raise OaEngineError("流程推进异常：未到达结束且无待办任务")


def start_instance(
    db: Session,
    *,
    tenant_no: str,
    app_id: str,
    initiator_user_id: str,
    initiator_dept_id: Optional[str] = None,
    form_data: Optional[Dict[str, Any]] = None,
    title: Optional[str] = None,
    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
    related_instance_ids: Optional[List[str]] = None,
    source_instance_id: Optional[str] = None,
) -> Dict[str, Any]:
    # source_instance_id 保留入参兼容，不再写入关联（复制开新单不挂旧单）
    _ = source_instance_id
    app = oa_crud.get_app(db, app_id, tenant_no)
    if not app or app.status != "1":
        raise OaEngineError("应用不存在或已停用")
    _check_initiator_scope(app, initiator_user_id)
    ver = oa_crud._find_latest_published(db, app_id, tenant_no)
    if not ver:
        raise OaEngineError("应用尚未发布，无法发起")

    schema = _loads(ver.form_schema_json, {})
    data = dict(form_data or {})
    try:
        data = apply_formulas_and_linkages(schema, data)
    except FormulaError as e:
        raise OaEngineError(f"公式计算失败: {e}") from e

    from app.services.oa_dept_select import validate_dept_form_values

    dept_errs = validate_dept_form_values(schema, data)
    if dept_errs:
        raise OaEngineError(dept_errs[0])

    user = (
        db.query(SysUser)
        .filter(SysUser.id == initiator_user_id, SysUser.del_flag == 0)
        .first()
    )
    dept_id = (initiator_dept_id or (user.dept_id if user else "") or "").strip() or None

    now = datetime.now()
    title_text = (title or app.app_name or "").strip()
    normalized_select = _normalize_initiator_select_by_node(initiator_select_by_node)
    inst = OaInstance(
        id=generate_uuid(),
        app_id=app_id,
        app_version_id=ver.id,
        title=title_text[:200],
        initiator_user_id=initiator_user_id,
        initiator_dept_id=dept_id,
        form_data_json=_dumps(data),
        initiator_select_json=_dumps(normalized_select),
        status="running",
        source_instance_id=None,
        current_round=1,
        tenant_no=tenant_no,
        order_num=0,
        remark=None,
        create_by=initiator_user_id,
        create_dept_id=dept_id or "",
        create_time=now,
        update_by=initiator_user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(inst)

    from app.services.oa_related_approval import (
        assert_linkable_instance,
        merge_relation_targets,
    )

    targets = merge_relation_targets(schema, data, related_instance_ids)
    for field_key, to_id in targets:
        target = (
            db.query(OaInstance)
            .filter(
                OaInstance.id == to_id,
                OaInstance.tenant_no == tenant_no,
                OaInstance.del_flag == 0,
            )
            .first()
        )
        assert_linkable_instance(target, user_id=initiator_user_id, tenant_no=tenant_no)
        db.add(
            OaRelation(
                id=generate_uuid(),
                from_instance_id=inst.id,
                to_instance_id=to_id,
                field_key=field_key,
                tenant_no=tenant_no,
                order_num=0,
                remark=None,
                create_by=initiator_user_id,
                create_dept_id=dept_id or "",
                create_time=now,
                update_by=initiator_user_id,
                update_time=now,
                del_flag=0,
            )
        )

    _add_instance_event(
        db,
        inst=inst,
        event_type="submit",
        actor_user_id=initiator_user_id,
        round_no=1,
    )
    advance_from(
        db,
        inst=inst,
        ver=ver,
        from_node=None,
        user_id=initiator_user_id,
        initiator_select_by_node=normalized_select,
    )
    db.commit()
    db.refresh(inst)
    return get_instance_detail(db, inst.id, tenant_no=tenant_no)


def _get_instance(db: Session, instance_id: str, tenant_no: str) -> OaInstance:
    inst = (
        db.query(OaInstance)
        .filter(OaInstance.id == instance_id, OaInstance.tenant_no == tenant_no, OaInstance.del_flag == 0)
        .first()
    )
    if not inst:
        raise OaEngineError("实例不存在")
    return inst


def get_instance_detail(db: Session, instance_id: str, *, tenant_no: str) -> Dict[str, Any]:
    inst = _get_instance(db, instance_id, tenant_no)
    app = oa_crud.get_app(db, inst.app_id, tenant_no)
    ver = (
        db.query(OaAppVersion)
        .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
        .first()
    )
    tasks = (
        db.query(OaTask)
        .filter(OaTask.instance_id == instance_id, OaTask.del_flag == 0)
        .all()
    )
    flow_root = None
    if ver:
        try:
            flow_root = _flow_root(_loads(ver.flow_tree_json, {}))
        except OaEngineError:
            flow_root = None
    tasks = _sort_tasks_for_detail(tasks, flow_root=flow_root)
    rels = (
        db.query(OaRelation)
        .filter(OaRelation.from_instance_id == instance_id, OaRelation.del_flag == 0)
        .all()
    )
    name_map = _user_name_map(
        db,
        [t.assignee_user_id for t in tasks if t.assignee_user_id]
        + ([inst.initiator_user_id] if inst.initiator_user_id else []),
    )
    data = _instance_to_dict(inst, app_name=app.app_name if app else "")
    data["initiatorUserName"] = name_map.get(inst.initiator_user_id or "", inst.initiator_user_id or "")
    data["initiatorDeptName"] = _dept_name(
        db, inst.initiator_dept_id, tenant_no=tenant_no
    )
    data["tasks"] = [
        _task_to_dict(t, assignee_user_name=name_map.get(t.assignee_user_id or "", ""))
        for t in tasks
    ]
    data["relatedInstanceIds"] = [r.to_instance_id for r in rels]
    data["formSchemaJson"] = _loads(ver.form_schema_json, {}) if ver else {}
    adv = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
    data["advancedSettingsJson"] = adv
    data["allowAddSign"] = bool(adv.get("allowAddSign", True))
    data["allowReturn"] = bool(adv.get("allowReturn", True))
    events = (
        db.query(OaInstanceEvent)
        .filter(OaInstanceEvent.instance_id == instance_id, OaInstanceEvent.del_flag == 0)
        .order_by(OaInstanceEvent.create_time.asc())
        .all()
    )
    event_actors = [e.actor_user_id for e in events if e.actor_user_id]
    if event_actors:
        name_map.update(_user_name_map(db, event_actors))
    data["progress"] = build_progress_items(
        inst=inst,
        flow_root=flow_root,
        tasks=tasks,
        name_map=name_map,
        events=events,
        db=db,
        advanced=adv,
    )
    # 不再暴露旧单/重提单关联跳转
    data["sourceInstanceId"] = ""
    data["resubmittedInstanceId"] = ""
    return data


def list_tasks(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    kind: str,
) -> List[Dict[str, Any]]:
    kind = (kind or "").strip()
    if kind == "pending":
        q = db.query(OaTask).filter(
            OaTask.tenant_no == tenant_no,
            OaTask.del_flag == 0,
            OaTask.assignee_user_id == user_id,
            OaTask.task_type == "approval",
            OaTask.status == "pending",
        )
    elif kind == "done":
        q = db.query(OaTask).filter(
            OaTask.tenant_no == tenant_no,
            OaTask.del_flag == 0,
            OaTask.assignee_user_id == user_id,
            OaTask.task_type == "approval",
            OaTask.status.in_(("approved", "rejected", "returned", "transferred")),
        )
    elif kind == "cc":
        q = db.query(OaTask).filter(
            OaTask.tenant_no == tenant_no,
            OaTask.del_flag == 0,
            OaTask.assignee_user_id == user_id,
            OaTask.task_type == "cc",
        )
    elif kind == "mine":
        insts = (
            db.query(OaInstance)
            .filter(
                OaInstance.tenant_no == tenant_no,
                OaInstance.del_flag == 0,
                OaInstance.initiator_user_id == user_id,
            )
            .order_by(OaInstance.create_time.desc())
            .all()
        )
        rows = [_instance_to_dict(i) for i in insts]
        return _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts)
    else:
        raise OaEngineError(f"不支持的列表类型: {kind}")

    rows_orm = q.order_by(OaTask.create_time.desc()).all()
    rows = [_task_to_dict(t) for t in rows_orm]
    inst_ids = list({t.instance_id for t in rows_orm if t.instance_id})
    insts = (
        db.query(OaInstance)
        .filter(
            OaInstance.tenant_no == tenant_no,
            OaInstance.del_flag == 0,
            OaInstance.id.in_(inst_ids),
        )
        .all()
        if inst_ids
        else []
    )
    return _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts)


def _approve_task_core(
    db: Session,
    *,
    task: OaTask,
    inst: OaInstance,
    ver: OaAppVersion,
    user_id: str,
    comment: Optional[str],
) -> bool:
    """将 pending 审批任务置为 approved 并按或签/会签决定是否推进。

    返回 True 表示已调用 advance_from 推进；False 表示会签未齐、未推进。
    调用方负责 commit（若需要）。
    """
    now = datetime.now()
    task.comment_text = (comment or "").strip() or None
    task.status = "approved"
    task.update_by = user_id
    task.update_time = now
    db.add(task)
    # Session 默认 autoflush=False：不先 flush 时，后续按 status=pending 的查询仍会命中本任务，
    # 或签「取消同节点其余待办」会把本人已同意覆盖成 done。
    db.flush()

    sign_mode = (task.sign_mode or "or").strip() or "or"
    pending_same = (
        db.query(OaTask)
        .filter(
            OaTask.instance_id == inst.id,
            OaTask.node_key == task.node_key,
            OaTask.task_type == "approval",
            OaTask.id != task.id,
            OaTask.status == "pending",
            OaTask.del_flag == 0,
        )
        .count()
    )
    if sign_mode == "and" and pending_same > 0:
        return False

    if sign_mode != "and":
        for s in (
            db.query(OaTask)
            .filter(
                OaTask.instance_id == inst.id,
                OaTask.node_key == task.node_key,
                OaTask.id != task.id,
                OaTask.status == "pending",
                OaTask.del_flag == 0,
            )
            .all()
        ):
            s.status = "done"
            s.update_by = user_id
            s.update_time = now
            db.add(s)

    root = _flow_root(_loads(ver.flow_tree_json, {}))
    node = _find_node(root, task.node_key or "")
    if not node:
        raise OaEngineError("找不到对应流程节点")
    advance_from(
        db,
        inst=inst,
        ver=ver,
        from_node=node,
        user_id=user_id,
        initiator_select_by_node=_load_initiator_select_by_node(inst),
    )
    return True


def _maybe_auto_pass_initiator(
    db: Session,
    *,
    inst: OaInstance,
    ver: OaAppVersion,
    tasks: List[OaTask],
    user_id: Optional[str],
) -> None:
    _ = user_id
    advanced = _advanced(ver)
    if not advanced.get("autoPassWhenApproverIsInitiator"):
        return
    if inst.status != "running":
        return
    initiator = (inst.initiator_user_id or "").strip()
    if not initiator:
        return
    self_task = None
    for t in tasks:
        if (
            t.task_type == "approval"
            and t.status == "pending"
            and (t.assignee_user_id or "") == initiator
        ):
            self_task = t
            break
    if not self_task:
        return
    _approve_task_core(
        db,
        task=self_task,
        inst=inst,
        ver=ver,
        user_id=initiator,
        comment=_AUTO_PASS_COMMENT,
    )


def complete_task(
    db: Session,
    task_id: str,
    *,
    tenant_no: str,
    user_id: str,
    action: str,
    comment: Optional[str] = None,
) -> Dict[str, Any]:
    action = (action or "").strip()
    if action not in ("approve", "reject"):
        raise OaEngineError("不支持的操作")

    task = (
        db.query(OaTask)
        .filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
        .first()
    )
    if not task:
        raise OaEngineError("任务不存在")
    if task.assignee_user_id != user_id:
        raise OaEngineError("无权处理该任务")
    if task.task_type != "approval" or task.status != "pending":
        raise OaEngineError("任务状态不可处理")

    inst = _get_instance(db, task.instance_id, tenant_no)
    if inst.status != "running":
        raise OaEngineError("实例已结束")

    ver = (
        db.query(OaAppVersion)
        .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
        .first()
    )
    if not ver:
        raise OaEngineError("流程版本不存在")

    advanced = _advanced(ver)
    if advanced.get("commentRequired") and not (comment or "").strip():
        raise OaEngineError("请填写审批意见")

    if action == "reject":
        now = datetime.now()
        task.comment_text = (comment or "").strip() or None
        task.status = "rejected"
        task.update_by = user_id
        task.update_time = now
        db.add(task)
        db.flush()
        siblings = (
            db.query(OaTask)
            .filter(
                OaTask.instance_id == inst.id,
                OaTask.node_key == task.node_key,
                OaTask.id != task.id,
                OaTask.status == "pending",
                OaTask.del_flag == 0,
            )
            .all()
        )
        for s in siblings:
            s.status = "done"
            s.update_by = user_id
            s.update_time = now
            db.add(s)
        inst.status = "rejected"
        inst.update_by = user_id
        inst.update_time = now
        db.add(inst)
        db.commit()
        return get_instance_detail(db, inst.id, tenant_no=tenant_no)

    _approve_task_core(
        db,
        task=task,
        inst=inst,
        ver=ver,
        user_id=user_id,
        comment=comment,
    )
    db.commit()
    return get_instance_detail(db, inst.id, tenant_no=tenant_no)


def add_sign(
    db: Session,
    task_id: str,
    *,
    tenant_no: str,
    user_id: str,
    user_ids: List[str],
    comment: Optional[str] = None,
) -> Dict[str, Any]:
    """当前待办处理人加签：为同节点新增待办审批人。"""
    task = (
        db.query(OaTask)
        .filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
        .first()
    )
    if not task:
        raise OaEngineError("任务不存在")
    if task.assignee_user_id != user_id:
        raise OaEngineError("无权加签该任务")
    if task.task_type != "approval" or task.status != "pending":
        raise OaEngineError("仅待处理的审批任务可加签")

    inst = _get_instance(db, task.instance_id, tenant_no)
    if inst.status != "running":
        raise OaEngineError("实例已结束")

    ver = (
        db.query(OaAppVersion)
        .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
        .first()
    )
    if not ver:
        raise OaEngineError("流程版本不存在")
    advanced = _advanced(ver)
    if not advanced.get("allowAddSign", True):
        raise OaEngineError("该应用不允许加签")

    ids = []
    seen = set()
    for uid in user_ids or []:
        uid = (uid or "").strip()
        if not uid or uid in seen:
            continue
        seen.add(uid)
        ids.append(uid)
    if not ids:
        raise OaEngineError("请选择加签人")

    existing = {
        r.assignee_user_id
        for r in db.query(OaTask)
        .filter(
            OaTask.instance_id == inst.id,
            OaTask.node_key == task.node_key,
            OaTask.del_flag == 0,
            OaTask.status == "pending",
        )
        .all()
    }
    now = datetime.now()
    created = 0
    next_order = _next_task_order_num(db, inst.id)
    for uid in ids:
        if uid in existing:
            continue
        # 校验用户有效
        u = (
            db.query(SysUser)
            .filter(
                SysUser.id == uid,
                SysUser.tenant_no == tenant_no,
                SysUser.del_flag == 0,
                SysUser.status == "1",
            )
            .first()
        )
        if not u:
            raise OaEngineError(f"加签用户无效：{uid}")
        db.add(
            OaTask(
                id=generate_uuid(),
                instance_id=inst.id,
                node_key=task.node_key,
                node_name=task.node_name,
                task_type="approval",
                assignee_user_id=uid,
                status="pending",
                comment_text=None,
                resolve_source="add_sign",
                sign_mode=task.sign_mode or "or",
                round_no=int(getattr(inst, "current_round", None) or getattr(task, "round_no", None) or 1),
                tenant_no=tenant_no,
                order_num=next_order,
                remark=(comment or "").strip() or None,
                create_by=user_id,
                create_dept_id=inst.initiator_dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
        )
        next_order += 1
        created += 1
    if created == 0:
        raise OaEngineError("所选用户已在当前节点待办中")
    # 加签后默认按会签语义收口更稳妥：保留原 sign_mode；若原或签则一人通过仍可结束
    if comment:
        task.comment_text = ((task.comment_text or "") + f"\n[加签说明] {comment}").strip()
        task.update_by = user_id
        task.update_time = now
        db.add(task)
    db.commit()
    return get_instance_detail(db, inst.id, tenant_no=tenant_no)


def transfer_task(
    db: Session,
    task_id: str,
    *,
    tenant_no: str,
    user_id: str,
    target_user_id: str,
    comment: Optional[str] = None,
) -> Dict[str, Any]:
    """当前待办处理人转交：结束本人待办并为同节点目标用户新建 pending 任务。"""
    task = (
        db.query(OaTask)
        .filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
        .first()
    )
    if not task:
        raise OaEngineError("任务不存在")
    if task.assignee_user_id != user_id:
        raise OaEngineError("无权转交该任务")
    if task.task_type != "approval" or task.status != "pending":
        raise OaEngineError("仅待处理的审批任务可转交")

    inst = _get_instance(db, task.instance_id, tenant_no)
    if inst.status != "running":
        raise OaEngineError("实例已结束")

    target = (target_user_id or "").strip()
    if not target:
        raise OaEngineError("请选择转交人")
    if target == user_id:
        raise OaEngineError("不能转交给自己")

    u = (
        db.query(SysUser)
        .filter(
            SysUser.id == target,
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .first()
    )
    if not u:
        raise OaEngineError(f"转交用户无效：{target}")

    existing = {
        r.assignee_user_id
        for r in db.query(OaTask)
        .filter(
            OaTask.instance_id == inst.id,
            OaTask.node_key == task.node_key,
            OaTask.del_flag == 0,
            OaTask.status == "pending",
        )
        .all()
    }
    if target in existing:
        raise OaEngineError("该用户已在当前节点待办中")

    now = datetime.now()
    comment_text = (comment or "").strip() or None
    round_no = int(getattr(inst, "current_round", None) or getattr(task, "round_no", None) or 1)

    task.status = "transferred"
    task.comment_text = comment_text
    task.update_by = user_id
    task.update_time = now
    db.add(task)

    next_order = _next_task_order_num(db, inst.id)
    db.add(
        OaTask(
            id=generate_uuid(),
            instance_id=inst.id,
            node_key=task.node_key,
            node_name=task.node_name,
            task_type="approval",
            assignee_user_id=target,
            status="pending",
            comment_text=None,
            resolve_source="transfer",
            sign_mode=task.sign_mode or "or",
            round_no=round_no,
            tenant_no=tenant_no,
            order_num=next_order,
            remark=comment_text,
            create_by=user_id,
            create_dept_id=inst.initiator_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
    )

    event_comment = comment_text or f"转交给 {target}"
    _add_instance_event(
        db,
        inst=inst,
        event_type="transfer",
        actor_user_id=user_id,
        round_no=round_no,
        comment=event_comment,
        task_id=task.id,
        node_key=task.node_key,
    )

    db.commit()
    return get_instance_detail(db, inst.id, tenant_no=tenant_no)


def return_task(
    db: Session,
    task_id: str,
    *,
    tenant_no: str,
    user_id: str,
    comment: Optional[str] = None,
) -> Dict[str, Any]:
    """审批人打回：本单终态 returned；再办须复制表单开新实例。"""
    task = (
        db.query(OaTask)
        .filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0)
        .first()
    )
    if not task:
        raise OaEngineError("任务不存在")
    if task.assignee_user_id != user_id:
        raise OaEngineError("无权打回该任务")
    if task.task_type != "approval" or task.status != "pending":
        raise OaEngineError("仅待处理的审批任务可打回")

    inst = _get_instance(db, task.instance_id, tenant_no)
    if inst.status != "running":
        raise OaEngineError("仅进行中的审批可打回")

    ver = (
        db.query(OaAppVersion)
        .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
        .first()
    )
    if not ver:
        raise OaEngineError("流程版本不存在")
    advanced = _advanced(ver)
    if not advanced.get("allowReturn", True):
        raise OaEngineError("该应用不允许打回")

    comment_text = (comment or "").strip()
    if not comment_text:
        raise OaEngineError("打回意见必填")

    now = datetime.now()
    round_no = int(getattr(inst, "current_round", None) or getattr(task, "round_no", None) or 1)
    task.status = "returned"
    task.comment_text = comment_text
    task.update_by = user_id
    task.update_time = now
    db.add(task)

    for t in (
        db.query(OaTask)
        .filter(
            OaTask.instance_id == inst.id,
            OaTask.status == "pending",
            OaTask.del_flag == 0,
            OaTask.id != task.id,
        )
        .all()
    ):
        t.status = "cancelled"
        t.update_by = user_id
        t.update_time = now
        db.add(t)

    inst.status = "returned"
    inst.update_by = user_id
    inst.update_time = now
    db.add(inst)
    _add_instance_event(
        db,
        inst=inst,
        event_type="return",
        actor_user_id=user_id,
        round_no=round_no,
        comment=comment_text,
        task_id=task.id,
        node_key=task.node_key,
    )
    db.commit()
    return get_instance_detail(db, inst.id, tenant_no=tenant_no)


def resubmit_instance(
    db: Session,
    instance_id: str,
    *,
    tenant_no: str,
    user_id: str,
    form_data: Optional[Dict[str, Any]] = None,
    initiator_select_by_node: Optional[Dict[str, List[str]]] = None,
) -> Dict[str, Any]:
    """已废弃：同实例再次提交。请复制表单重新发起新实例。"""
    _ = (db, instance_id, tenant_no, user_id, form_data, initiator_select_by_node)
    raise OaEngineError("同实例再次提交已废弃，请复制表单重新发起")


def withdraw_instance(
    db: Session, instance_id: str, *, tenant_no: str, user_id: str
) -> Dict[str, Any]:
    inst = _get_instance(db, instance_id, tenant_no)
    if inst.initiator_user_id != user_id:
        raise OaEngineError("仅发起人可撤销")
    if inst.status != "running":
        raise OaEngineError("仅运行中的实例可撤销")
    ver = (
        db.query(OaAppVersion)
        .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0)
        .first()
    )
    advanced = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
    if not advanced.get("allowWithdrawRunning", True):
        raise OaEngineError("该应用不允许撤销进行中的审批")

    now = datetime.now()
    inst.status = "cancelled"
    inst.update_by = user_id
    inst.update_time = now
    db.add(inst)
    for t in (
        db.query(OaTask)
        .filter(OaTask.instance_id == inst.id, OaTask.status == "pending", OaTask.del_flag == 0)
        .all()
    ):
        t.status = "cancelled"
        t.update_by = user_id
        t.update_time = now
        db.add(t)
    _add_instance_event(
        db,
        inst=inst,
        event_type="withdraw",
        actor_user_id=user_id,
        round_no=int(getattr(inst, "current_round", None) or 1),
    )
    db.commit()
    return get_instance_detail(db, inst.id, tenant_no=tenant_no)