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 import or_
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,
    OaTask,
    SysDept,
    SysPost,
    SysUser,
)
from app.services.oa_assignee import (
    AssigneeResolveError,
    normalize_cc_assignee_types,
    resolve_assignees,
    resolve_cc_assignees,
)
from app.services.oa_formula import FormulaError, apply_formulas_and_linkages, eval_formula
from app.services.oa_confirmation_kit_sync import apply_confirmation_kit_user_sync
from app.services.oa_resignation_kit_sync import apply_resignation_kit_user_sync

import logging

logger = logging.getLogger(__name__)


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


_AUTO_PASS_COMMENT = "系统自动通过"
_DEDUP_AUTO_PASS_COMMENT = "审批人去重，系统自动通过"

INITIATOR_CONDITION_FIELD = "__initiator__"


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 not _subtree_contains(g.get("child"), end_id):
                    continue
                # 优先内层条件的汇合，再回退到本层
                nested = walk(g.get("child"))
                if nested is not None:
                    return nested
                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,
    name_map: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    name_map = name_map or {}
    uid = ""
    uname = ""
    comment = ""
    # 撤销终态：归因发起人，勿让任务节点承担「已撤销」语义
    if status == "cancelled":
        uid = (getattr(inst, "initiator_user_id", None) or "").strip()
        uname = name_map.get(uid, "") or uid
        comment = "发起人撤销"
    return {
        "kind": "terminal",
        "roundNo": getattr(inst, "current_round", None) or 1,
        "title": title,
        "status": status,
        "assigneeUserId": uid,
        "assigneeUserName": uname,
        "timestamp": _fmt_dt(getattr(inst, "update_time", None)),
        "commentText": comment,
        "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 ""
    )
    resolve_source = assignee_type
    if ntype == "cc":
        types = normalize_cc_assignee_types(cfg)
        assignees, meta = resolve_cc_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,
        )
        resolve_source = str(meta.get("resolve_source") or "")
        if not resolve_source:
            resolve_source = "multi" if len(types) > 1 else (types[0] if types else "")
        for uid in assignees:
            people.append(
                {
                    "id": uid,
                    "nickName": "",
                    "status": "not_reached",
                    "timestamp": "",
                    "commentText": "",
                    "taskId": "",
                }
            )
    elif 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, resolve_source


def _iter_resolved_path_nodes(
    root: Optional[Dict[str, Any]],
    form_data: Optional[Dict[str, Any]],
    *,
    initiator_user_id: Optional[str] = None,
    flow_root: Optional[Dict[str, Any]] = None,
) -> Optional[List[Dict[str, Any]]]:
    """沿条件命中路径收集节点；无法匹配时返回 None。"""
    if not isinstance(root, dict):
        return []
    data = form_data or {}
    tree_root = flow_root if isinstance(flow_root, dict) else root
    out: List[Dict[str, Any]] = []
    cur: Optional[Dict[str, Any]] = root
    while cur and isinstance(cur, dict):
        ntype = _node_type(cur)
        if ntype == "end":
            end_id = (cur.get("id") or "").strip()
            # 分支叶子 end 若后面还有汇合主链，不算真正结束
            if end_id and _merge_after_branch_end(tree_root, end_id) is not None:
                break
            out.append(cur)
            break
        if ntype == "condition":
            g = _match_condition_group(
                cur.get("conditionGroups") or [], data, initiator_user_id=initiator_user_id
            )
            if not g:
                return None
            child = g.get("child")
            if isinstance(child, dict):
                sub = _iter_resolved_path_nodes(
                    child,
                    data,
                    initiator_user_id=initiator_user_id,
                    flow_root=tree_root,
                )
                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, initiator_user_id=initiator_user_id
    )
    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"]
        # 整节点仅系统清待办(cancelled)时勿带时间，避免像审批人操作过
        if node_status == "cancelled":
            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, name_map))
    elif inst.status == "rejected":
        items.append(_terminal_item("rejected", "流程结束（已拒绝）", inst, name_map))
    elif inst.status == "cancelled":
        items.append(_terminal_item("cancelled", "已撤销", inst, name_map))
    # 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 "",
        "isRead": bool(int(getattr(t, "is_read", 0) or 0)),
        "readTime": (
            t.read_time.strftime("%Y-%m-%d %H:%M:%S")
            if getattr(t, "read_time", None)
            else ""
        ),
        "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],
    preview_limit: int = 3,
) -> List[Dict[str, Any]]:
    if not rows:
        return []
    from app.services.oa_related_approval import flatten_form_fields_for_summary

    limit = max(0, int(preview_limit or 3))
    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}

    ver_ids = list({i.app_version_id for i in instances if i.app_version_id})
    ver_by_id: Dict[str, OaAppVersion] = {}
    if ver_ids:
        for v in (
            db.query(OaAppVersion)
            .filter(OaAppVersion.id.in_(ver_ids), OaAppVersion.del_flag == 0)
            .all()
        ):
            ver_by_id[v.id] = v

    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,
        )
        ver = ver_by_id.get(inst.app_version_id or "")
        schema = _loads(getattr(ver, "form_schema_json", None), {}) if ver else {}
        form_data = _loads(inst.form_data_json, {}) or {}
        preview = []
        for f in flatten_form_fields_for_summary(schema, form_data)[:limit]:
            item = dict(f)
            item["label"] = item.get("label") or ""
            item["value"] = item.get("value") or ""
            preview.append(item)
        merged = dict(row)
        merged.update(summary)
        merged["previewFields"] = preview
        st = (inst.status or "").strip()
        if st in _FINISH_STATUSES and inst.update_time:
            merged["finishTime"] = inst.update_time.strftime("%Y-%m-%d %H:%M:%S")
        else:
            merged["finishTime"] = None
        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": "指定成员",
    "posts": "职位",
    "dept_leader": "发起人部门负责人",
    "direct_supervisor": "发起人的主管",
    "initiator_select": "发起人自选",
}


def _post_name_map(db: Session, tenant_no: str, post_ids: List[str]) -> List[str]:
    names: List[str] = []
    ids = [str(x).strip() for x in (post_ids or []) if str(x).strip()]
    if not ids:
        return names
    rows = (
        db.query(SysPost)
        .filter(SysPost.tenant_no == tenant_no, SysPost.id.in_(ids), SysPost.del_flag == 0)
        .all()
    )
    by_id = {r.id: (r.post_name or r.post_code or r.id) for r in rows}
    for pid in ids:
        label = by_id.get(pid)
        if label and 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 == "posts":
        names = _post_name_map(
            db,
            tenant_no,
            cfg.get("postIds") or cfg.get("post_ids") or [],
        )
        if names:
            return f"{base} · {'/'.join(names)}"
    return base


