122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
"""小程序鉴权：微信手机号快捷登录。"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends
|
|
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 (
|
|
WechatMiniAuthError,
|
|
code2session,
|
|
get_phone_number,
|
|
)
|
|
|
|
router = APIRouter(prefix="/mini/auth", tags=["小程序鉴权"])
|
|
|
|
|
|
@router.post("/wechat-login", response_model=dict)
|
|
def wechat_login(body: dict, db: Session = Depends(get_db)):
|
|
"""手机号快捷登录：须 loginCode + phoneCode；匹配 phone_no 并绑定 wx_openid。"""
|
|
login_code = (body.get("loginCode") or "").strip()
|
|
phone_code = (body.get("phoneCode") or "").strip()
|
|
if not phone_code:
|
|
return error_response(400, "请授权手机号登录，或使用账号密码登录")
|
|
|
|
try:
|
|
sess = code2session(login_code)
|
|
except WechatMiniAuthError as exc:
|
|
return error_response(400, str(exc))
|
|
|
|
try:
|
|
phone = get_phone_number(phone_code)
|
|
except WechatMiniAuthError as 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:
|
|
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:
|
|
data = issue_web_login_tokens(db, user)
|
|
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": "电脑端即将登录，可关闭本页"})
|