411 lines
12 KiB
Python
411 lines
12 KiB
Python
"""Pydantic 请求/响应模型"""
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Optional, Any
|
|
from pydantic import BaseModel, Field, ConfigDict, model_serializer
|
|
|
|
from .huya_defaults import (
|
|
HUYA_DEFAULT_BIND_ACT_ID,
|
|
HUYA_DEFAULT_OUTER_ACT_ID,
|
|
HUYA_DEFAULT_PAY_CHANNEL,
|
|
HUYA_DEFAULT_ROOM_PID,
|
|
HUYA_DEFAULT_SID,
|
|
)
|
|
|
|
|
|
def _ensure_tz(dt: Optional[datetime]) -> Optional[datetime]:
|
|
"""确保 datetime 带有 UTC 时区信息，无时区的视为 UTC。"""
|
|
if dt is None:
|
|
return None
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
|
|
|
|
# ---- 认证 ----
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class TokenResponse(BaseModel):
|
|
access_token: str
|
|
token_type: str = "bearer"
|
|
role: str
|
|
username: str
|
|
permissions: list[str]
|
|
|
|
|
|
class UserInfo(BaseModel):
|
|
id: int
|
|
username: str
|
|
role: str
|
|
is_active: bool
|
|
remark: str = ""
|
|
created_at: Optional[datetime] = None
|
|
permissions: list[str] = []
|
|
custom_permissions: Optional[list[str]] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"username": self.username,
|
|
"role": self.role,
|
|
"is_active": self.is_active,
|
|
"remark": self.remark,
|
|
"created_at": _ensure_tz(self.created_at).isoformat() if self.created_at else None,
|
|
"permissions": self.permissions,
|
|
"custom_permissions": self.custom_permissions,
|
|
}
|
|
|
|
|
|
# ---- 用户管理 ----
|
|
class UserCreate(BaseModel):
|
|
username: str = Field(..., min_length=2, max_length=64)
|
|
password: str = Field(..., min_length=6, max_length=128)
|
|
role: str = Field("support", pattern="^(super_admin|operation|support)$")
|
|
remark: str = ""
|
|
|
|
|
|
class UserUpdate(BaseModel):
|
|
password: Optional[str] = None
|
|
role: Optional[str] = None
|
|
is_active: Optional[bool] = None
|
|
remark: Optional[str] = None
|
|
custom_permissions: Optional[list[str]] = None
|
|
|
|
|
|
# ---- 账号 ----
|
|
class AccountImport(BaseModel):
|
|
"""批量导入，文本格式：用户名|密码|邮箱|邮箱密码"""
|
|
text: str
|
|
|
|
|
|
class AccountAssign(BaseModel):
|
|
assigned_to: Optional[int] = None
|
|
|
|
|
|
class BatchAssign(BaseModel):
|
|
account_ids: list[int]
|
|
assigned_to: Optional[int] = None # None=取消分配
|
|
|
|
|
|
class AccountTag(BaseModel):
|
|
tag: Optional[str] = None
|
|
account_ids: Optional[list[int]] = None
|
|
|
|
|
|
class AccountOut(BaseModel):
|
|
id: int
|
|
username: str
|
|
# 敏感字段根据角色决定是否返回
|
|
password: Optional[str] = None
|
|
email: Optional[str] = None
|
|
email_password: Optional[str] = None
|
|
tag: str = ""
|
|
assigned_to: Optional[int] = None
|
|
assigned_username: Optional[str] = None
|
|
remark: str = ""
|
|
created_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"username": self.username,
|
|
"password": self.password,
|
|
"email": self.email,
|
|
"email_password": self.email_password,
|
|
"tag": self.tag,
|
|
"assigned_to": self.assigned_to,
|
|
"assigned_username": self.assigned_username,
|
|
"remark": self.remark,
|
|
"created_at": _ensure_tz(self.created_at).isoformat() if self.created_at else None,
|
|
}
|
|
|
|
|
|
# ---- 登录任务 ----
|
|
class LoginBatchRequest(BaseModel):
|
|
account_ids: list[int]
|
|
max_login_retries: int = 0 # 登录整体重试次数，0=无限重试直到成功（后端兜底 20 次）
|
|
max_total_time: float = 300 # 单账号登录总时长上限（秒），0=不限制
|
|
concurrency: int = 3 # 并发数，1-10
|
|
api_strategy: str = "wgapi" # 接口策略: wgapi（新版）或 iframe（旧版备选）
|
|
|
|
|
|
class LoginTaskOut(BaseModel):
|
|
id: int
|
|
batch_id: str
|
|
account_id: int
|
|
account_username: str = ""
|
|
status: str
|
|
cookie: str = ""
|
|
message: str = ""
|
|
created_by: int
|
|
created_at: Optional[datetime] = None
|
|
finished_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"batch_id": self.batch_id,
|
|
"account_id": self.account_id,
|
|
"account_username": self.account_username,
|
|
"status": self.status,
|
|
"cookie": self.cookie,
|
|
"message": self.message,
|
|
"created_by": self.created_by,
|
|
"created_at": _ensure_tz(self.created_at).isoformat() if self.created_at else None,
|
|
"finished_at": _ensure_tz(self.finished_at).isoformat() if self.finished_at else None,
|
|
}
|
|
|
|
|
|
# ---- 虎牙 ----
|
|
class HuyaCookieImport(BaseModel):
|
|
"""批量导入虎牙 Cookie。支持纯 CK 或 账号----密码----CK。"""
|
|
text: str
|
|
tag: str = ""
|
|
|
|
|
|
class HuyaPasswordLoginRequest(BaseModel):
|
|
"""虎牙账号密码登录并保存 Cookie。"""
|
|
username: str = Field(..., min_length=1, max_length=128)
|
|
password: str = Field(..., min_length=1, max_length=128)
|
|
tag: str = ""
|
|
cookie: str = ""
|
|
|
|
|
|
class HuyaAccountOut(BaseModel):
|
|
id: int
|
|
uid: str = ""
|
|
yyuid: str = ""
|
|
username: str = ""
|
|
nickname: str = ""
|
|
cookie: str = ""
|
|
cookie_preview: str = ""
|
|
tag: str = ""
|
|
remark: str = ""
|
|
status: str = ""
|
|
points: Optional[int] = None
|
|
game_name: str = ""
|
|
game_channel: str = ""
|
|
game_phone: str = ""
|
|
assigned_to: Optional[int] = None
|
|
assigned_username: Optional[str] = None
|
|
created_at: Optional[datetime] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"uid": self.uid,
|
|
"yyuid": self.yyuid,
|
|
"username": self.username,
|
|
"nickname": self.nickname,
|
|
"cookie": self.cookie,
|
|
"cookie_preview": self.cookie_preview,
|
|
"tag": self.tag,
|
|
"remark": self.remark,
|
|
"status": self.status,
|
|
"points": self.points,
|
|
"game_name": self.game_name,
|
|
"game_channel": self.game_channel,
|
|
"game_phone": self.game_phone,
|
|
"assigned_to": self.assigned_to,
|
|
"assigned_username": self.assigned_username,
|
|
"created_at": _ensure_tz(self.created_at).isoformat() if self.created_at else None,
|
|
"updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class HuyaConfigOut(BaseModel):
|
|
room_pid: str = HUYA_DEFAULT_ROOM_PID
|
|
sid: str = HUYA_DEFAULT_SID
|
|
outer_act_id: str = HUYA_DEFAULT_OUTER_ACT_ID
|
|
bind_act_id: str = HUYA_DEFAULT_BIND_ACT_ID
|
|
pay_channel: str = HUYA_DEFAULT_PAY_CHANNEL
|
|
updated_at: Optional[datetime] = None
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"room_pid": self.room_pid,
|
|
"sid": self.sid,
|
|
"outer_act_id": self.outer_act_id,
|
|
"bind_act_id": self.bind_act_id,
|
|
"pay_channel": self.pay_channel,
|
|
"updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class HuyaConfigUpdate(BaseModel):
|
|
room_pid: Optional[str] = None
|
|
sid: Optional[str] = None
|
|
outer_act_id: Optional[str] = None
|
|
bind_act_id: Optional[str] = None
|
|
pay_channel: Optional[str] = None
|
|
|
|
|
|
class HuyaTaskBatchRequest(BaseModel):
|
|
account_ids: list[int]
|
|
task_type: str
|
|
concurrency: int = 3
|
|
payload: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class HuyaTaskOut(BaseModel):
|
|
id: int
|
|
batch_id: str
|
|
account_id: int
|
|
account_uid: str = ""
|
|
account_nickname: str = ""
|
|
task_type: str
|
|
status: str
|
|
message: str = ""
|
|
result: Optional[dict[str, Any]] = None
|
|
created_by: int
|
|
created_at: Optional[datetime] = None
|
|
finished_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"batch_id": self.batch_id,
|
|
"account_id": self.account_id,
|
|
"account_uid": self.account_uid,
|
|
"account_nickname": self.account_nickname,
|
|
"task_type": self.task_type,
|
|
"status": self.status,
|
|
"message": self.message,
|
|
"result": self.result,
|
|
"created_by": self.created_by,
|
|
"created_at": _ensure_tz(self.created_at).isoformat() if self.created_at else None,
|
|
"finished_at": _ensure_tz(self.finished_at).isoformat() if self.finished_at else None,
|
|
}
|
|
|
|
|
|
class HuyaGoodsOut(BaseModel):
|
|
id: int
|
|
product_id: str
|
|
name: str = ""
|
|
price: Optional[int] = None
|
|
remain_text: str = ""
|
|
raw: Optional[dict[str, Any]] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"product_id": self.product_id,
|
|
"name": self.name,
|
|
"price": self.price,
|
|
"remain_text": self.remain_text,
|
|
"raw": self.raw,
|
|
"updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
class HuyaRechargeGoodsOut(BaseModel):
|
|
id: int
|
|
spu_id: str
|
|
sku_id: str = ""
|
|
name: str = ""
|
|
price: Optional[int] = None
|
|
stock: Optional[int] = None
|
|
buy_limit: Optional[int] = None
|
|
icon: str = ""
|
|
description: str = ""
|
|
task_id: str = ""
|
|
task_name: str = ""
|
|
raw: Optional[dict[str, Any]] = None
|
|
updated_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@model_serializer
|
|
def _serialize(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"spu_id": self.spu_id,
|
|
"sku_id": self.sku_id,
|
|
"name": self.name,
|
|
"price": self.price,
|
|
"stock": self.stock,
|
|
"buy_limit": self.buy_limit,
|
|
"icon": self.icon,
|
|
"description": self.description,
|
|
"task_id": self.task_id,
|
|
"task_name": self.task_name,
|
|
"raw": self.raw,
|
|
"updated_at": _ensure_tz(self.updated_at).isoformat() if self.updated_at else None,
|
|
}
|
|
|
|
|
|
# ---- 代理配置 ----
|
|
class ProxyConfigOut(BaseModel):
|
|
enabled: bool = False
|
|
api_url: str = ""
|
|
http: str = ""
|
|
https: str = ""
|
|
whitelist_enabled: bool = False
|
|
whitelist_platform: str = "xiequ"
|
|
whitelist_credentials: Optional[dict] = None
|
|
# 旧字段保留（向后兼容）
|
|
whitelist_uid: str = ""
|
|
whitelist_ukey: str = ""
|
|
|
|
|
|
class ProxyConfigUpdate(BaseModel):
|
|
enabled: Optional[bool] = None
|
|
api_url: Optional[str] = None
|
|
http: Optional[str] = None
|
|
https: Optional[str] = None
|
|
whitelist_enabled: Optional[bool] = None
|
|
whitelist_platform: Optional[str] = "xiequ"
|
|
whitelist_credentials: Optional[dict] = None
|
|
# 旧字段保留（向后兼容）
|
|
whitelist_uid: Optional[str] = None
|
|
whitelist_ukey: Optional[str] = None
|
|
|
|
|
|
# ---- 代理平台元信息 ----
|
|
class PlatformFieldDef(BaseModel):
|
|
"""平台凭据字段定义。"""
|
|
key: str
|
|
label: str
|
|
placeholder: str = ""
|
|
|
|
|
|
class PlatformInfo(BaseModel):
|
|
"""平台元信息。"""
|
|
name: str
|
|
label: str
|
|
credential_fields: list[PlatformFieldDef]
|
|
|
|
|
|
# ---- 通用 ----
|
|
class AppInfo(BaseModel):
|
|
version: str
|
|
|
|
|
|
class MessageResponse(BaseModel):
|
|
message: str
|
|
success: bool = True
|