def _assignee_meta_for_cc(
    db: Session, tenant_no: str, cfg: Dict[str, Any], types: List[str]
) -> str:
    if not types:
        return "未配置抄送人"
    if len(types) == 1:
        return _assignee_meta(db, tenant_no, {**cfg, "assigneeType": types[0]})
    labels = [_ASSIGNEE_TYPE_LABEL.get(t, t) for t in types]
    return " + ".join(labels)


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]:
    """收集流程树中非默认条件分支引用的表单字段（去重保序）。

    系统伪字段 ``__initiator__`` 不依赖 formData，不纳入预览门闩。
    """
    fields: List[str] = []
    seen = set()

    def add(name: Any) -> None:
        key = str(name or "").strip()
        if not key or key in seen:
            return
        if key == INITIATOR_CONDITION_FIELD:
            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]],
    flow_root: Optional[Dict[str, Any]] = None,
) -> bool:
    """沿最终路径写入 steps。条件无法匹配时返回 False。"""
    cur = node
    data = form_data or {}
    tree_root = (
        flow_root if isinstance(flow_root, dict) else (node if isinstance(node, dict) else {})
    )
    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":
            end_id = (cur.get("id") or "").strip()
            # 分支叶子 end 后若有汇合主链，不写入「结束」（与引擎推进一致）
            if end_id and _merge_after_branch_end(tree_root, end_id) is not None:
                break
            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, initiator_user_id=initiator_user_id
            )
            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,
                    flow_root=tree_root,
                )
                if not ok:
                    return False
            cur = cur.get("child")
            continue

        if ntype in ("approval", "cc"):
            cfg = cur.get("approverConfig") or {}
            sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "").strip()
            people: List[Dict[str, str]] = []
            resolve_error = ""
            node_select_ids = _ids_for_initiator_select_node(
                initiator_select_by_node, cur.get("id") or ""
            )
            if ntype == "cc":
                types = normalize_cc_assignee_types(cfg)
                if "initiator_select" in types:
                    display_type = "initiator_select"
                elif len(types) == 1:
                    display_type = types[0]
                elif types:
                    display_type = "multi"
                else:
                    display_type = ""
                sign_mode = "cc"
                meta = _assignee_meta_for_cc(db, tenant_no, cfg, types)
                assignees, _meta = resolve_cc_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)
                steps.append(
                    {
                        "id": cur.get("id") or "",
                        "type": ntype,
                        "name": cur.get("name") or "抄送人",
                        "signMode": sign_mode,
                        "assigneeType": display_type,
                        "assigneeTypes": types,
                        "meta": meta,
                        "people": people,
                        "resolveError": resolve_error,
                    }
                )
            else:
                assignee_type = (
                    cfg.get("assigneeType") or cfg.get("assignee_type") or ""
                ).strip()
                if not sign_mode and assignee_type != "initiator_select":
                    sign_mode = "or"
                meta = _assignee_meta(db, tenant_no, cfg)
                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 "审批人",
                        "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,
        flow_root=root,
    )
    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 {}
    node_select_ids = _ids_for_initiator_select_node(
        initiator_select_by_node, node.get("id") or ""
    )
    advanced = _advanced(ver)

    if ntype == "cc":
        assignees, meta = resolve_cc_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,
        )
        if not assignees:
            return created
    else:
        assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
        if assignee_type == "initiator_select" and not node_select_ids:
            raise OaEngineError("发起人未选择审批人")
        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,
            is_read=0,
            read_time=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 _eval_condition_rules(
    rules: List[Any],
    form_data: Dict[str, Any],
    *,
    initiator_user_id: Optional[str] = None,
) -> bool:
    """AND 多条 {field, op, value}；contains 走人员/职位/部门集合交集，其余走公式。"""
    from app.services.oa_dept_select import dept_contains_match
    from app.services.oa_person_select import person_contains_match
    from app.services.oa_post_select import post_contains_match

    for r in rules:
        if not isinstance(r, dict):
            return False
        field = r.get("field")
        op = r.get("op") or "=="
        val = r.get("value")
        if field is None:
            return False
        if op == "contains":
            if field == INITIATOR_CONDITION_FIELD:
                uid = (initiator_user_id or "").strip()
                if isinstance(val, list):
                    allowed = {str(x or "").strip() for x in val if str(x or "").strip()}
                elif isinstance(val, str) and val.strip():
                    allowed = {val.strip()}
                else:
                    allowed = set()
                if not uid or uid not in allowed:
                    return False
                continue
            form_val = form_data.get(field)
            if not (
                person_contains_match(form_val, val)
                or post_contains_match(form_val, val)
                or dept_contains_match(form_val, val)
            ):
                return False
            continue
        if isinstance(val, str):
            expr = f"{{{field}}} {op} '{val}'"
        else:
            expr = f"{{{field}}} {op} {val}"
        try:
            if not eval_formula(expr, form_data):
                return False
        except FormulaError:
            return False
    return True


