feat: add WeChat unlimited mini program code generator

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/wechat_mini_auth.py

@@ -127,3 +127,48 @@ def get_phone_number(phone_code: str) -> str:
     if not phone:
         raise WechatMiniAuthError("微信未返回手机号")
     return phone
+
+
+_WX_WXACODE_UNLIMITED = "https://api.weixin.qq.com/wxa/getwxacodeunlimit"
+
+
+def get_unlimited_wxacode(
+    *,
+    scene: str,
+    page: str,
+    env_version: Optional[str] = None,
+) -> bytes:
+    """获取不限制小程序码 PNG/JPEG 字节；失败抛 WechatMiniAuthError。"""
+    sc = (scene or "").strip()
+    if not sc or len(sc) > 32:
+        raise WechatMiniAuthError("scene 无效或超过 32 字符")
+    pg = (page or "").strip()
+    if not pg:
+        raise WechatMiniAuthError("page 不能为空")
+    ver = (env_version or getattr(settings, "WX_QR_ENV_VERSION", None) or "release").strip()
+    if ver not in ("release", "trial", "develop"):
+        ver = "release"
+    token = _get_access_token()
+    with httpx.Client(timeout=20.0) as client:
+        r = client.post(
+            f"{_WX_WXACODE_UNLIMITED}?access_token={token}",
+            json={
+                "scene": sc,
+                "page": pg,
+                "check_path": False,
+                "env_version": ver,
+                "width": 280,
+            },
+        )
+        r.raise_for_status()
+        ctype = (r.headers.get("Content-Type") or "").lower()
+        if "application/json" in ctype:
+            data = r.json()
+            logger.warning("getwxacodeunlimit failed: %s", data)
+            raise WechatMiniAuthError(
+                "生成小程序码失败",
+                wx_errcode=data.get("errcode"),
+            )
+        if not r.content:
+            raise WechatMiniAuthError("生成小程序码失败：空响应")
+        return r.content

backend/tests/test_wxacode.py

@@ -0,0 +1,47 @@
+from unittest.mock import MagicMock, patch
+
+from app.services.wechat_mini_auth import get_unlimited_wxacode, WechatMiniAuthError
+
+
+def test_get_unlimited_wxacode_returns_png_bytes():
+    token_resp = MagicMock()
+    token_resp.raise_for_status = MagicMock()
+    token_resp.json.return_value = {"access_token": "tok"}
+
+    code_resp = MagicMock()
+    code_resp.raise_for_status = MagicMock()
+    code_resp.headers = {"Content-Type": "image/jpeg"}
+    code_resp.content = b"\xff\xd8fake"
+
+    with patch("app.services.wechat_mini_auth._require_mini_credentials", return_value=("app", "sec")):
+        with patch("app.services.wechat_mini_auth.httpx.Client") as Client:
+            client = MagicMock()
+            Client.return_value.__enter__.return_value = client
+            client.get.return_value = token_resp
+            client.post.return_value = code_resp
+            raw = get_unlimited_wxacode(scene="abc", page="pages/login/web-qr/index")
+    assert raw.startswith(b"\xff\xd8")
+
+
+def test_get_unlimited_wxacode_json_error():
+    token_resp = MagicMock()
+    token_resp.raise_for_status = MagicMock()
+    token_resp.json.return_value = {"access_token": "tok"}
+
+    code_resp = MagicMock()
+    code_resp.raise_for_status = MagicMock()
+    code_resp.headers = {"Content-Type": "application/json"}
+    code_resp.content = b'{"errcode":40001,"errmsg":"fail"}'
+    code_resp.json.return_value = {"errcode": 40001, "errmsg": "fail"}
+
+    with patch("app.services.wechat_mini_auth._require_mini_credentials", return_value=("app", "sec")):
+        with patch("app.services.wechat_mini_auth.httpx.Client") as Client:
+            client = MagicMock()
+            Client.return_value.__enter__.return_value = client
+            client.get.return_value = token_resp
+            client.post.return_value = code_resp
+            try:
+                get_unlimited_wxacode(scene="abc", page="pages/login/web-qr/index")
+                assert False, "expected error"
+            except WechatMiniAuthError:
+                pass