1import secrets
  2from datetime import datetime
  3from enum import Enum
  4from typing import Annotated
  5
  6from pydantic import (
  7    ConfigDict,
  8    BaseModel,
  9    Field,
 10    computed_field,
 11    StringConstraints,
 12    model_validator,
 13)
 14
 15from app.config.env import SUBSCRIPTION_URL_PREFIX
 16
 17USERNAME_REGEXP = r"^\w{3,32}$"
 18
 19
 20class ReminderType(str, Enum):
 21    expiration_date = "expiration_date"
 22    data_usage = "data_usage"
 23
 24
 25class UserStatus(str, Enum):
 26    ACTIVE = "active"
 27    INACTIVE = "inactive"
 28
 29
 30class UserDataUsageResetStrategy(str, Enum):
 31    no_reset = "no_reset"
 32    day = "day"
 33    week = "week"
 34    month = "month"
 35    year = "year"
 36
 37
 38class UserExpireStrategy(str, Enum):
 39    NEVER = "never"
 40    FIXED_DATE = "fixed_date"
 41    START_ON_FIRST_USE = "start_on_first_use"
 42
 43
 44class User(BaseModel):
 45    id: int | None = None
 46    username: Annotated[
 47        str, StringConstraints(to_lower=True, pattern=USERNAME_REGEXP)
 48    ]
 49    expire_strategy: UserExpireStrategy
 50    expire_date: datetime | None = Field(None)
 51    usage_duration: int | None = Field(None)
 52    activation_deadline: datetime | None = Field(None)
 53    key: str = Field(default_factory=lambda: secrets.token_hex(16))
 54    data_limit: int | None = Field(
 55        ge=0, default=None, description="data_limit can be 0 or greater"
 56    )
 57    data_limit_reset_strategy: UserDataUsageResetStrategy = (
 58        UserDataUsageResetStrategy.no_reset
 59    )
 60    note: Annotated[str, Field(max_length=500)] | None = None
 61    sub_updated_at: datetime | None = Field(None)
 62    sub_last_user_agent: str | None = Field(None)
 63    online_at: datetime | None = Field(None)
 64
 65    model_config = ConfigDict(from_attributes=True)
 66
 67
 68class UserCreate(User):
 69    service_ids: list[int] = Field([])
 70    model_config = ConfigDict(
 71        json_schema_extra={
 72            "example": {
 73                "username": "user1234",
 74                "services": [1, 2, 3],
 75                "expire_strategy": "start_on_first_use",
 76                "usage_duration": 86400 * 14,
 77                "activation_deadline": "2024-11-03T20:30:00",
 78                "data_limit": 0,
 79                "data_limit_reset_strategy": "no_reset",
 80                "note": "",
 81            }
 82        }
 83    )
 84
 85    @model_validator(mode="after")
 86    def validate_expiry(self):
 87        if self.expire_strategy == UserExpireStrategy.START_ON_FIRST_USE:
 88            if not self.usage_duration:
 89                raise ValueError(
 90                    "User expire_strategy can not be start_on_first_use without a valid usage_duration."
 91                )
 92        elif (
 93            self.expire_strategy == UserExpireStrategy.FIXED_DATE
 94            and not self.expire_date
 95        ):
 96            raise ValueError(
 97                "User expire_strategy can not be fixed_date without a valid expire date."
 98            )
 99        return self
100
101
102class UserModify(UserCreate):
103    service_ids: list[int] | None = Field(None)
104    data_limit_reset_strategy: UserDataUsageResetStrategy | None = Field(None)
105    model_config = ConfigDict(
106        json_schema_extra={
107            "example": {
108                "username": "mammad1234",
109                "service_ids": [1, 2, 3],
110                "expire_strategy": "fixed_date",
111                "expire_date": "2023-11-03T20:30:00",
112                "data_limit": 0,
113                "data_limit_reset_strategy": "no_reset",
114                "note": "",
115            }
116        }
117    )
118
119
120class UserResponse(User):
121    id: int
122    activated: bool
123    is_active: bool
124    expired: bool
125    data_limit_reached: bool
126    enabled: bool
127    used_traffic: int
128    lifetime_used_traffic: int
129    created_at: datetime
130    service_ids: list[int]
131
132    model_config = ConfigDict(from_attributes=True)
133
134    @computed_field
135    @property
136    def subscription_url(self) -> str:
137        salt = secrets.token_hex(8)
138        url_prefix = SUBSCRIPTION_URL_PREFIX.replace("*", salt)
139        return f"{url_prefix}/sub/{self.username}/{self.key}"
140
141
142class UserUsageResponse(BaseModel):
143    node_id: int | None = None
144    node_name: str
145    used_traffic: int
146
147
148class UserUsagesResponse(BaseModel):
149    username: str
150    usages: list[UserUsageResponse]