1import secrets
2from datetime import datetime
3from enum import Enum
4from typing import List, Optional, Union, Annotated, Literal
5
6from pydantic import (
7 field_validator,
8 ConfigDict,
9 BaseModel,
10 Field,
11 computed_field,
12 StringConstraints,
13 model_validator,
14)
15
16from config import XRAY_SUBSCRIPTION_URL_PREFIX
17
18USERNAME_REGEXP = r"^\w{3,32}$"
19
20
21class ReminderType(str, Enum):
22 expiration_date = "expiration_date"
23 data_usage = "data_usage"
24
25
26class UserStatus(str, Enum):
27 active = "active"
28 on_hold = "on_hold"
29 limited = "limited"
30 expired = "expired"
31
32
33class UserStatusModify(str, Enum):
34 active = "active"
35 on_hold = "on_hold"
36
37
38class UserStatusCreate(str, Enum):
39 active = "active"
40 on_hold = "on_hold"
41
42
43class UserDataLimitResetStrategy(str, Enum):
44 no_reset = "no_reset"
45 day = "day"
46 week = "week"
47 month = "month"
48 year = "year"
49
50
51class UserBase(BaseModel):
52 id: Optional[int] = None
53 username: Annotated[str, StringConstraints(to_lower=True, pattern=USERNAME_REGEXP)]
54 expire: Union[datetime, None, Literal[0]] = Field(None)
55 key: str = Field(default_factory=lambda: secrets.token_hex(16))
56 data_limit: Optional[int] = Field(
57 ge=0, default=None, description="data_limit can be 0 or greater"
58 )
59 enabled: bool = Field(default=True)
60 data_limit_reset_strategy: UserDataLimitResetStrategy = (
61 UserDataLimitResetStrategy.no_reset
62 )
63 note: Optional[Annotated[str, Field(max_length=500)]] = None
64 sub_updated_at: Optional[datetime] = Field(None)
65 sub_last_user_agent: Optional[str] = Field(None)
66 online_at: Optional[datetime] = Field(None)
67 on_hold_expire_duration: Optional[int] = Field(None)
68 on_hold_timeout: Optional[datetime] = Field(None)
69
70 model_config = ConfigDict(from_attributes=True)
71
72
73from app.models.service import ServiceBase
74from app.models.proxy import InboundBase
75
76
77class User(UserBase):
78 pass
79
80
81class UserCreate(User):
82 status: UserStatusCreate = Field(UserStatusCreate.active)
83 services: List[int] = Field([])
84 model_config = ConfigDict(
85 json_schema_extra={
86 "example": {
87 "username": "user1234",
88 "services": [1, 2, 3],
89 "expire": "2023-11-03T20:30:00",
90 "data_limit": 0,
91 "data_limit_reset_strategy": "no_reset",
92 "status": "active",
93 "note": "",
94 "on_hold_timeout": "2023-11-03T20:30:00",
95 "on_hold_expire_duration": 0,
96 }
97 }
98 )
99
100 @model_validator(mode="after")
101 def validate_expiry(self):
102 if self.status == UserStatusCreate.on_hold:
103 if not self.on_hold_expire_duration:
104 raise ValueError(
105 "User cannot be on hold without a valid on_hold_expire_duration."
106 )
107 if self.expire:
108 raise ValueError("User cannot be on hold with specified expire.")
109 elif self.status == UserStatusCreate.active and (self.on_hold_expire_duration or self.on_hold_timeout):
110 raise ValueError(
111 "on hold parametrs set when user status isn't on_hold."
112 )
113 return self
114
115 @field_validator("expire")
116 @classmethod
117 def validate_expire(cls, v: Union[datetime, None, Literal[0]]):
118 if isinstance(v, datetime) and (v.tzinfo is not None):
119 raise ValueError(
120 "Expire date should be offset naive, and preferably in utc timezone."
121 )
122 if isinstance(v, datetime) and v < datetime.utcnow():
123 raise ValueError("Expire date should be in the future, not in the past.")
124 return v
125
126
127class UserModify(User):
128 status: Optional[UserStatusModify] = Field(None)
129 services: List[int] = Field([])
130 data_limit_reset_strategy: Optional[UserDataLimitResetStrategy] = Field(None)
131 model_config = ConfigDict(
132 json_schema_extra={
133 "example": {
134 "username": "mammad1234",
135 "services": [1, 2, 3],
136 "expire": 0,
137 "data_limit": 0,
138 "data_limit_reset_strategy": "no_reset",
139 "status": "active",
140 "note": "",
141 "on_hold_timeout": "2023-11-03T20:30:00",
142 "on_hold_expire_duration": 0,
143 }
144 }
145 )
146
147 @model_validator(mode="after")
148 def validate_expiry(self):
149 if self.on_hold_expire_duration and self.expire:
150 raise ValueError(
151 "can't set both expire and on hold expire at the same time."
152 )
153 return self
154
155 @field_validator("expire")
156 @classmethod
157 def validate_expire(cls, v: Union[datetime, None, Literal[0]]):
158 if isinstance(v, datetime) and v.tzinfo is not None:
159 raise ValueError(
160 "Expire date should be offset naive, and preferably in utc timezone."
161 )
162 if isinstance(v, datetime) and v < datetime.utcnow():
163 raise ValueError("Expire date should be in the future, not in the past.")
164 return v
165
166
167class UserResponse(User):
168 id: int
169 status: UserStatus
170 used_traffic: int
171 lifetime_used_traffic: int = 0
172 created_at: datetime
173 inbounds: List[InboundBase] = Field([])
174 services: List[ServiceBase] = Field([])
175 links: List[str] | None = [] # Field(None)
176
177 # subscription_url: str = ""
178 model_config = ConfigDict(from_attributes=True)
179
180 @computed_field
181 @property
182 def service_ids(self) -> List[int]:
183 return [s.id for s in self.services]
184
185 @computed_field
186 @property
187 def subscription_url(self) -> str:
188 salt = secrets.token_hex(8)
189 url_prefix = XRAY_SUBSCRIPTION_URL_PREFIX.replace("*", salt)
190 return f"{url_prefix}/sub/{self.username}/{self.key}"
191
192
193class UserUsageResponse(BaseModel):
194 node_id: Union[int, None] = None
195 node_name: str
196 used_traffic: int
197
198
199class UserUsagesResponse(BaseModel):
200 username: str
201 usages: List[UserUsageResponse]