feat(api): add web wx-qr create and status endpoints.

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

backend/app/api/public.py

@@ -1,4 +1,6 @@
 """公共接口（Web 登录、刷新 Token、登出；用户模型对齐 docs/sql/shanxia2.0.sql 基线）"""
+import base64
+
 from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm import Session
@@ -34,6 +36,8 @@ from app.services.web_login_tokens import (
     issue_web_login_tokens,
 )
 from app.services.mp_webview_ticket import TICKET_TTL_SEC, consume_ticket, issue_ticket
+from app.services import wx_qr_session
+from app.services.wechat_mini_auth import WechatMiniAuthError, get_unlimited_wxacode
 from app.crud.crm.quote import view_by_share_code
 
 router = APIRouter(prefix="", tags=["公共接口"])
@@ -77,6 +81,49 @@ def web_login(
     return success_response(data)
 
 
+@router.post("/web/auth/wx-qr/create", response_model=dict)
+async def wx_qr_create():
+    session_id, short = wx_qr_session.create_session()
+    page = (settings.WX_QR_PAGE or "pages/login/web-qr/index").strip()
+    try:
+        raw = get_unlimited_wxacode(
+            scene=short,
+            page=page,
+            env_version=settings.WX_QR_ENV_VERSION,
+        )
+    except WechatMiniAuthError as exc:
+        wx_qr_session.invalidate(session_id)
+        return error_response(400, str(exc))
+    b64 = base64.b64encode(raw).decode("ascii")
+    mime = "image/jpeg" if raw[:2] == b"\xff\xd8" else "image/png"
+    return success_response(
+        {
+            "sessionId": session_id,
+            "expiresIn": settings.WX_QR_SESSION_TTL_SEC,
+            "qrImageBase64": f"data:{mime};base64,{b64}",
+        }
+    )
+
+
+@router.get("/web/auth/wx-qr/status", response_model=dict)
+async def wx_qr_status(sessionId: str = ""):
+    sid = (sessionId or "").strip()
+    data = wx_qr_session.get_session(sid)
+    if not data:
+        return success_response({"status": "expired", "message": "二维码已过期，请刷新"})
+    status = data.get("status") or "expired"
+    if status == "success":
+        tokens = wx_qr_session.consume_success(sid)
+        if not tokens:
+            return success_response({"status": "expired", "message": "登录凭证已失效，请刷新二维码"})
+        return success_response({"status": "success", **tokens})
+    if status == "failed":
+        return success_response({"status": "failed", "message": data.get("message") or "登录失败"})
+    if status in ("expired", "consumed"):
+        return success_response({"status": "expired", "message": "二维码已过期，请刷新"})
+    return success_response({"status": status})
+
+
 @router.post("/web/auth/refresh-token", response_model=dict)
 def refresh_token(body: dict, db: Session = Depends(get_db)):
     """刷新 Token"""

backend/tests/test_wx_qr_api.py

@@ -0,0 +1,33 @@
+"""web wx-qr create/status API。"""
+from unittest.mock import patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.services import wx_qr_session
+
+client = TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def force_memory_backend(monkeypatch):
+    monkeypatch.setattr(wx_qr_session, "_redis", lambda: None)
+    wx_qr_session._memory_clear_for_tests()
+
+
+def test_wx_qr_create_and_poll_pending():
+    with patch(
+        "app.api.public.get_unlimited_wxacode",
+        return_value=b"\x89PNG\r\n\x1a\n",
+    ):
+        r = client.post("/api/web/auth/wx-qr/create")
+        assert r.status_code == 200
+        body = r.json()
+        assert body.get("code") == 200
+        data = body["data"]
+        sid = data["sessionId"]
+        assert data["qrImageBase64"].startswith("data:image/")
+        st = client.get("/api/web/auth/wx-qr/status", params={"sessionId": sid})
+    assert st.status_code == 200
+    assert st.json()["data"]["status"] == "pending"