def _match_condition_group(
    groups: List[Any],
    form_data: Dict[str, Any],
    *,
    initiator_user_id: Optional[str] = None,
) -> 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
        if when:
            try:
                if eval_formula(str(when), form_data):
                    return g
            except FormulaError:
                continue
            continue
        if isinstance(rules, str):
            try:
                if eval_formula(str(rules), form_data):
                    return g
            except FormulaError:
                continue
            continue
        if isinstance(rules, list) and rules:
            try:
                if _eval_condition_rules(
                    rules, form_data, initiator_user_id=initiator_user_id
                ):
                    return g
            except FormulaError:
                continue
            continue
        default_g = g
    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,
) -> List[str]:
    """从某节点的下一跳开始推进；from_node 为 None 表示从 start 进入。

    返回本次推进新创建的任务 id 列表（含抄送与嵌套自动通过产生的后续任务）。
    """
    root = _flow_root(_loads(ver.flow_tree_json, {}))
    form_data = _loads(inst.form_data_json, {}) or {}
    created_ids: List[str] = []

    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, initiator_user_id=inst.initiator_user_id
            )
            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)
            try:
                apply_confirmation_kit_user_sync(db, inst)
            except Exception:
                logger.exception(
                    "confirmation kit user sync failed instance=%s", getattr(inst, "id", None)
                )
            try:
                apply_resignation_kit_user_sync(db, inst)
            except Exception:
                logger.exception(
                    "resignation kit user sync failed instance=%s", getattr(inst, "id", None)
                )
            return created_ids
        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,
            )
            created_ids.extend([t.id for t in tasks if getattr(t, "id", None)])
            next_order += max(len(tasks), 1)
            if ntype == "cc":
                # 抄送不阻断，继续往下
                cur = next_node(cur)
                continue
            if not tasks:
                raise OaEngineError("审批节点未生成任务")
            nested = _maybe_auto_pass_initiator(
                db,
                inst=inst,
                ver=ver,
                tasks=tasks,
                user_id=user_id,
            )
            created_ids.extend(nested or [])
            deduped = _maybe_auto_pass_approver_dedup(
                db,
                inst=inst,
                ver=ver,
                tasks=tasks,
            )
            created_ids.extend(deduped or [])
            return created_ids
        # 未知类型跳过子节点
        cur = next_node(cur)

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


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 {})
    from app.services.oa_datetime_normalize import normalize_datetime_form_values

    # 小程序历史默认到分、PC 默认到秒：入库前对「期望到秒」字段补 :00
    data = normalize_datetime_form_values(schema, data)
    try:
        data = apply_formulas_and_linkages(schema, data)
    except FormulaError as e:
        raise OaEngineError(f"公式计算失败: {e}") from e

    from app.services.oa_leave_hours import apply_leave_kit_hours, validate_leave_kit_form_values
    from app.services.oa_trip_days import apply_trip_kit_days, validate_trip_kit_form_values
    from app.services.oa_overtime_hours import (
        apply_overtime_kit_hours,
        validate_overtime_kit_form_values,
    )
    from app.services.cn_holiday import is_rest_day as cn_is_rest_day

    data = apply_leave_kit_hours(schema, data)
    leave_errs = validate_leave_kit_form_values(schema, data)
    if leave_errs:
        raise OaEngineError(leave_errs[0])

    data = apply_trip_kit_days(schema, data)
    trip_errs = validate_trip_kit_form_values(schema, data)
    if trip_errs:
        raise OaEngineError(trip_errs[0])

    def _ot_rest(ymd: str) -> bool:
        from datetime import datetime as _dt

        d = _dt.strptime(ymd, "%Y-%m-%d").date()
        return cn_is_rest_day(db, d)

    data = apply_overtime_kit_hours(schema, data, _ot_rest)
    ot_errs = validate_overtime_kit_form_values(schema, data)
    if ot_errs:
        raise OaEngineError(ot_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
    from app.services.oa_owner_fields import apply_owner_field_defaults

    data = apply_owner_field_defaults(
        schema,
        data,
        user_id=initiator_user_id,
        user_name=(user.nick_name if user else "") or "",
        dept_id=dept_id,
        dept_name=_dept_name(db, dept_id, tenant_no=tenant_no) if dept_id else None,
    )

    from app.services.oa_dept_select import validate_dept_form_values
    from app.services.oa_money_field import validate_money_form_values
    from app.services.oa_person_select import validate_person_form_values
    from app.services.oa_post_select import validate_post_form_values
    from app.services.oa_receive_account import validate_receive_account_form_values

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

    money_errs = validate_money_form_values(schema, data)
    if money_errs:
        raise OaEngineError(money_errs[0])

    person_errs = validate_person_form_values(schema, data)
    if person_errs:
        raise OaEngineError(person_errs[0])

    post_errs = validate_post_form_values(schema, data)
    if post_errs:
        raise OaEngineError(post_errs[0])

    recv_errs = validate_receive_account_form_values(schema, data)
    if recv_errs:
        raise OaEngineError(recv_errs[0])

    from app.services.oa_receive_account import validate_receive_account_ownership

    own_errs = validate_receive_account_ownership(
        db,
        schema,
        data,
        tenant_no=tenant_no,
        initiator_user_id=initiator_user_id,
    )
    if own_errs:
        raise OaEngineError(own_errs[0])

    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=db
        )
        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,
    )
    created = advance_from(
        db,
        inst=inst,
        ver=ver,
        from_node=None,
        user_id=initiator_user_id,
        initiator_select_by_node=normalized_select,
    )
    db.commit()
    from app.services.oa_wechat_notify import schedule_oa_task_notifies

    schedule_oa_task_notifies(created or [])
    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, viewer_user_id: 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 {}
    from app.services.oa_datetime_normalize import normalize_datetime_form_values

    # 详情回显：历史到分值补秒，避免 PC Element 日期控件解析失败显示空白
    data["formDataJson"] = normalize_datetime_form_values(
        data["formSchemaJson"], data.get("formDataJson") or {}
    )
    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"] = ""
    from app.services import oa_instance_comment as oa_comment

    data["comments"] = oa_comment.list_comments(
        db,
        instance_id=instance_id,
        tenant_no=tenant_no,
        viewer_user_id=viewer_user_id or "",
    )
    return data


def _normalize_cc_read_status(raw: Optional[str]) -> str:
    v = (raw or "").strip().lower() or "unread"
    if v not in ("unread", "read", "all"):
        raise OaEngineError("readStatus 仅支持 unread、read、all")
    return v


def _apply_cc_task_read(
    db: Session,
    *,
    task: OaTask,
    user_id: str,
    now: datetime,
) -> bool:
    """将抄送任务标为已读。返回是否发生了写入。幂等时返回 False。"""
    if int(getattr(task, "is_read", 0) or 0) == 1:
        return False
    task.is_read = 1
    task.read_time = now
    task.update_by = user_id
    task.update_time = now
    db.add(task)
    return True


def count_cc_unread(db: Session, *, tenant_no: str, user_id: str) -> int:
    return (
        db.query(OaTask)
        .filter(
            OaTask.tenant_no == tenant_no,
            OaTask.del_flag == 0,
            OaTask.assignee_user_id == user_id,
            OaTask.task_type == "cc",
            OaTask.is_read == 0,
        )
        .count()
    )


def count_pending_tasks(db: Session, *, tenant_no: str, user_id: str) -> int:
    return (
        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",
        )
        .count()
    )


def mark_cc_task_read(
    db: Session, *, tenant_no: str, user_id: str, task_id: str
) -> 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 or task.task_type != "cc" or task.assignee_user_id != user_id:
        raise OaEngineError("抄送任务不存在或无权操作")
    now = datetime.now()
    _apply_cc_task_read(db, task=task, user_id=user_id, now=now)
    db.commit()
    db.refresh(task)
    return _task_to_dict(task)


