feat: add mini auth web-qr scanned and confirm endpoints

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

backend/app/api/mini_auth.py

@@ -10,6 +10,7 @@ from app.core.auth import get_current_user_by_token
 from app.core.database import get_db
 from app.core.response import error_response, success_response
 from app.crud.user import user_crud
+from app.services import wx_qr_session
 from app.services.issue_user_tokens import issue_tokens_for_user
 from app.services.wechat_login_service import (
     resolve_wechat_login_by_openid_only,
@@ -167,3 +168,65 @@ async def mp_bind(request: Request, body: dict, db: Session = Depends(get_db)):
     db.add(user)
     db.commit()
     return success_response({"bound": True})
+
+
+@router.post("/web-qr-scanned", response_model=dict)
+async def web_qr_scanned(body: dict):
+    raw = (body.get("sessionId") or body.get("scene") or "").strip()
+    sid = wx_qr_session.resolve_session_id(raw) or raw
+    data = wx_qr_session.get_session(sid)
+    if not data or data.get("status") in ("expired", "consumed"):
+        return error_response(400, "二维码已过期，请在电脑上刷新后重扫")
+    wx_qr_session.mark_scanned(sid)
+    return success_response({"ok": True})
+
+
+@router.post("/web-qr-confirm", response_model=dict)
+async def web_qr_confirm(body: dict, db: Session = Depends(get_db)):
+    raw = (body.get("sessionId") or body.get("scene") or "").strip()
+    sid = wx_qr_session.resolve_session_id(raw) or raw
+    data = wx_qr_session.get_session(sid)
+    if not data or data.get("status") in ("expired", "consumed", "failed"):
+        return error_response(400, "二维码已过期，请在电脑上刷新后重扫")
+    if data.get("status") == "success":
+        return success_response({"ok": True, "message": "电脑端即将登录"})
+
+    wx_qr_session.mark_scanned(sid)
+
+    login_code = (body.get("loginCode") or "").strip()
+    phone_code = (body.get("phoneCode") or "").strip()
+    if not login_code or not phone_code:
+        wx_qr_session.mark_failed(sid, "需要授权手机号才能完成扫码登录")
+        return error_response(400, "需要授权手机号才能完成扫码登录")
+
+    try:
+        sess = code2session(login_code)
+        phone = get_phone_number(phone_code)
+    except WechatMiniAuthError as exc:
+        wx_qr_session.mark_failed(sid, str(exc))
+        return error_response(400, str(exc))
+
+    user, err_msg, err_code = resolve_wechat_login_user(
+        db, openid=sess["openid"], phone=phone
+    )
+    if err_msg:
+        wx_qr_session.mark_failed(sid, err_msg)
+        return error_response(err_code or 400, err_msg)
+
+    unionid = (sess.get("unionid") or "").strip()
+    if unionid and not (getattr(user, "wx_unionid", None) or "").strip():
+        user.wx_unionid = unionid
+        user.update_time = datetime.now()
+        db.add(user)
+        db.commit()
+        db.refresh(user)
+
+    try:
+        tokens = await issue_tokens_for_user(db, user)
+    except ValueError as exc:
+        wx_qr_session.mark_failed(sid, str(exc))
+        return error_response(403, str(exc))
+
+    if not wx_qr_session.mark_success(sid, tokens):
+        return error_response(400, "二维码已过期，请在电脑上刷新后重扫")
+    return success_response({"ok": True, "message": "电脑端即将登录，可关闭本页"})

backend/tests/test_wx_qr_api.py

@@ -1,5 +1,6 @@
 """web wx-qr create/status API。"""
-from unittest.mock import patch
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
 
 from fastapi.testclient import TestClient
 
@@ -28,3 +29,33 @@ def test_wx_qr_create_and_poll_pending():
         st = client.get("/api/web/auth/wx-qr/status", params={"sessionId": sid})
     assert st.status_code == 200
     assert st.json()["data"]["status"] == "pending"
+
+
+def test_web_qr_confirm_success_sets_session():
+    fake_user = SimpleNamespace(
+        id="u1", phone_no="13800138000", wx_openid=None, wx_unionid=None, status="1"
+    )
+    tokens = {
+        "token": "access",
+        "refreshToken": "refresh",
+        "expiresIn": 3600,
+        "userInfo": {"id": "u1"},
+    }
+    with patch("app.services.wx_qr_session._redis", return_value=None):
+        wx_qr_session._memory_clear_for_tests()
+        sid, _short = wx_qr_session.create_session(ttl_sec=60)
+        with patch("app.api.mini_auth.code2session", return_value={"openid": "oid1", "unionid": ""}), \
+             patch("app.api.mini_auth.get_phone_number", return_value="13800138000"), \
+             patch("app.api.mini_auth.resolve_wechat_login_user", return_value=(fake_user, None, None)), \
+             patch("app.api.mini_auth.issue_tokens_for_user", new=AsyncMock(return_value=tokens)):
+            r = client.post(
+                "/api/mini/auth/web-qr-confirm",
+                json={"sessionId": sid, "loginCode": "lc", "phoneCode": "pc"},
+            )
+        assert r.status_code == 200
+        body = r.json()
+        assert body.get("code") == 200
+        assert body["data"].get("ok") is True
+        assert "token" not in (body.get("data") or {})
+        consumed = wx_qr_session.consume_success(sid)
+        assert consumed and consumed["token"] == "access"