shanxia2.0/backend/app/services/wechat_mini_auth.py

"""微信小程序登录：code2session、手机号快速验证。"""
from __future__ import annotations

import logging
import re
from typing import Any, Dict, Optional

import httpx

from app.core.config import settings

logger = logging.getLogger(__name__)

_WX_CODE2SESSION = "https://api.weixin.qq.com/sns/jscode2session"
_WX_TOKEN = "https://api.weixin.qq.com/cgi-bin/token"
_WX_PHONE = "https://api.weixin.qq.com/wxa/business/getuserphonenumber"


class WechatMiniAuthError(Exception):
    def __init__(self, message: str, *, wx_errcode: Optional[int] = None):
        super().__init__(message)
        self.wx_errcode = wx_errcode


def normalize_cn_mobile(raw: Optional[str]) -> str:
    if raw is None:
        return ""
    s = re.sub(r"\s+", "", str(raw).strip())
    if s.startswith("+86"):
        s = s[3:]
    elif s.startswith("86") and len(s) == 13:
        s = s[2:]
    return s


def phones_match(account_phone: Optional[str], authorized_phone: Optional[str]) -> bool:
    """账号手机号与本次微信授权手机号是否一致（规范化后比较）。"""
    a = normalize_cn_mobile(account_phone)
    b = normalize_cn_mobile(authorized_phone)
    return bool(a) and bool(b) and a == b


def _require_mini_credentials() -> tuple:
    appid = (settings.WECHAT_MP_APPID or "").strip()
    secret = (settings.WECHAT_MP_SECRET or "").strip()
    if not appid or not secret:
        raise WechatMiniAuthError("未配置小程序凭证（WECHAT_MP_APPID / WECHAT_MP_SECRET）")
    return appid, secret


def code2session(login_code: str) -> Dict[str, Any]:
    code = (login_code or "").strip()
    if not code:
        raise WechatMiniAuthError("loginCode 不能为空")
    appid, secret = _require_mini_credentials()
    with httpx.Client(timeout=15.0) as client:
        r = client.get(
            _WX_CODE2SESSION,
            params={
                "appid": appid,
                "secret": secret,
                "js_code": code,
                "grant_type": "authorization_code",
            },
        )
        r.raise_for_status()
        data = r.json()
    if data.get("errcode"):
        logger.warning("code2session failed: %s", data)
        raise WechatMiniAuthError(
            "微信登录凭证无效或已过期", wx_errcode=data.get("errcode")
        )
    openid = data.get("openid")
    if not openid:
        raise WechatMiniAuthError("微信未返回 openid")
    return {
        "openid": openid,
        "session_key": data.get("session_key"),
        "unionid": data.get("unionid"),
    }


def _get_access_token() -> str:
    appid, secret = _require_mini_credentials()
    with httpx.Client(timeout=15.0) as client:
        r = client.get(
            _WX_TOKEN,
            params={
                "grant_type": "client_credential",
                "appid": appid,
                "secret": secret,
            },
        )
        r.raise_for_status()
        data = r.json()
    token = data.get("access_token")
    if not token:
        logger.warning("get access_token failed: %s", data)
        raise WechatMiniAuthError(
            "获取微信 access_token 失败", wx_errcode=data.get("errcode")
        )
    return token


def get_phone_number(phone_code: str) -> str:
    code = (phone_code or "").strip()
    if not code:
        raise WechatMiniAuthError("phoneCode 不能为空")
    token = _get_access_token()
    with httpx.Client(timeout=15.0) as client:
        r = client.post(
            f"{_WX_PHONE}?access_token={token}",
            json={"code": code},
        )
        r.raise_for_status()
        data = r.json()
    if data.get("errcode") not in (0, None):
        logger.warning("getuserphonenumber failed: %s", data)
        raise WechatMiniAuthError(
            "获取手机号失败，请改用账号密码登录",
            wx_errcode=data.get("errcode"),
        )
    info = data.get("phone_info") or {}
    phone = normalize_cn_mobile(
        info.get("purePhoneNumber") or info.get("phoneNumber")
    )
    if not phone:
        raise WechatMiniAuthError("微信未返回手机号")
    return phone