def mark_cc_tasks_read_batch(
    db: Session, *, tenant_no: str, user_id: str, ids: List[str]
) -> Dict[str, Any]:
    id_list = [x.strip() for x in (ids or []) if (x or "").strip()]
    if not id_list:
        return {"updated": 0}
    rows = (
        db.query(OaTask)
        .filter(
            OaTask.tenant_no == tenant_no,
            OaTask.del_flag == 0,
            OaTask.assignee_user_id == user_id,
            OaTask.task_type == "cc",
            OaTask.id.in_(id_list),
        )
        .all()
    )
    now = datetime.now()
    updated = 0
    for t in rows:
        if _apply_cc_task_read(db, task=t, user_id=user_id, now=now):
            updated += 1
    db.commit()
    return {"updated": updated}


_INSTANCE_STATUS_FILTER_VALUES = frozenset(
    {"running", "completed", "rejected", "returned", "cancelled"}
)
_FINISH_STATUSES = frozenset({"completed", "rejected", "cancelled"})


def _parse_day_start(s: Optional[str]) -> Optional[datetime]:
    raw = (s or "").strip()
    if not raw:
        return None
    try:
        d = datetime.strptime(raw[:10], "%Y-%m-%d")
    except ValueError:
        return None
    return d.replace(hour=0, minute=0, second=0, microsecond=0)


def _parse_day_end(s: Optional[str]) -> Optional[datetime]:
    raw = (s or "").strip()
    if not raw:
        return None
    try:
        d = datetime.strptime(raw[:10], "%Y-%m-%d")
    except ValueError:
        return None
    return d.replace(hour=23, minute=59, second=59, microsecond=999999)


def _apply_instance_list_filters(
    q,
    *,
    app_name: Optional[str] = None,
    initiator_name: Optional[str] = None,
    keyword: Optional[str] = None,
    apply_time_from: Optional[str] = None,
    apply_time_to: Optional[str] = None,
    instance_status: Optional[str] = None,
    joined_app: bool = False,
    joined_initiator: bool = False,
):
    """在已包含 OaInstance 的 query 上叠加列表筛选。返回 (query, joined_app, joined_initiator)。"""
    combined = (keyword or "").strip()
    if combined:
        if not joined_app:
            q = q.outerjoin(
                OaApp,
                (OaApp.id == OaInstance.app_id) & (OaApp.del_flag == 0),
            )
            joined_app = True
        if not joined_initiator:
            q = q.outerjoin(
                SysUser,
                (SysUser.id == OaInstance.initiator_user_id) & (SysUser.del_flag == 0),
            )
            joined_initiator = True
        like = f"%{combined}%"
        q = q.filter(
            or_(
                OaApp.app_name.like(like),
                OaInstance.title.like(like),
                SysUser.nick_name.like(like),
            )
        )
    else:
        kw = (app_name or "").strip()
        if kw:
            if not joined_app:
                q = q.outerjoin(
                    OaApp,
                    (OaApp.id == OaInstance.app_id) & (OaApp.del_flag == 0),
                )
                joined_app = True
            like = f"%{kw}%"
            q = q.filter(or_(OaApp.app_name.like(like), OaInstance.title.like(like)))

        iname = (initiator_name or "").strip()
        if iname:
            if not joined_initiator:
                q = q.outerjoin(
                    SysUser,
                    (SysUser.id == OaInstance.initiator_user_id) & (SysUser.del_flag == 0),
                )
                joined_initiator = True
            q = q.filter(SysUser.nick_name.like(f"%{iname}%"))

    t0 = _parse_day_start(apply_time_from)
    if t0 is not None:
        q = q.filter(OaInstance.create_time >= t0)
    t1 = _parse_day_end(apply_time_to)
    if t1 is not None:
        q = q.filter(OaInstance.create_time <= t1)

    st = (instance_status or "").strip()
    if st in _INSTANCE_STATUS_FILTER_VALUES:
        q = q.filter(OaInstance.status == st)

    return q, joined_app, joined_initiator


def _filter_by_initiator_dept(q, db: Session, *, tenant_no: str, initiator_dept_id: Optional[str]):
    """按发起部门（含下级）过滤；空参数不改 query。"""
    did = (initiator_dept_id or "").strip()
    if not did:
        return q
    from app.core.data_scope import _get_child_dept_ids

    dept_ids = _get_child_dept_ids(db, did, tenant_no)
    if not dept_ids:
        return q.filter(OaInstance.id == "")
    return q.filter(OaInstance.initiator_dept_id.in_(dept_ids))


def list_instance_history(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    can_view_all: bool,
    app_id: str,
    instance_id: Optional[str] = None,
    initiator_name: Optional[str] = None,
    apply_time_from: Optional[str] = None,
    apply_time_to: Optional[str] = None,
    finish_time_from: Optional[str] = None,
    finish_time_to: Optional[str] = None,
    instance_status: Optional[str] = None,
    initiator_dept_id: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
) -> Dict[str, Any]:
    app_id = (app_id or "").strip()
    if not app_id:
        raise OaEngineError("appId 不能为空")
    page = max(1, int(page or 1))
    page_size = min(100, max(1, int(page_size or 20)))
    offset = (page - 1) * page_size

    q = db.query(OaInstance).filter(
        OaInstance.tenant_no == tenant_no,
        OaInstance.del_flag == 0,
        OaInstance.app_id == app_id,
    )
    if not can_view_all:
        q = q.filter(OaInstance.initiator_user_id == user_id)
        initiator_name = None

    q = _filter_by_initiator_dept(
        q, db, tenant_no=tenant_no, initiator_dept_id=initiator_dept_id
    )

    iid = (instance_id or "").strip()
    if iid:
        q = q.filter(OaInstance.id.like(f"%{iid}%"))

    q, _, _ = _apply_instance_list_filters(
        q,
        initiator_name=initiator_name,
        apply_time_from=apply_time_from,
        apply_time_to=apply_time_to,
        instance_status=instance_status,
    )

    ft0 = _parse_day_start(finish_time_from)
    ft1 = _parse_day_end(finish_time_to)
    if ft0 is not None or ft1 is not None:
        q = q.filter(OaInstance.status.in_(tuple(_FINISH_STATUSES)))
        if ft0 is not None:
            q = q.filter(OaInstance.update_time >= ft0)
        if ft1 is not None:
            q = q.filter(OaInstance.update_time <= ft1)

    total = q.count()
    insts = (
        q.order_by(OaInstance.create_time.desc())
        .offset(offset)
        .limit(page_size)
        .all()
    )
    rows = [_instance_to_dict(i) for i in insts]
    return {
        "list": _enrich_list_rows(
            db,
            tenant_no=tenant_no,
            rows=rows,
            instances=insts,
            preview_limit=6,
        ),
        "total": int(total),
        "canViewAll": bool(can_view_all),
    }


