change user to usermedia and add new user

This commit is contained in:
Hirad 2025-07-18 11:19:36 +03:30
parent abafe60058
commit f902efaf88
5 changed files with 178 additions and 31 deletions

View file

@ -1,19 +1,35 @@
from synclean.api.synapse import SynapseApiClient from synclean.api.synapse import SynapseApiClient
from synclean.models.pagination import UserPaginationParams from synclean.models.pagination import UserMediaPaginationParams, UserPaginationParams
from synclean.models.user import UserMediaList from synclean.models.user import User, UserList
from synclean.models.user_media import UserMediaList
class UserAPI: class UserAPI:
def __init__(self, client: SynapseApiClient): def __init__(self, client: SynapseApiClient):
self.client = client self.client = client
def get_users_list_by_media(self, pagination: UserPaginationParams = None) -> UserMediaList: def get_users_list_by_media(self, pagination: UserMediaPaginationParams = None) -> UserMediaList:
if pagination is None: if pagination is None:
pagination = UserPaginationParams() pagination = UserMediaPaginationParams()
params = pagination.to_query_params() params = pagination.to_query_params()
response = self.client.request("GET", "/v1/statistics/users/media", params) response = self.client.request("GET", "/v1/statistics/users/media", params)
return UserMediaList.from_api_response(response) return UserMediaList.from_api_response(response)
def get_users_list(self, pagination: UserPaginationParams) -> UserList:
if pagination is None:
pagination = UserPaginationParams()
params = pagination.to_query_params()
response = self.client.request("GET", "/v2/users", params)
return UserList.from_api_response(response)
def get_user_details_by_name(self, username: str) -> User:
response = self.client.request("GET", f"/v2/users/{username}")
return User.from_api_response(response)

View file

@ -16,13 +16,25 @@ class RoomOrderBy(Enum):
HISTORY_VISIBILITY = "history_visibility" HISTORY_VISIBILITY = "history_visibility"
STATE_EVENTS = "state_events" STATE_EVENTS = "state_events"
class UserOrderBy(Enum): class UserMediaOrderBy(Enum):
"""Available user ordering options.""" """Available user ordering options for media."""
USER_ID = "user_id" USER_ID = "user_id"
DISPLAY_NAME = "display_name" DISPLAY_NAME = "display_name"
MEDIA_LENGTH = "media_length" MEDIA_LENGTH = "media_length"
MEDIA_COUNT = "media_count" MEDIA_COUNT = "media_count"
class UserOrderBy(Enum):
"""Available user ordering options."""
NAME = "name"
# IS_GUEST = "is_guest"
# ADMIN = "admin"
# USER_TYPE = "user_type"
# DEACTIVATED = "deactivated"
# SHADOW_BANNED = "shadow_banned"
# DISPLAY_NAME = "display_name"
# AVATAR_URL = "avatar_url"
CREATION_TS = "creation_ts"
# LAST_SEEN_TS = "last_seen_ts"
class Direction(Enum): class Direction(Enum):
"""Sort direction.""" """Sort direction."""

View file

@ -1,6 +1,6 @@
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional, Dict, Any, TypeVar, Generic from typing import Optional, Dict, Any, TypeVar, Generic
from synclean.models.enums import RoomOrderBy, Direction, UserOrderBy from synclean.models.enums import RoomOrderBy, Direction, UserMediaOrderBy, UserOrderBy
OrderByType = TypeVar("OrderByType") OrderByType = TypeVar("OrderByType")
@ -38,4 +38,10 @@ class RoomPaginationParams(PaginationParams[RoomOrderBy]):
@dataclass @dataclass
class UserPaginationParams(PaginationParams[UserOrderBy]): class UserPaginationParams(PaginationParams[UserOrderBy]):
"""Pagination parameters for users.""" """Pagination parameters for users."""
order_by = UserOrderBy = UserOrderBy.MEDIA_LENGTH order_by = UserOrderBy = UserOrderBy.NAME
@dataclass
class UserMediaPaginationParams(PaginationParams[UserMediaOrderBy]):
"""Pagination parameters for users based on media."""
order_by = UserOrderBy = UserMediaOrderBy.MEDIA_LENGTH

View file

