shanxia2.0/backend/app/api/mini_auth.py

"""小程序鉴权：微信手机号快捷登录。"""
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)