diff --git a/backend/app/api/mini_auth.py b/backend/app/api/mini_auth.py index dad3feccf..9114a15b7 100644 --- a/backend/app/api/mini_auth.py +++ b/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": "电脑端即将登录,可关闭本页"}) diff --git a/backend/tests/test_wx_qr_api.py b/backend/tests/test_wx_qr_api.py index 42d03d48f..32dc5a500 100644 --- a/backend/tests/test_wx_qr_api.py +++ b/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"