wechat-scan/auth.py

"""
授权接口与核心业务逻辑

路由（MFC 侧）：
    POST /auth/create_scene  生成 scene_str、创建微信临时二维码并落库
    GET  /auth/status        轮询扫码授权结果

业务函数（微信事件侧，由 wechat.py 调用）：
    handle_scan()            处理扫码事件：建用户、发免费授权、绑定场景、签发会话
"""

import logging
import secrets

import aiomysql
from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel

import db
from config import FREE_AUTH_DAYS, SCENE_TTL_SECONDS, SESSION_TTL_HOURS
from wechat_api import create_temp_qrcode

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/auth", tags=["auth"])

SCENE_PREFIX = "pc_"


class CreateSceneRequest(BaseModel):
    device_id: str


@router.post("/create_scene")
async def create_scene(payload: CreateSceneRequest):
    """生成唯一 scene_str，调用微信接口创建临时二维码，写入 auth_scenes"""
    scene_str = SCENE_PREFIX + secrets.token_hex(16)
    try:
        qr_url = await create_temp_qrcode(scene_str, SCENE_TTL_SECONDS)
    except Exception as exc:
        # 微信接口不可用（appid/secret 未配置、网络异常、token 失效等）
        logger.exception("创建二维码失败 scene_str=%s", scene_str)
        raise HTTPException(status_code=502, detail=f"创建微信二维码失败: {exc}")

    async with db.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "INSERT INTO auth_scenes (scene_str, device_id, status, created_at, expires_at) "
                "VALUES (%s, %s, 'pending', NOW(), DATE_ADD(NOW(), INTERVAL %s SECOND))",
                (scene_str, payload.device_id, SCENE_TTL_SECONDS),
            )

    return {"scene_str": scene_str, "qr_url": qr_url, "expires_in": SCENE_TTL_SECONDS}


@router.get("/status")
async def get_status(scene: str = Query(..., description="create_scene 返回的 scene_str")):
    """查询扫码授权状态：pending / authorized / expired / need_purchase"""
    async with db.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute(
                "SELECT id, status, user_id, (expires_at > NOW()) AS not_expired "
                "FROM auth_scenes WHERE scene_str = %s",
                (scene,),
            )
            scene_row = await cur.fetchone()
            if scene_row is None:
                raise HTTPException(status_code=404, detail="scene 不存在")

            # 尚未扫码：过期则惰性置为 expired
            if scene_row["status"] in ("pending", "scanned"):
                if not scene_row["not_expired"]:
                    await cur.execute(
                        "UPDATE auth_scenes SET status = 'expired' "
                        "WHERE id = %s AND status IN ('pending', 'scanned')",
                        (scene_row["id"],),
                    )
                    return {"status": "expired"}
                return {"status": "pending"}

            if scene_row["status"] == "expired":
                return {"status": "expired"}

            # 已扫码授权：判断用户当前是否有可用授权
            auth_row = await _get_active_authorization(cur, scene_row["user_id"])
            if auth_row is None:
                # 免费已领过且无有效授权 → 引导充值（阶段 3）
                return {"status": "need_purchase"}

            await cur.execute(
                "SELECT token, (expires_at > NOW()) AS not_expired FROM sessions "
                "WHERE scene_id = %s ORDER BY id DESC LIMIT 1",
                (scene_row["id"],),
            )
            session_row = await cur.fetchone()
            if session_row is None or not session_row["not_expired"]:
                return {"status": "expired"}

            return {
                "status": "authorized",
                "session_token": session_row["token"],
                "authorization": _serialize_authorization(auth_row),
            }


# ---------------------------------------------------------------------------
# 业务逻辑
# ---------------------------------------------------------------------------