@ -1,42 +1,113 @@
from dataclasses import dataclass from dataclasses import dataclass
from typing import List, Optional, Any, Dict from datetime import datetime
from typing import Optional, Dict, Any, List
@dataclass @dataclass
class UserMedia: class User:
name: str
user_type: Optional[str]
is_guest: bool
admin: Optional[bool]
deactivated: bool
shadow_banned: bool
display_name: str display_name: str
media_count: int avatar_url: Optional[str]
media_length: int creation_ts: int
user_id: str approved: bool
erased: bool
def media_length_mb(self, precision: int = 2) -> float: last_seen_ts: Optional[int]
"""Convert media length to MB.""" locked: bool
mb = self.media_length / (1024 * 1024)
return round(mb, precision)
@classmethod @classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "UserMedia": def from_api_response(cls, data: Dict[str, Any]) -> "User":
"""Create UserMedia from API response.""" """Create a User instance from json"""
return cls( return cls(
name=data.get("name"),
user_type=data.get("user_type"),
is_guest=data.get("is_guest"),
admin=data.get("admin"),
deactivated=data.get("deactivated"),
shadow_banned=data.get("shadow_banned"),
display_name=data.get("displayname"), display_name=data.get("displayname"),
media_count=data.get("media_count"), avatar_url=data.get("avatar_url"),
media_length=data.get("media_length"), creation_ts=data.get("creation_ts"),
user_id=data.get("user_id") approved=data.get("approved"),
erased=data.get("erased"),
last_seen_ts=data.get("last_seen_ts"),
locked=data.get("locked"),
) )
@property
def creation_datetime(self) -> datetime:
"""Get the creation timestamp as a datetime object"""
return datetime.fromtimestamp(self.creation_ts / 1000)
@property
def last_seen_datetime(self) -> Optional[datetime]:
"""Get the last seen timestamp as a datetime object"""
if self.last_seen_ts is None:
return None
return datetime.fromtimestamp(self.last_seen_ts / 1000)
@property
def is_active(self) -> bool:
"""Check if the user is active"""
return not (self.deactivated or self.erased or self.shadow_banned or self.locked)
@property
def has_avatar(self) -> bool:
"""Check if the user has an avatar"""
return self.avatar_url is not None and self.avatar_url.strip() != ""
def __str__(self) -> str:
"""String representation of the User class"""
return f"User(name={self.name!r}, displayname={self.display_name!r})"
@dataclass @dataclass
class UserMediaList: class UserList:
users: List[UserMedia] users: List[User]
next_token: Optional[str]
total: int total: int
next_token: Optional[int] = None
def __post_init__(self) -> None:
"""Validate response data after initialization"""
if self.total < 0:
raise ValueError("Total count cannot be negative")
@property
def users_count(self) -> int:
"""Get the number of users in the current response"""
return len(self.users)
@property
def has_more_users(self) -> bool:
"""Check if there are more users to fetch"""
return self.next_token is not None
@property
def active_users(self) -> List[User]:
"""Get a list of active users"""
return [user for user in self.users if user.is_active]
@property
def admin_users(self) -> List[User]:
"""Get a list of admin users"""
return [user for user in self.users if user.admin]
@property
def guest_users(self) -> List[User]:
"""Get a list of guest users"""
return [user for user in self.users if user.is_guest]
@classmethod @classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "UserMediaList": def from_api_response(cls, data: Dict[str, Any]) -> "UserList":
"""Create UserMediaList from API response.""" """Create a UserList instance from API response"""
users = [UserMedia.from_api_response(user_data) for user_data in data.get("users", [])] users = [User.from_api_response(user_data) for user_data in data.get("users", [])]
return cls( return cls(
users=users, users=users,
total=data.get("total", 0), next_token=data.get("next_token"),
next_token=data.get("next_token") total=data.get("total", 0)
) )

View file

@ -0,0 +1,42 @@
from dataclasses import dataclass
from typing import List, Optional, Any, Dict
@dataclass
class UserMedia:
display_name: str
media_count: int
media_length: int
user_id: str
def media_length_mb(self, precision: int = 2) -> float:
"""Convert media length to MB."""
mb = self.media_length / (1024 * 1024)
return round(mb, precision)
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "UserMedia":
"""Create UserMedia from API response."""
return cls(
display_name=data.get("displayname"),
media_count=data.get("media_count"),
media_length=data.get("media_length"),
user_id=data.get("user_id")
)
@dataclass
class UserMediaList:
users: List[UserMedia]
total: int
next_token: Optional[int] = None
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "UserMediaList":
"""Create UserMediaList from API response."""
users = [UserMedia.from_api_response(user_data) for user_data in data.get("users", [])]
return cls(
users=users,
total=data.get("total", 0),
next_token=data.get("next_token")
)