_HISTORY_EXPORT_MAX_ROWS = 5000

_EXPORT_RESULT_TEXT = {
    "completed": "审批通过",
    "approved": "审批通过",
    "rejected": "审批拒绝",
    "returned": "已打回",
    "cancelled": "已撤销",
    "running": "",
    "pending": "",
}

_EXPORT_STATUS_TEXT = {
    "completed": "已结束",
    "approved": "已结束",
    "rejected": "已结束",
    "cancelled": "已结束",
    "returned": "已打回",
    "running": "进行中",
    "pending": "待处理",
}

_EXPORT_ACTION_TEXT = {
    "submitted": "提交申请",
    "approved": "同意",
    "rejected": "拒绝",
    "returned": "打回",
    "completed": "同意",
    "pending": "待处理",
    "cancelled": "取消",
    "transferred": "转交",
    "done": "抄送",
}

_FIXED_HISTORY_EXPORT_COLUMNS: List[Tuple[str, str]] = [
    ("instanceNo", "审批编号"),
    ("createTime", "创建时间"),
    ("initiatorName", "创建人"),
    ("currentAssignee", "当前负责人"),
    ("approvalResult", "审批结果"),
    ("approvalStatus", "审批状态"),
    ("updateTime", "更新时间"),
    ("finishTime", "完成时间"),
    ("initiatorDept", "创建人部门"),
    ("title", "审批单标题"),
    ("historyApprovers", "历史审批人"),
    ("duration", "耗时(时:分:秒)"),
    ("approvalRecord", "审批记录"),
]


def _format_export_duration(start: Optional[datetime], end: Optional[datetime]) -> str:
    if not start or not end:
        return ""
    delta = end - start
    if delta.total_seconds() < 0:
        return ""
    total = int(delta.total_seconds())
    h, rem = divmod(total, 3600)
    m, s = divmod(rem, 60)
    return f"{h}:{m}:{s}"


def _export_upload_cell_value(field: Dict[str, Any]) -> str:
    files = field.get("files") or []
    if not isinstance(files, list):
        return str(field.get("value") or "")
    urls: List[str] = []
    for f in files:
        if not isinstance(f, dict):
            continue
        u = str(f.get("url") or "").strip()
        if u:
            urls.append(u)
    if urls:
        return ",".join(urls)
    return str(field.get("value") or "")


def _export_progress_action(status: str, *, task_type: str = "") -> str:
    st = (status or "").strip()
    if (task_type or "").strip() == "cc" or st == "done":
        return "抄送"
    return _EXPORT_ACTION_TEXT.get(st, st or "")


def _format_export_progress_record(progress: List[Dict[str, Any]]) -> str:
    lines: List[str] = []
    for item in progress or []:
        kind = (item.get("kind") or "").strip()
        title = (item.get("title") or "").strip()
        ts = (item.get("timestamp") or "").strip()
        task_type = (item.get("taskType") or "").strip()
        if kind == "submit":
            name = (item.get("assigneeUserName") or "").strip()
            if not name:
                continue
            lines.append(f"{name}|提交申请|{ts}|提交申请")
            continue
        people = item.get("people") or []
        if isinstance(people, list) and people:
            for p in people:
                if not isinstance(p, dict):
                    continue
                name = (p.get("nickName") or p.get("assigneeUserName") or "").strip()
                if not name:
                    continue
                pst = (p.get("status") or "").strip()
                if pst in ("pending", "running", "cancelled") and task_type != "cc":
                    continue
                node = title or (item.get("nodeKey") or "").strip() or "-"
                pts = (p.get("timestamp") or ts or "").strip()
                action = _export_progress_action(pst, task_type=task_type)
                comment = (p.get("commentText") or "").strip()
                seg = f"{name}|{node}|{pts}|{action}"
                if comment:
                    seg += f"|{comment}"
                lines.append(seg)
            continue
        name = (item.get("assigneeUserName") or "").strip()
        if not name:
            continue
        st = (item.get("status") or "").strip()
        if st in ("pending", "running", "cancelled") and task_type != "cc":
            continue
        node = title or (item.get("nodeKey") or "").strip() or "-"
        action = _export_progress_action(st, task_type=task_type)
        comment = (item.get("commentText") or "").strip()
        seg = f"{name}|{node}|{ts}|{action}"
        if comment:
            seg += f"|{comment}"
        lines.append(seg)
    return ";\r\n".join(lines)


def _collect_history_approver_names(progress: List[Dict[str, Any]]) -> str:
    names: List[str] = []
    seen = set()
    for item in progress or []:
        if (item.get("kind") or "").strip() == "submit":
            continue
        people = item.get("people") or []
        if isinstance(people, list) and people:
            for p in people:
                if not isinstance(p, dict):
                    continue
                st = (p.get("status") or "").strip()
                if st not in ("approved", "completed", "rejected", "returned", "transferred"):
                    continue
                name = (p.get("nickName") or p.get("assigneeUserName") or "").strip()
                if name and name not in seen:
                    seen.add(name)
                    names.append(name)
            continue
        st = (item.get("status") or "").strip()
        if st not in ("approved", "completed", "rejected", "returned", "transferred"):
            continue
        name = (item.get("assigneeUserName") or "").strip()
        if name and name not in seen:
            seen.add(name)
            names.append(name)
    return ",".join(names)


def _build_history_export_business_columns(schema: Any) -> List[Dict[str, Any]]:
    from app.services.oa_related_approval import LAYOUT_SKIP, _schema_rules

    cols: List[Dict[str, Any]] = []
    for rule in _schema_rules(schema):
        typ = str(rule.get("type") or "")
        if not typ or typ in LAYOUT_SKIP:
            continue
        field = str(rule.get("field") or "").strip()
        if not field:
            continue
        label = str(rule.get("title") or rule.get("label") or field)
        if typ == "relatedApproval":
            cols.append(
                {
                    "key": field,
                    "label": label,
                    "kind": "related",
                    "subLabel": "关联标题",
                }
            )
        else:
            cols.append({"key": field, "label": label, "kind": "business"})
    return cols


