zkxw/backend/app/services/wechat_mp_message.py

"""微信服务号模板消息：OA 审批/抄送通知。"""
from __future__ import annotations

import logging
from typing import Any, Dict

import httpx

from app.core.config import settings
from app.services.wechat_mp_auth import WechatMpAuthError, get_mp_access_token

logger = logging.getLogger(__name__)

_WX_TEMPLATE_SEND = "https://api.weixin.qq.com/cgi-bin/message/template/send"

# 微信 thing 类字段建议约 20 汉字；按字符保守截断
_THING_MAX_CHARS = 20


def _truncate_thing(value: str, max_chars: int = _THING_MAX_CHARS) -> str:
    s = (value or "").strip()
    if len(s) <= max_chars:
        return s
    return s[:max_chars]


def _build_oa_data(
    *,
    title: str,
    applicant: str,
    node_name: str,
    time_text: str,
    process_name: str = "",
    title_key: str = "",
    applicant_key: str = "",
    node_key: str = "",
    time_key: str = "",
    process_key: str = "",
) -> Dict[str, Dict[str, str]]:
    """按字段名组装 data；字段名留空则跳过（兼容临时少字段模板）。"""
    data: Dict[str, Dict[str, str]] = {}
    title_key = (title_key or "").strip()
    applicant_key = (applicant_key or "").strip()
    node_key = (node_key or "").strip()
    time_key = (time_key or "").strip()
    process_key = (process_key or "").strip()
    if title_key:
        data[title_key] = {"value": _truncate_thing(title)}
    if applicant_key:
        data[applicant_key] = {"value": _truncate_thing(applicant)}
    if node_key:
        data[node_key] = {"value": _truncate_thing(node_name)}
    if time_key:
        data[time_key] = {"value": (time_text or "").strip()}
    if process_key:
        data[process_key] = {"value": _truncate_thing(process_name)}
    return data


def _send_template(
    *,
    mp_openid: str,
    template_id: str,
    data: Dict[str, Any],
    pagepath: str,
) -> Dict[str, Any]:
    oid = (mp_openid or "").strip()
    if not oid:
        return {"success": False, "error": "missing_mp_openid"}

    tid = (template_id or "").strip()
    if not tid:
        return {"success": False, "error": "missing_template_id"}

    mini_appid = (settings.WXPAY_MINI_APPID or "").strip()
    if not mini_appid:
        return {"success": False, "error": "missing_mini_appid"}

    try:
        access_token = get_mp_access_token()
    except WechatMpAuthError as exc:
        logger.warning("mp template send aborted: %s", exc)
        return {"success": False, "error": str(exc)}

    path = (pagepath or "").lstrip("/")
    use_mini_jump = bool(mini_appid and path)
    payload: Dict[str, Any] = {
        "touser": oid,
        "template_id": tid,
        "data": data,
    }
    if use_mini_jump:
        payload["miniprogram"] = {"appid": mini_appid, "pagepath": path}

    def _post(body: Dict[str, Any]) -> Dict[str, Any]:
        with httpx.Client(timeout=15.0) as client:
            r = client.post(
                f"{_WX_TEMPLATE_SEND}?access_token={access_token}",
                json=body,
            )
            r.raise_for_status()
            return r.json()

    try:
        result = _post(payload)
    except Exception as exc:
        logger.exception("mp template send request failed")
        return {"success": False, "error": str(exc)}

    jumped = use_mini_jump
    # 服务号未关联小程序 / 页面未发布时微信返回 40165；降级为纯模板通知，保证先可达
    if result.get("errcode") == 40165 and use_mini_jump:
        logger.warning(
            "mp template miniprogram jump rejected (40165), retry without jump: %s",
            result.get("errmsg"),
        )
        jumped = False
        payload_no_jump = {
            "touser": oid,
            "template_id": tid,
            "data": data,
        }
        try:
            result = _post(payload_no_jump)
        except Exception as exc:
            logger.exception("mp template fallback send request failed")
            return {"success": False, "error": str(exc)}

    if result.get("errcode") in (0, None):
        logger.info(
            "mp template send ok: openid=%s msgid=%s jumped=%s",
            oid,
            result.get("msgid"),
            jumped,
        )
        return {"success": True, "msgid": result.get("msgid")}

    logger.error("mp template send failed: %s", result)
    return {
        "success": False,
        "error": result.get("errmsg") or "send_failed",
        "errcode": result.get("errcode"),
    }


def send_oa_approval_notify(
    *,
    mp_openid: str,
    pagepath: str,
    title: str,
    applicant: str,
    node_name: str,
    time_text: str,
    process_name: str = "",
) -> dict:
    return _send_template(
        mp_openid=mp_openid,
        template_id=settings.WX_MP_TEMPLATE_APPROVAL,
        data=_build_oa_data(
            title=title,
            applicant=applicant,
            node_name=node_name,
            time_text=time_text,
            process_name=process_name,
            title_key=settings.WX_MP_TPL_FIELD_TITLE,
            applicant_key=settings.WX_MP_TPL_FIELD_APPLICANT,
            node_key=settings.WX_MP_TPL_FIELD_NODE,
            time_key=settings.WX_MP_TPL_FIELD_TIME,
            process_key=settings.WX_MP_TPL_FIELD_PROCESS,
        ),
        pagepath=pagepath,
    )


