77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
|
|
from core.deps import get_auth_context
|
|
from external.rw import build_subscription_link, delete_hwid, get_hwid_list, get_rw_user, get_sdk
|
|
from schemas.common import OperationData
|
|
from schemas.devices import Device
|
|
from schemas.dto import AuthContext
|
|
from schemas.user import SubscriptionData, UserInfo
|
|
|
|
router = APIRouter(prefix="/users")
|
|
|
|
|
|
@router.get("/me", response_model=UserInfo)
|
|
async def get_me(ctx: AuthContext = Depends(get_auth_context)):
|
|
return UserInfo(
|
|
username=ctx.user.username,
|
|
telegram_id=ctx.user.telegram_id,
|
|
referal_code=ctx.user.referal_code,
|
|
)
|
|
|
|
|
|
@router.get("/subscription")
|
|
async def get_subscription(ctx: AuthContext = Depends(get_auth_context)):
|
|
sub = ctx.user.subscription
|
|
if not sub:
|
|
return SubscriptionData(
|
|
has_subscription=False,
|
|
devices=None,
|
|
expires_at=None,
|
|
addon_ids=[],
|
|
subscription_link=None,
|
|
)
|
|
rw_user = await get_rw_user(get_sdk(), ctx.user.telegram_id, ctx.user.username)
|
|
|
|
return SubscriptionData(
|
|
has_subscription=True,
|
|
devices=sub.devices,
|
|
expires_at=sub.expires_at,
|
|
addon_ids=[a.addon_id for a in sub.addons],
|
|
subscription_link=build_subscription_link(rw_user.short_uuid),
|
|
)
|
|
|
|
|
|
@router.get("/subscription/hwid", response_model=list[Device])
|
|
async def get_hwid(ctx: AuthContext = Depends(get_auth_context)):
|
|
rw_user = await get_rw_user(get_sdk(), ctx.user.telegram_id, ctx.user.username)
|
|
if not rw_user:
|
|
raise HTTPException(403, detail="RW user not found.")
|
|
hwid_list = await get_hwid_list(get_sdk(), user_uuid=rw_user.uuid)
|
|
devices = []
|
|
|
|
if not hwid_list:
|
|
return devices
|
|
|
|
for device in hwid_list:
|
|
devices.append(
|
|
Device(
|
|
os=device.platform,
|
|
model=device.device_model,
|
|
client=device.user_agent.split("/")[0],
|
|
hwid=device.hwid,
|
|
)
|
|
)
|
|
|
|
return devices
|
|
|
|
|
|
@router.delete("/subscription/hwid")
|
|
async def delete_hwid_endpoint(hwid: str, ctx: AuthContext = Depends(get_auth_context)):
|
|
rw_user = await get_rw_user(get_sdk(), ctx.user.telegram_id, ctx.user.username)
|
|
if not rw_user:
|
|
raise HTTPException(403, detail="RW user not found.")
|
|
|
|
success = await delete_hwid(get_sdk(), rw_user.uuid, hwid)
|
|
|
|
return OperationData(success=success)
|