"""微信小程序「发货信息录入」upload_shipping_info，用于交易类订单资金解冻前的平台发货同步。 文档：https://developers.weixin.qq.com/miniprogram/dev/platform-capabilities/business-capabilities/order-shipping/order-shipping.html """ from __future__ import annotations import logging from datetime import datetime, timedelta from typing import Any, Optional, Tuple import requests from sqlalchemy import and_ from sqlalchemy.orm import Session from app.core.config import settings from app.crud.order import order_crud logger = logging.getLogger(__name__) _ACCESS_TOKEN: Optional[str] = None _ACCESS_EXPIRES_AT: Optional[datetime] = None # 微信：同城配送（无快递单号场景） _LOGISTICS_TYPE_SAME_CITY = 2 _DELIVERY_MODE_UNIFIED = 1 # 已发货 / 内容未变化 —— 视为幂等成功 _IDEMPOTENT_ERRCODES = frozenset({10060002, 10060023}) def _now_shanghai_iso_millis() -> str: try: from zoneinfo import ZoneInfo now = datetime.now(ZoneInfo("Asia/Shanghai")) except Exception: now = datetime.utcnow() + timedelta(hours=8) # 2022-12-15T13:29:35.120+08:00 s = now.strftime("%Y-%m-%dT%H:%M:%S") ms = f"{now.microsecond // 1000:03d}" return f"{s}.{ms}+08:00" def _get_cgi_access_token() -> Optional[str]: global _ACCESS_TOKEN, _ACCESS_EXPIRES_AT now = datetime.now() if _ACCESS_TOKEN and _ACCESS_EXPIRES_AT and now < _ACCESS_EXPIRES_AT: return _ACCESS_TOKEN appid = (settings.WXPAY_MINI_APPID or "").strip() secret = (settings.WXPAY_SECRET or "").strip() if not appid or not secret: logger.warning("WXPAY_MINI_APPID / WXPAY_SECRET 未配置，无法获取 access_token") return None try: r = requests.get( "https://api.weixin.qq.com/cgi-bin/token", params={"grant_type": "client_credential", "appid": appid, "secret": secret}, timeout=10, ) data = r.json() except Exception as e: logger.exception("获取微信 access_token 请求异常: %s", e) return None token = data.get("access_token") if not token: logger.error("获取微信 access_token 失败: %s", data) return None expires_in = int(data.get("expires_in") or 7200) _ACCESS_TOKEN = token _ACCESS_EXPIRES_AT = now + timedelta(seconds=max(60, expires_in - 300)) return token def _build_item_desc(db: Session, order: Any) -> str: from app.models.models import BizOrderItem, BasProduct rows = ( db.query(BasProduct.name) .select_from(BizOrderItem) .outerjoin( BasProduct, and_(BasProduct.id == BizOrderItem.product_id, BasProduct.is_del == 0), ) .filter(BizOrderItem.order_id == order.id, BizOrderItem.is_del == 0) .order_by(BizOrderItem.create_at.asc()) .limit(12) .all() ) names = [str(n[0]).strip() for n in rows if n and n[0] and str(n[0]).strip()] if names: base = "、".join(names) if len(base) > 80: base = base[:77] + "..." text = f"{base}（单号{order.order_no}）" else: text = f"生鲜配送订单（单号{order.order_no}）" if len(text) > 120: text = text[:117] + "..." return text def _post_upload_shipping_info(body: dict) -> Tuple[bool, Optional[int], str]: token = _get_cgi_access_token() if not token: return False, None, "无法获取微信 access_token" url = f"https://api.weixin.qq.com/wxa/sec/order/upload_shipping_info?access_token={token}" try: r = requests.post(url, json=body, timeout=15) data = r.json() if r.content else {} except Exception as e: logger.exception("upload_shipping_info 请求异常: %s", e) return False, None, f"请求微信平台异常：{e}" errcode = data.get("errcode") if errcode == 0: return True, 0, "ok" try: ec = int(errcode) except (TypeError, ValueError): ec = -1 errmsg = (data.get("errmsg") or "").strip() or "unknown" return False, ec, errmsg def sync_trade_managed_shipping_for_delivered_order(db: Session, order: Any) -> Optional[str]: """ 配送完成时：若为线上微信支付且已付，向微信录入发货信息（同城配送、统一发货）。 返回 None 表示无需同步或已成功；返回字符串为应展示给前端的错误说明（应阻止本地「确认送达」提交）。 """ if not order: return None if order_crud.normalize_payment_method(getattr(order, "payment_method", None)) != "online": return None if (getattr(order, "pay_status", None) or "").strip() != "paid": return None appid = (settings.WXPAY_MINI_APPID or "").strip() secret = (settings.WXPAY_SECRET or "").strip() if not appid or not secret: if settings.DEBUG: logger.warning("DEBUG：未配置小程序 AppID/Secret，跳过微信发货信息录入（线上单 %s）", getattr(order, "order_no", "")) return None return "服务端未配置微信小程序 AppID/Secret，无法向微信平台录入发货信息" from app.models.models import BizWechatPay, SysUser wp = ( db.query(BizWechatPay) .filter(BizWechatPay.order_id == order.id, BizWechatPay.is_del == 0) .order_by(BizWechatPay.pay_time.desc(), BizWechatPay.create_at.desc()) .first() ) if not wp: return "线上支付订单缺少微信支付流水记录，无法录入发货信息" ts = (getattr(wp, "trade_state", None) or "").strip().upper() if ts and ts != "SUCCESS": return f"微信支付状态为「{getattr(wp, 'trade_state', '')}」，暂不可录入发货信息" transaction_id = (getattr(wp, "transaction_id", None) or "").strip() out_trade_no = (getattr(wp, "out_trade_no", None) or "").strip() mchid = (settings.WXPAY_MCHID or "").strip() client = db.get(SysUser, order.client_id) payer_openid = (getattr(client, "wx_mini_openid", None) or "").strip() if client else "" if not payer_openid: return "客户账号未绑定微信小程序，无法向微信平台录入发货信息（缺少 payer.openid）" if transaction_id: order_key: dict[str, Any] = {"order_number_type": 2, "transaction_id": transaction_id} elif out_trade_no and mchid: order_key = {"order_number_type": 1, "mchid": mchid, "out_trade_no": out_trade_no} else: return "缺少微信支付单号（transaction_id）或商户单号+商户号，无法录入发货信息" item_desc = _build_item_desc(db, order) body = { "order_key": order_key, "logistics_type": _LOGISTICS_TYPE_SAME_CITY, "delivery_mode": _DELIVERY_MODE_UNIFIED, "shipping_list": [{"item_desc": item_desc}], "upload_time": _now_shanghai_iso_millis(), "payer": {"openid": payer_openid}, } ok, errcode, errmsg = _post_upload_shipping_info(body) if ok: logger.info( "upload_shipping_info 成功 order_no=%s transaction_id=%s", getattr(order, "order_no", ""), transaction_id or out_trade_no, ) return None if errcode in _IDEMPOTENT_ERRCODES: logger.info( "upload_shipping_info 幂等忽略 errcode=%s order_no=%s msg=%s", errcode, getattr(order, "order_no", ""), errmsg, ) return None logger.error( "upload_shipping_info 失败 order_no=%s errcode=%s errmsg=%s", getattr(order, "order_no", ""), errcode, errmsg, ) return f"微信平台发货信息录入失败（{errcode}）：{errmsg}"