def send_oa_cc_notify(
    *,
    mp_openid: str,
    pagepath: str,
    title: str,
    applicant: str,
    node_name: str,
    time_text: str,
    process_name: str = "",
) -> dict:
    return _send_template(
        mp_openid=mp_openid,
        template_id=settings.WX_MP_TEMPLATE_CC,
        data=_build_oa_data(
            title=title,
            applicant=applicant,
            node_name=node_name,
            time_text=time_text,
            process_name=process_name,
            title_key=settings.WX_MP_TPL_CC_FIELD_TITLE,
            applicant_key=settings.WX_MP_TPL_CC_FIELD_APPLICANT,
            node_key="",  # 抄送模板无当前环节
            time_key=settings.WX_MP_TPL_CC_FIELD_TIME,
            process_key=settings.WX_MP_TPL_CC_FIELD_PROCESS,
        ),
        pagepath=pagepath,
    )


def send_oa_applicant_progress_notify(
    *,
    mp_openid: str,
    pagepath: str,
    business_name: str,
    process_name: str,
    approver_name: str,
    result: str,
    time_text: str,
) -> dict:
    """通知申请人：某步审批结果（const 枚举：通过/失败/转办/其他）。"""
    data: Dict[str, Dict[str, str]] = {}
    biz_key = (settings.WX_MP_TPL_PROGRESS_FIELD_BUSINESS or "").strip()
    process_key = (settings.WX_MP_TPL_PROGRESS_FIELD_PROCESS or "").strip()
    approver_key = (settings.WX_MP_TPL_PROGRESS_FIELD_APPROVER or "").strip()
    result_key = (settings.WX_MP_TPL_PROGRESS_FIELD_RESULT or "").strip()
    time_key = (settings.WX_MP_TPL_PROGRESS_FIELD_TIME or "").strip()
    if biz_key:
        data[biz_key] = {"value": _truncate_thing(business_name)}
    if process_key:
        data[process_key] = {"value": _truncate_thing(process_name)}
    if approver_key:
        data[approver_key] = {"value": _truncate_thing(approver_name)}
    if result_key:
        data[result_key] = {"value": (result or "").strip()}
    if time_key:
        data[time_key] = {"value": (time_text or "").strip()}
    return _send_template(
        mp_openid=mp_openid,
        template_id=settings.WX_MP_TEMPLATE_APPLICANT_PROGRESS,
        data=data,
        pagepath=pagepath,
    )


def send_basic_experiment_delivery_remind(
    *,
    mp_openid: str,
    customer_name: str,
    product_name: str,
    quantity: str,
    contact_name: str,
    time_text: str,
    pagepath: str = "",
) -> dict:
    """基础实验：交付日前一天出库提醒。"""
    data: Dict[str, Dict[str, str]] = {}
    customer_key = (settings.WX_MP_TPL_BE_DELIVERY_FIELD_CUSTOMER or "").strip()
    product_key = (settings.WX_MP_TPL_BE_DELIVERY_FIELD_PRODUCT or "").strip()
    qty_key = (settings.WX_MP_TPL_BE_DELIVERY_FIELD_QTY or "").strip()
    contact_key = (settings.WX_MP_TPL_BE_DELIVERY_FIELD_CONTACT or "").strip()
    time_key = (settings.WX_MP_TPL_BE_DELIVERY_FIELD_TIME or "").strip()
    if customer_key:
        data[customer_key] = {"value": _truncate_thing(customer_name)}
    if product_key:
        data[product_key] = {"value": _truncate_thing(product_name)}
    if qty_key:
        data[qty_key] = {"value": (quantity or "").strip()}
    if contact_key:
        data[contact_key] = {"value": _truncate_thing(contact_name)}
    if time_key:
        data[time_key] = {"value": (time_text or "").strip()}
    return _send_template(
        mp_openid=mp_openid,
        template_id=settings.WX_MP_TEMPLATE_BASIC_EXPERIMENT_DELIVERY,
        data=data,
        pagepath=pagepath or "",
    )


def send_sample_collect_notify(
    *,
    mp_openid: str,
    title: str,
    applicant: str,
    time_text: str,
    task_type: str,
    sample_type: str = "",
    pagepath: str = "",
) -> dict:
    """协议新增样本：样本采集工单通知。"""
    data: Dict[str, Dict[str, str]] = {}
    title_key = (settings.WX_MP_TPL_SC_FIELD_TITLE or "").strip()
    applicant_key = (settings.WX_MP_TPL_SC_FIELD_APPLICANT or "").strip()
    time_key = (settings.WX_MP_TPL_SC_FIELD_TIME or "").strip()
    type_key = (settings.WX_MP_TPL_SC_FIELD_TYPE or "").strip()
    sample_type_key = (settings.WX_MP_TPL_SC_FIELD_SAMPLE_TYPE or "").strip()
    if title_key:
        data[title_key] = {"value": _truncate_thing(title)}
    if applicant_key:
        data[applicant_key] = {"value": _truncate_thing(applicant)}
    if time_key:
        data[time_key] = {"value": (time_text or "").strip()}
    if type_key:
        # phrase：纯汉字短词，不做 thing 截断以外的改写
        data[type_key] = {"value": (task_type or "").strip()}
    if sample_type_key:
        # thing 至少 1 个字符；没有样本类型时不传该字段，避免空值被微信拒绝
        sample_type_value = _truncate_thing(sample_type)
        if sample_type_value:
            data[sample_type_key] = {"value": sample_type_value}
    return _send_template(
        mp_openid=mp_openid,
        template_id=settings.WX_MP_TEMPLATE_SAMPLE_COLLECT,
        data=data,
        pagepath=pagepath or "",
    )