zkxw/backend/app/services/oa_wechat_notify.py

"""OA 任务创建后服务号通知（commit 后异步）。"""
from __future__ import annotations

import logging
import threading
from datetime import datetime
from typing import Iterable, List, Optional

from urllib.parse import urlencode

from app.core.config import settings
from app.core.database import SessionLocal
from app.models.models import OaApp, OaInstance, OaTask, SysUser
from app.services.wechat_mp_message import (
    send_oa_applicant_progress_notify,
    send_oa_approval_notify,
    send_oa_cc_notify,
)

logger = logging.getLogger(__name__)

# 微信模板 const4 枚举，必须原样下发
RESULT_PASS = "通过"
RESULT_FAIL = "失败"
RESULT_TRANSFER = "转办"
RESULT_OTHER = "其他"

# 模板消息关闭详情后回退的小程序页（与列表入口对齐）
RETURN_PENDING = "/pages/oa/pending/index"
RETURN_CC = "/pages/oa/cc/index"
RETURN_MINE = "/pages/oa/mine/index"

_ACTION_RESULT = {
    "approve": RESULT_PASS,
    "reject": RESULT_FAIL,
    "transfer": RESULT_TRANSFER,
    "return": RESULT_OTHER,
    "add_sign": RESULT_OTHER,
}


def map_oa_action_to_result(action: str) -> str:
    return _ACTION_RESULT.get((action or "").strip(), RESULT_OTHER)


def build_oa_detail_pagepath(
    *,
    instance_id: str,
    task_id: str = "",
    mode: str,
    return_path: str = "",
) -> str:
    """拼小程序 pagepath（无前导 /）；query 做 urlencode，保证 returnPath 含斜杠可用。"""
    params = {
        "instanceId": (instance_id or "").strip(),
        "mode": (mode or "view").strip() or "view",
    }
    tid = (task_id or "").strip()
    if tid and tid != "0":
        params["taskId"] = tid
    rp = (return_path or "").strip()
    if rp:
        params["returnPath"] = rp
    return "pages/oa/webview/index?" + urlencode(params)


def _build_matter_title(*, applicant: str, process_name: str, fallback: str) -> str:
    """申请事项/业务名称：申请人 - 流程名称。"""
    parts = [p for p in [(applicant or "").strip(), (process_name or "").strip()] if p]
    if parts:
        return " - ".join(parts)
    return (fallback or "").strip() or "审批"


def _resolve_process_and_applicant(db, inst: Optional[OaInstance]):
    applicant = ""
    process_name = ""
    if not inst:
        return applicant, process_name
    if inst.initiator_user_id:
        ini = db.query(SysUser).filter(SysUser.id == inst.initiator_user_id).first()
        applicant = (ini.nick_name if ini else "") or ""
    if inst.app_id:
        app = (
            db.query(OaApp)
            .filter(OaApp.id == inst.app_id, OaApp.del_flag == 0)
            .first()
        )
        process_name = (app.app_name if app else "") or ""
    if not process_name:
        process_name = (inst.title or "").strip()
    return applicant, process_name


def _send_one(db, task: OaTask) -> None:
    uid = (task.assignee_user_id or "").strip()
    if not uid:
        return
    user = db.query(SysUser).filter(SysUser.id == uid, SysUser.del_flag == 0).first()
    mp = (getattr(user, "wx_mp_openid", None) or "").strip() if user else ""
    if not mp:
        logger.info("oa wx notify skip: no mp_openid user=%s task=%s", uid, task.id)
        return
    inst = (
        db.query(OaInstance)
        .filter(OaInstance.id == task.instance_id, OaInstance.del_flag == 0)
        .first()
    )
    applicant, process_name = _resolve_process_and_applicant(db, inst)
    fallback = (inst.title if inst else "") or (task.node_name or "审批")
    title = _build_matter_title(
        applicant=applicant, process_name=process_name, fallback=fallback
    )
    time_text = datetime.now().strftime("%Y-%m-%d %H:%M")
    node_name = task.node_name or ""
    if task.task_type == "approval" and (task.status or "") == "pending":
        pagepath = build_oa_detail_pagepath(
            instance_id=task.instance_id,
            task_id=task.id,
            mode="approve",
            return_path=RETURN_PENDING,
        )
        send_oa_approval_notify(
            mp_openid=mp,
            pagepath=pagepath,
            title=title,
            applicant=applicant,
            node_name=node_name,
            time_text=time_text,
            process_name=process_name,
        )
    elif task.task_type == "cc":
        pagepath = build_oa_detail_pagepath(
            instance_id=task.instance_id,
            task_id=task.id,
            mode="view",
            return_path=RETURN_CC,
        )
        send_oa_cc_notify(
            mp_openid=mp,
            pagepath=pagepath,
            title=title,
            applicant=applicant,
            node_name=node_name,
            time_text=time_text,
            process_name=process_name,
        )