async def handle_scan(scene_str: str, openid: str) -> bool:
    """
    处理扫码事件。

    返回 True 表示本次扫码完成授权，False 表示场景无效、已过期或已被处理。
    整个流程在事务内完成，并对 scene 行加排他锁，保证同一 scene 只被处理一次。
    """
    async with db.acquire() as conn:
        await conn.begin()
        try:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                await cur.execute(
                    "SELECT id, status, device_id, (expires_at > NOW()) AS not_expired "
                    "FROM auth_scenes WHERE scene_str = %s FOR UPDATE",
                    (scene_str,),
                )
                scene_row = await cur.fetchone()

                if scene_row is None:
                    await conn.rollback()
                    return False

                if scene_row["status"] == "authorized":
                    # 重复扫码：只处理第一次，后续忽略
                    await conn.rollback()
                    return False

                if not scene_row["not_expired"]:
                    await cur.execute(
                        "UPDATE auth_scenes SET status = 'expired' "
                        "WHERE id = %s AND status IN ('pending', 'scanned')",
                        (scene_row["id"],),
                    )
                    await conn.commit()
                    return False

                user_id = await _find_or_create_user(cur, openid)
                await _grant_free_authorization(cur, user_id)

                await cur.execute(
                    "UPDATE auth_scenes SET status = 'authorized', user_id = %s, authorized_at = NOW() "
                    "WHERE id = %s",
                    (user_id, scene_row["id"]),
                )

                token = secrets.token_urlsafe(32)
                await cur.execute(
                    "INSERT INTO sessions (token, user_id, device_id, scene_id, created_at, expires_at) "
                    "VALUES (%s, %s, %s, %s, NOW(), DATE_ADD(NOW(), INTERVAL %s HOUR))",
                    (token, user_id, scene_row["device_id"], scene_row["id"], SESSION_TTL_HOURS),
                )

            await conn.commit()
            logger.info("扫码授权成功 scene_str=%s openid=%s", scene_str, openid)
            return True
        except Exception:
            await conn.rollback()
            raise


async def _find_or_create_user(cur, openid: str) -> int:
    """按 openid 查找用户，不存在则创建，并刷新 last_seen_at"""
    await cur.execute("SELECT id FROM users WHERE openid = %s", (openid,))
    row = await cur.fetchone()
    if row is None:
        # INSERT IGNORE + 重查：并发扫码时避免唯一键冲突报错
        await cur.execute(
            "INSERT IGNORE INTO users (openid, created_at, last_seen_at, has_claimed_free) "
            "VALUES (%s, NOW(), NOW(), 0)",
            (openid,),
        )
        await cur.execute("SELECT id FROM users WHERE openid = %s", (openid,))
        row = await cur.fetchone()

    await cur.execute("UPDATE users SET last_seen_at = NOW() WHERE id = %s", (row["id"],))
    return row["id"]


async def _grant_free_authorization(cur, user_id: int) -> None:
    """
    首次关注赠送 7 天时间授权。

    以 has_claimed_free 的条件更新作为幂等闸门：只有把 0 改成 1 的那一次才真正发授权。
    若用户已有 active 授权（互斥原则），新授权以 pending 保存。
    """
    await cur.execute(
        "UPDATE users SET has_claimed_free = 1 WHERE id = %s AND has_claimed_free = 0",
        (user_id,),
    )
    if cur.rowcount != 1:
        return

    status = "pending" if await _has_active_authorization(cur, user_id) else "active"
    await cur.execute(
        "INSERT INTO authorizations "
        "(user_id, type, start_at, end_at, remaining_points, total_points, source, status, created_at, updated_at) "
        "VALUES (%s, 'time', NOW(), DATE_ADD(NOW(), INTERVAL %s DAY), 0, 0, 'free', %s, NOW(), NOW())",
        (user_id, FREE_AUTH_DAYS, status),
    )
    logger.info("已发放免费授权 user_id=%s days=%s status=%s", user_id, FREE_AUTH_DAYS, status)


async def _has_active_authorization(cur, user_id: int) -> bool:
    await cur.execute(
        "SELECT 1 FROM authorizations WHERE user_id = %s AND status = 'active' LIMIT 1",
        (user_id,),
    )
    return await cur.fetchone() is not None


async def _get_active_authorization(cur, user_id: int):
    """取用户当前 active 授权；已失效的惰性置为 expired / exhausted 并返回 None"""
    await cur.execute(
        "SELECT id, type, end_at, remaining_points, total_points, "
        "(end_at IS NOT NULL AND end_at > NOW()) AS time_valid "
        "FROM authorizations WHERE user_id = %s AND status = 'active' ORDER BY id DESC LIMIT 1",
        (user_id,),
    )
    row = await cur.fetchone()
    if row is None:
        return None

    if row["type"] == "time":
        if not row["time_valid"]:
            await cur.execute(
                "UPDATE authorizations SET status = 'expired' WHERE id = %s AND status = 'active'",
                (row["id"],),
            )
            return None
        return row

    if row["remaining_points"] <= 0:
        await cur.execute(
            "UPDATE authorizations SET status = 'exhausted' WHERE id = %s AND status = 'active'",
            (row["id"],),
        )
        return None
    return row


def _serialize_authorization(row) -> dict:
    return {
        "type": row["type"],
        "end_at": row["end_at"].isoformat() if row["end_at"] else None,
        "remaining_points": row["remaining_points"],
    }