alphax/app/web/shared.py

"""Shared web-layer request models and auth helpers."""

from contextvars import ContextVar

from fastapi import Cookie, HTTPException, Request
from fastapi.responses import RedirectResponse
from pydantic import BaseModel

from app.db import auth_db

current_request = ContextVar("current_request", default=None)


class RegisterRequest(BaseModel):
    email: str
    password: str
    invite_code: str = ""


class VerifyEmailRequest(BaseModel):
    email: str
    code: str


class LoginRequest(BaseModel):
    email: str
    password: str


class ResendVerificationRequest(BaseModel):
    email: str


class SendCodeRequest(BaseModel):
    email: str


class CompleteRegistrationRequest(BaseModel):
    email: str
    code: str
    password: str
    invite_code: str = ""


class ChangePasswordRequest(BaseModel):
    old_password: str
    new_password: str


class WatchlistRequest(BaseModel):
    symbol: str


class ObservationRequest(BaseModel):
    rec_id: int
    note: str = ""


class PushRulesRequest(BaseModel):
    watchlist_only: bool = False
    min_score: int = 0
    min_rr: float = 0
    push_buy_now: bool = True
    push_wait_pullback: bool = True
    push_observe: bool = False
    quiet_start: str = ""
    quiet_end: str = ""


class SchedulerToggleRequest(BaseModel):
    enabled: bool


class SchedulerIntervalRequest(BaseModel):
    every_seconds: int


class SchedulerTriggerRequest(BaseModel):
    force: bool = False


class RuntimeConfigRequest(BaseModel):
    config: dict | list
    description: str = ""


class ChatSessionRequest(BaseModel):
    title: str = ""


class ChatSendRequest(BaseModel):
    session_id: int = 0
    message: str


def auth_error(exc: Exception, status_code: int = 400):
    raise HTTPException(status_code=status_code, detail=str(exc))


def is_local_request(request: Request = None) -> bool:
    request = request or current_request.get()
    if not request or not request.client:
        return False
    host = (request.client.host or "").split(":")[0]
    return host in ("127.0.0.1", "localhost", "::1", "testclient")


def local_debug_user():
    return {"id": 0, "email": "local@alphax.dev", "is_admin": True, "local_debug": True}


def require_user(altcoin_session: str = ""):
    user = auth_db.get_user_by_session_token(altcoin_session)
    if user:
        return user
    if is_local_request():
        return local_debug_user()
    if not user:
        raise HTTPException(status_code=401, detail="请先登录")
    return user


def require_active_subscription(altcoin_session: str = ""):
    user = require_user(altcoin_session)
    if user.get("local_debug"):
        return user, {"plan_name": "本地调试", "local_debug": True}
    sub = auth_db.get_current_subscription(user["id"])
    if not sub:
        raise HTTPException(status_code=402, detail="订阅已过期或未开通，请先开通订阅")
    return user, sub


def require_admin(altcoin_session: str = ""):
    user = require_user(altcoin_session)
    if user.get("local_debug"):
        return user
    if not auth_db.is_user_admin(user["id"]):
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return user


def login_redirect():
    return RedirectResponse(url="/auth?tab=login", status_code=302)


def subscription_redirect():
    return RedirectResponse(url="/subscription?expired=1", status_code=302)


def has_active_subscription(user) -> bool:
    if not user:
        return False
    if user.get("local_debug"):
        return True
    try:
        if auth_db.is_user_admin(user["id"]):
            return True
    except Exception:
        pass
    return bool(auth_db.get_current_subscription(user["id"]))


def require_page_user(request: Request, require_subscription: bool = True):
    user = auth_db.get_user_by_session_token(request.cookies.get("altcoin_session", ""))
    if user:
        if require_subscription and not has_active_subscription(user):
            return user, subscription_redirect()
        return user, None
    if is_local_request(request):
        return local_debug_user(), None
    if not user:
        return None, login_redirect()
    return user, None


def require_api_user_with_subscription(altcoin_session: str = Cookie(default="")):
    user = require_user(altcoin_session)
    if user.get("local_debug"):
        return user
    if not has_active_subscription(user):
        raise HTTPException(status_code=402, detail="订阅已过期或未开通，请先开通订阅")
    return user