feat(api): confirm web QR login from mini program phone auth.

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

backend/app/api/mini_auth.py

@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
 
 from app.core.database import get_db
 from app.core.response import error_response, success_response
+from app.services import wx_qr_session
 from app.services.web_login_tokens import issue_web_login_tokens
 from app.services.wechat_login_service import resolve_wechat_login_user
 from app.services.wechat_mini_auth import (
@@ -56,3 +57,65 @@ def wechat_login(body: dict, db: Session = Depends(get_db)):
     except ValueError as exc:
         return error_response(403, str(exc))
     return success_response(data)
+
+
+@router.post("/web-qr-scanned", response_model=dict)
+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)
+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 = issue_web_login_tokens(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,4 +1,5 @@
 """web wx-qr create/status API。"""
+from types import SimpleNamespace
 from unittest.mock import patch
 
 import pytest
@@ -31,3 +32,38 @@ 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"},
+    }
+    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_web_login_tokens", 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"