def _merge_history_export_business_columns(col_lists: List[List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
    merged: List[Dict[str, Any]] = []
    seen = set()
    for cols in col_lists:
        for c in cols:
            key = str(c.get("key") or "")
            if not key or key in seen:
                continue
            seen.add(key)
            merged.append(dict(c))
    return merged


def _history_export_query(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    can_view_all: bool,
    app_id: str,
    instance_id: Optional[str] = None,
    instance_ids: Optional[List[str]] = None,
    initiator_name: Optional[str] = None,
    apply_time_from: Optional[str] = None,
    apply_time_to: Optional[str] = None,
    finish_time_from: Optional[str] = None,
    finish_time_to: Optional[str] = None,
    instance_status: Optional[str] = None,
    initiator_dept_id: Optional[str] = None,
):
    q = db.query(OaInstance).filter(
        OaInstance.tenant_no == tenant_no,
        OaInstance.del_flag == 0,
        OaInstance.app_id == app_id,
    )
    if not can_view_all:
        q = q.filter(OaInstance.initiator_user_id == user_id)
        initiator_name = None

    q = _filter_by_initiator_dept(
        q, db, tenant_no=tenant_no, initiator_dept_id=initiator_dept_id
    )

    clean_ids = [str(x).strip() for x in (instance_ids or []) if str(x).strip()]
    if clean_ids:
        q = q.filter(OaInstance.id.in_(clean_ids))
    else:
        iid = (instance_id or "").strip()
        if iid:
            q = q.filter(OaInstance.id.like(f"%{iid}%"))

    q, _, _ = _apply_instance_list_filters(
        q,
        initiator_name=initiator_name,
        apply_time_from=apply_time_from,
        apply_time_to=apply_time_to,
        instance_status=instance_status,
    )

    ft0 = _parse_day_start(finish_time_from)
    ft1 = _parse_day_end(finish_time_to)
    if ft0 is not None or ft1 is not None:
        q = q.filter(OaInstance.status.in_(tuple(_FINISH_STATUSES)))
        if ft0 is not None:
            q = q.filter(OaInstance.update_time >= ft0)
        if ft1 is not None:
            q = q.filter(OaInstance.update_time <= ft1)
    return q


def export_instance_history(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    can_view_all: bool,
    app_id: str,
    instance_id: Optional[str] = None,
    instance_ids: Optional[List[str]] = None,
    initiator_name: Optional[str] = None,
    apply_time_from: Optional[str] = None,
    apply_time_to: Optional[str] = None,
    finish_time_from: Optional[str] = None,
    finish_time_to: Optional[str] = None,
    instance_status: Optional[str] = None,
    initiator_dept_id: Optional[str] = None,
) -> Dict[str, Any]:
    from app.services.oa_related_approval import flatten_form_fields_for_summary

    app_id = (app_id or "").strip()
    if not app_id:
        raise OaEngineError("appId 不能为空")

    app = oa_crud.get_app(db, app_id, tenant_no)
    app_name = (app.app_name or "") if app else ""

    q = _history_export_query(
        db,
        tenant_no=tenant_no,
        user_id=user_id,
        can_view_all=can_view_all,
        app_id=app_id,
        instance_id=instance_id,
        instance_ids=instance_ids,
        initiator_name=initiator_name,
        apply_time_from=apply_time_from,
        apply_time_to=apply_time_to,
        finish_time_from=finish_time_from,
        finish_time_to=finish_time_to,
        instance_status=instance_status,
        initiator_dept_id=initiator_dept_id,
    )
    total = q.count()
    if total > _HISTORY_EXPORT_MAX_ROWS:
        raise OaEngineError(f"导出条数超过上限 {_HISTORY_EXPORT_MAX_ROWS}，请缩小筛选范围")
    insts = q.order_by(OaInstance.create_time.desc()).limit(_HISTORY_EXPORT_MAX_ROWS).all()
    if not insts:
        pub = oa_crud._find_latest_published(db, app_id, tenant_no)
        schema = _loads(getattr(pub, "form_schema_json", None), {}) if pub else {}
        business_columns = _build_history_export_business_columns(schema)
        return {
            "appName": app_name,
            "businessColumns": business_columns,
            "fixedColumns": [{"key": k, "label": lb} for k, lb in _FIXED_HISTORY_EXPORT_COLUMNS],
            "rows": [],
            "total": 0,
        }

    ver_by_id: Dict[str, OaAppVersion] = {}
    ver_ids = list({i.app_version_id for i in insts if i.app_version_id})
    if ver_ids:
        for v in (
            db.query(OaAppVersion)
            .filter(OaAppVersion.id.in_(ver_ids), OaAppVersion.del_flag == 0)
            .all()
        ):
            ver_by_id[v.id] = v

    pub = oa_crud._find_latest_published(db, app_id, tenant_no)
    pub_schema = _loads(getattr(pub, "form_schema_json", None), {}) if pub else {}
    col_lists = [_build_history_export_business_columns(pub_schema)]
    for inst in insts:
        ver = ver_by_id.get(inst.app_version_id or "")
        schema = _loads(getattr(ver, "form_schema_json", None), {}) if ver else {}
        col_lists.append(_build_history_export_business_columns(schema))
    business_columns = _merge_history_export_business_columns(col_lists)

    inst_ids = [i.id for i in insts]
    all_tasks = (
        db.query(OaTask)
        .filter(OaTask.instance_id.in_(inst_ids), OaTask.del_flag == 0)
        .all()
    )
    tasks_by_inst: Dict[str, List[OaTask]] = {}
    user_ids: List[str] = []
    for t in all_tasks:
        tasks_by_inst.setdefault(t.instance_id, []).append(t)
        if t.assignee_user_id:
            user_ids.append(t.assignee_user_id)

    all_events = (
        db.query(OaInstanceEvent)
        .filter(OaInstanceEvent.instance_id.in_(inst_ids), OaInstanceEvent.del_flag == 0)
        .order_by(OaInstanceEvent.create_time.asc())
        .all()
    )
    events_by_inst: Dict[str, List[OaInstanceEvent]] = {}
    for e in all_events:
        events_by_inst.setdefault(e.instance_id, []).append(e)
        if e.actor_user_id:
            user_ids.append(e.actor_user_id)

    for inst in insts:
        if inst.initiator_user_id:
            user_ids.append(inst.initiator_user_id)

    name_map = _user_name_map(db, user_ids)

    related_ids: List[str] = []
    flat_by_inst: Dict[str, List[Dict[str, Any]]] = {}
    for inst in insts:
        ver = ver_by_id.get(inst.app_version_id or "")
        schema = _loads(getattr(ver, "form_schema_json", None), {}) if ver else {}
        form_data = _loads(inst.form_data_json, {}) or {}
        flat = flatten_form_fields_for_summary(schema, form_data)
        flat_by_inst[inst.id] = flat
        for f in flat:
            for rid in f.get("relatedIds") or []:
                rid_s = str(rid or "").strip()
                if rid_s:
                    related_ids.append(rid_s)

    related_title_map: Dict[str, str] = {}
    if related_ids:
        rel_insts = (
            db.query(OaInstance)
            .filter(
                OaInstance.tenant_no == tenant_no,
                OaInstance.del_flag == 0,
                OaInstance.id.in_(list(set(related_ids))),
            )
            .all()
        )
        rel_app_ids = list({r.app_id for r in rel_insts if r.app_id})
        rel_app_map = {}
        if rel_app_ids:
            for a in db.query(OaApp).filter(OaApp.id.in_(rel_app_ids)).all():
                rel_app_map[a.id] = a.app_name or ""
        for r in rel_insts:
            initiator = name_map.get(r.initiator_user_id or "", "")
            related_title_map[r.id] = (
                f"{initiator}提交的{rel_app_map.get(r.app_id, '')}" if initiator else (r.title or r.id)
            )

    pending_by_inst: Dict[str, List[OaTask]] = {}
    for t in all_tasks:
        if (t.task_type or "") == "approval" and (t.status or "") == "pending":
            pending_by_inst.setdefault(t.instance_id, []).append(t)

    rows_out: List[Dict[str, Any]] = []
    for idx, inst in enumerate(insts, start=1):
        ver = ver_by_id.get(inst.app_version_id or "")
        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_by_inst.get(inst.id, []), flow_root=flow_root)
        events = events_by_inst.get(inst.id, [])
        adv = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS)
        progress = build_progress_items(
            inst=inst,
            flow_root=flow_root,
            tasks=tasks,
            name_map=name_map,
            events=events,
            db=db,
            advanced=adv,
        )
        summary = _instance_progress_summary(
            inst,
            app_name=app_name,
            initiator_user_name=name_map.get(inst.initiator_user_id or "", ""),
            pending_approval_tasks=pending_by_inst.get(inst.id, []),
            name_map=name_map,
        )
        st = (inst.status or "").strip()
        finish_time = (
            inst.update_time.strftime("%Y-%m-%d %H:%M:%S")
            if st in _FINISH_STATUSES and inst.update_time
            else ""
        )
        duration_end = inst.update_time if st in _FINISH_STATUSES else None
        business: Dict[str, Any] = {}
        flat_map = {str(f.get("fieldKey") or ""): f for f in flat_by_inst.get(inst.id, [])}
        for col in business_columns:
            key = str(col.get("key") or "")
            field = flat_map.get(key) or {}
            if col.get("kind") == "related":
                ids = field.get("relatedIds") or []
                business[key] = f"已关联 {len(ids)} 条" if ids else ""
                titles = [related_title_map.get(str(i), str(i)) for i in ids]
                business[f"{key}__titles"] = "、".join(t for t in titles if t)
            elif field.get("fieldType") in ("upload", "fcUpload", "image", "oaAttachment"):
                business[key] = _export_upload_cell_value(field)
            else:
                business[key] = str(field.get("value") or "")

        rows_out.append(
            {
                "index": idx,
                "id": inst.id,
                "business": business,
                "fixed": {
                    "instanceNo": inst.id,
                    "createTime": summary.get("applyTime") or "",
                    "initiatorName": summary.get("initiatorUserName") or "",
                    "currentAssignee": summary.get("currentApproverNames") or "-",
                    "approvalResult": _EXPORT_RESULT_TEXT.get(st, st),
                    "approvalStatus": _EXPORT_STATUS_TEXT.get(st, st),
                    "updateTime": inst.update_time.strftime("%Y-%m-%d %H:%M:%S")
                    if inst.update_time
                    else "",
                    "finishTime": finish_time,
                    "initiatorDept": _dept_name(
                        db, inst.initiator_dept_id, tenant_no=tenant_no
                    ),
                    "title": inst.title or "",
                    "historyApprovers": _collect_history_approver_names(progress),
                    "duration": _format_export_duration(inst.create_time, duration_end),
                    "approvalRecord": _format_export_progress_record(progress),
                },
            }
        )

    return {
        "appName": app_name,
        "businessColumns": business_columns,
        "fixedColumns": [{"key": k, "label": lb} for k, lb in _FIXED_HISTORY_EXPORT_COLUMNS],
        "rows": rows_out,
        "total": len(rows_out),
    }