def _send_applicant_progress(
    db,
    *,
    instance_id: str,
    actor_user_id: str,
    result: str,
    task_id: str = "",
) -> None:
    iid = (instance_id or "").strip()
    if not iid:
        return
    inst = (
        db.query(OaInstance)
        .filter(OaInstance.id == iid, OaInstance.del_flag == 0)
        .first()
    )
    if not inst or not (inst.initiator_user_id or "").strip():
        return
    ini = (
        db.query(SysUser)
        .filter(SysUser.id == inst.initiator_user_id, SysUser.del_flag == 0)
        .first()
    )
    mp = (getattr(ini, "wx_mp_openid", None) or "").strip() if ini else ""
    if not mp:
        logger.info(
            "oa wx applicant progress skip: no mp_openid initiator=%s inst=%s",
            inst.initiator_user_id,
            iid,
        )
        return
    applicant = (ini.nick_name if ini else "") or ""
    process_name = ""
    if inst.app_id:
        app = (
            db.query(OaApp)
            .filter(OaApp.id == inst.app_id, OaApp.del_flag == 0)
            .first()
        )
        process_name = (app.app_name if app else "") or ""
    if not process_name:
        process_name = (inst.title or "").strip()
    business_name = _build_matter_title(
        applicant=applicant,
        process_name=process_name,
        fallback=(inst.title or "审批"),
    )
    actor = None
    aid = (actor_user_id or "").strip()
    if aid:
        actor = (
            db.query(SysUser)
            .filter(SysUser.id == aid, SysUser.del_flag == 0)
            .first()
        )
    approver_name = (actor.nick_name if actor else "") or ""
    pagepath = build_oa_detail_pagepath(
        instance_id=iid,
        task_id="",
        mode="view",
        return_path=RETURN_MINE,
    )
    send_oa_applicant_progress_notify(
        mp_openid=mp,
        pagepath=pagepath,
        business_name=business_name,
        process_name=process_name,
        approver_name=approver_name,
        result=(result or RESULT_OTHER).strip() or RESULT_OTHER,
        time_text=datetime.now().strftime("%Y-%m-%d %H:%M"),
    )


def _oa_wx_notify_enabled() -> bool:
    return bool(getattr(settings, "WX_MP_NOTIFY_ENABLED", False))


def notify_oa_tasks_sync(task_ids: Iterable[str]) -> None:
    if not _oa_wx_notify_enabled():
        logger.info("oa wx notify skipped: WX_MP_NOTIFY_ENABLED=false")
        return
    ids = [str(x).strip() for x in task_ids if str(x).strip()]
    if not ids:
        return
    db = None
    try:
        db = SessionLocal()
        tasks = (
            db.query(OaTask)
            .filter(OaTask.id.in_(ids), OaTask.del_flag == 0)
            .all()
        )
        for t in tasks:
            try:
                _send_one(db, t)
            except Exception:
                logger.exception("oa wx notify failed task=%s", t.id)
    except Exception:
        logger.exception("oa wx notify batch failed ids=%s", ids)
    finally:
        if db is not None:
            db.close()


def notify_oa_applicant_progress_sync(
    *,
    instance_id: str,
    actor_user_id: str,
    result: str,
    task_id: str = "",
) -> None:
    if not _oa_wx_notify_enabled():
        logger.info("oa wx applicant progress skipped: WX_MP_NOTIFY_ENABLED=false")
        return
    db = None
    try:
        db = SessionLocal()
        _send_applicant_progress(
            db,
            instance_id=instance_id,
            actor_user_id=actor_user_id,
            result=result,
            task_id=task_id,
        )
    except Exception:
        logger.exception(
            "oa wx applicant progress failed inst=%s actor=%s",
            instance_id,
            actor_user_id,
        )
    finally:
        if db is not None:
            db.close()


def schedule_oa_task_notifies(task_ids: List[str]) -> None:
    if not _oa_wx_notify_enabled():
        logger.info("oa wx notify schedule skipped: WX_MP_NOTIFY_ENABLED=false")
        return
    ids = [str(x).strip() for x in (task_ids or []) if str(x).strip()]
    if not ids:
        return
    threading.Thread(
        target=notify_oa_tasks_sync, args=(ids,), daemon=True, name="oa-wx-notify"
    ).start()


def schedule_oa_applicant_progress(
    *,
    instance_id: str,
    actor_user_id: str,
    action: str,
    task_id: str = "",
) -> None:
    if not _oa_wx_notify_enabled():
        logger.info(
            "oa wx applicant progress schedule skipped: WX_MP_NOTIFY_ENABLED=false"
        )
        return
    iid = (instance_id or "").strip()
    if not iid:
        return
    result = map_oa_action_to_result(action)
    threading.Thread(
        target=notify_oa_applicant_progress_sync,
        kwargs={
            "instance_id": iid,
            "actor_user_id": (actor_user_id or "").strip(),
            "result": result,
            "task_id": (task_id or "").strip(),
        },
        daemon=True,
        name="oa-wx-applicant-progress",
    ).start()