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 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": "电脑端即将登录，可关闭本页"})