def list_tasks(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    kind: str,
    read_status: Optional[str] = None,
    app_name: Optional[str] = None,
    initiator_name: Optional[str] = None,
    keyword: Optional[str] = None,
    apply_time_from: Optional[str] = None,
    apply_time_to: Optional[str] = None,
    instance_status: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
) -> Dict[str, Any]:
    kind = (kind or "").strip()
    page = max(1, int(page or 1))
    page_size = min(100, max(1, int(page_size or 20)))
    offset = (page - 1) * page_size

    if kind == "mine":
        q = db.query(OaInstance).filter(
            OaInstance.tenant_no == tenant_no,
            OaInstance.del_flag == 0,
            OaInstance.initiator_user_id == user_id,
        )
        q, _, _ = _apply_instance_list_filters(
            q,
            app_name=app_name,
            initiator_name=initiator_name,
            keyword=keyword,
            apply_time_from=apply_time_from,
            apply_time_to=apply_time_to,
            instance_status=instance_status,
        )
        total = q.count()
        insts = (
            q.order_by(OaInstance.create_time.desc())
            .offset(offset)
            .limit(page_size)
            .all()
        )
        rows = [_instance_to_dict(i) for i in insts]
        return {
            "list": _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts),
            "total": int(total),
        }

    if kind == "mention":
        from app.services.oa_instance_comment import list_mentioned_instance_ids

        ids = list_mentioned_instance_ids(db, tenant_no=tenant_no, user_id=user_id)
        if not ids:
            return {"list": [], "total": 0}
        q = db.query(OaInstance).filter(
            OaInstance.tenant_no == tenant_no,
            OaInstance.del_flag == 0,
            OaInstance.id.in_(ids),
        )
        q, _, _ = _apply_instance_list_filters(
            q,
            app_name=app_name,
            initiator_name=initiator_name,
            keyword=keyword,
            apply_time_from=apply_time_from,
            apply_time_to=apply_time_to,
            instance_status=instance_status,
        )
        insts = q.all()
        rank = {iid: i for i, iid in enumerate(ids)}
        insts.sort(key=lambda inst: rank.get(inst.id, len(ids)))
        total = len(insts)
        insts = insts[offset : offset + page_size]
        rows = [_instance_to_dict(i) for i in insts]
        return {
            "list": _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts),
            "total": int(total),
        }

    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",
        )
        rs = _normalize_cc_read_status(read_status)
        if rs == "unread":
            q = q.filter(OaTask.is_read == 0)
        elif rs == "read":
            q = q.filter(OaTask.is_read == 1)
        # all: 不加 is_read 条件
    else:
        raise OaEngineError(f"不支持的列表类型: {kind}")

    need_inst_join = any(
        [
            (keyword or "").strip(),
            (app_name or "").strip(),
            (initiator_name or "").strip(),
            (apply_time_from or "").strip(),
            (apply_time_to or "").strip(),
            (instance_status or "").strip() in _INSTANCE_STATUS_FILTER_VALUES,
        ]
    )
    if need_inst_join:
        q = q.join(
            OaInstance,
            (OaInstance.id == OaTask.instance_id)
            & (OaInstance.tenant_no == tenant_no)
            & (OaInstance.del_flag == 0),
        )
        q, _, _ = _apply_instance_list_filters(
            q,
            app_name=app_name,
            initiator_name=initiator_name,
            keyword=keyword,
            apply_time_from=apply_time_from,
            apply_time_to=apply_time_to,
            instance_status=instance_status,
        )

    total = q.count()
    rows_orm = (
        q.order_by(OaTask.create_time.desc()).offset(offset).limit(page_size).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 {
        "list": _enrich_list_rows(db, tenant_no=tenant_no, rows=rows, instances=insts),
        "total": int(total),
    }


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

    返回 advance_from 新创建的任务 id 列表；会签未齐、未推进时返回空列表。
    调用方负责 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 []

    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("找不到对应流程节点")
    return advance_from(
        db,
        inst=inst,
        ver=ver,
        from_node=node,
        user_id=user_id,
        initiator_select_by_node=_load_initiator_select_by_node(inst),
    )


def _maybe_auto_pass_initiator(
    db: Session,
    *,
    inst: OaInstance,
    ver: OaAppVersion,
    tasks: List[OaTask],
    user_id: Optional[str],
) -> List[str]:
    _ = 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 []
    return _approve_task_core(
        db,
        task=self_task,
        inst=inst,
        ver=ver,
        user_id=initiator,
        comment=_AUTO_PASS_COMMENT,
    )


def _prior_approved_assignees(db: Session, instance_id: str) -> set:
    rows = (
        db.query(OaTask.assignee_user_id)
        .filter(
            OaTask.instance_id == instance_id,
            OaTask.task_type == "approval",
            OaTask.status == "approved",
            OaTask.del_flag == 0,
        )
        .all()
    )
    return {(row[0] or "").strip() for row in rows if (row[0] or "").strip()}


def _maybe_auto_pass_approver_dedup(
    db: Session,
    *,
    inst: OaInstance,
    ver: OaAppVersion,
    tasks: List[OaTask],
) -> List[str]:
    """同一实例内已同意过的审批人，后续节点不再人工审批。

    或签：该人自动通过后整节点通过；会签：仅自动完成本人，其他人仍待批。
    仅统计 status=approved 的审批任务；拒绝等状态不触发。
    """
    advanced = _advanced(ver)
    if not advanced.get("approverDedup"):
        return []
    if inst.status != "running":
        return []
    prior = _prior_approved_assignees(db, inst.id)
    if not prior:
        return []
    created: List[str] = []
    for task in tasks:
        if inst.status != "running":
            break
        if task.task_type != "approval" or task.status != "pending":
            continue
        assignee = (task.assignee_user_id or "").strip()
        if assignee not in prior:
            continue
        created.extend(
            _approve_task_core(
                db,
                task=task,
                inst=inst,
                ver=ver,
                user_id=assignee,
                comment=_DEDUP_AUTO_PASS_COMMENT,
            )
            or []
        )
    return created


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()
        from app.services.oa_wechat_notify import schedule_oa_applicant_progress

        schedule_oa_applicant_progress(
            instance_id=inst.id,
            actor_user_id=user_id,
            action="reject",
            task_id=task.id,
        )
        return get_instance_detail(db, inst.id, tenant_no=tenant_no)

    created = _approve_task_core(
        db,
        task=task,
        inst=inst,
        ver=ver,
        user_id=user_id,
        comment=comment,
    )
    db.commit()
    from app.services.oa_wechat_notify import (
        schedule_oa_applicant_progress,
        schedule_oa_task_notifies,
    )

    schedule_oa_task_notifies(created or [])
    schedule_oa_applicant_progress(
        instance_id=inst.id,
        actor_user_id=user_id,
        action="approve",
        task_id=task.id,
    )
    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_ids: List[str] = []
    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}")
        new_id = generate_uuid()
        db.add(
            OaTask(
                id=new_id,
                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_ids.append(new_id)
    if not created_ids:
        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()
    from app.services.oa_wechat_notify import (
        schedule_oa_applicant_progress,
        schedule_oa_task_notifies,
    )

    schedule_oa_task_notifies(created_ids)
    schedule_oa_applicant_progress(
        instance_id=inst.id,
        actor_user_id=user_id,
        action="add_sign",
        task_id=task.id,
    )
    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)
    new_task_id = generate_uuid()
    db.add(
        OaTask(
            id=new_task_id,
            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()
    from app.services.oa_wechat_notify import (
        schedule_oa_applicant_progress,
        schedule_oa_task_notifies,
    )

    schedule_oa_task_notifies([new_task_id])
    schedule_oa_applicant_progress(
        instance_id=inst.id,
        actor_user_id=user_id,
        action="transfer",
        task_id=task.id,
    )
    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()
    from app.services.oa_wechat_notify import schedule_oa_applicant_progress

    schedule_oa_applicant_progress(
        instance_id=inst.id,
        actor_user_id=user_id,
        action="return",
        task_id=task.id,
    )
    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)