59 lines
1.9 KiB
Python
59 lines
1.9 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.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)
|