xiangqinxiaochengxu/backend/app/utils/wx_api.py

import httpx

from app.config import settings
from app.services.notify_service import get_access_token


async def code2session(code: str) -> dict:
    if not settings.wx_appid or not settings.wx_secret:
        return {
            "openid": f"dev_openid_{code or 'anonymous'}",
            "unionid": None,
        }

    if settings.wx_appid.startswith("your_") or settings.wx_secret.startswith("your_"):
        return {
            "openid": f"dev_openid_{code or 'anonymous'}",
            "unionid": None,
        }

    url = "https://api.weixin.qq.com/sns/jscode2session"
    params = {
        "appid": settings.wx_appid,
        "secret": settings.wx_secret,
        "js_code": code,
        "grant_type": "authorization_code",
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(url, params=params)
        return response.json()


async def generate_unlimited_qrcode(*, page: str, scene: str) -> bytes | None:
    if not settings.wx_appid or not settings.wx_secret:
        return None
    if settings.wx_appid.startswith("your_") or settings.wx_secret.startswith("your_"):
        return None

    access_token = await get_access_token()
    if not access_token:
        return None

    url = f"https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={access_token}"
    payload = {
        "page": page,
        "scene": scene,
        "check_path": False,
        "env_version": "develop",
    }
    async with httpx.AsyncClient(timeout=20.0) as client:
        response = await client.post(url, json=payload)
        content_type = response.headers.get("content-type", "")
        if "application/json" in content_type:
            data = response.json()
            if data.get("errcode"):
                return None
        return response.content