From f902efaf882ceb04968cfecbb80de0ac8d3d5427 Mon Sep 17 00:00:00 2001 From: Hirad Date: Fri, 18 Jul 2025 11:19:36 +0330 Subject: [PATCH] change user to usermedia and add new user --- src/synclean/api/users.py | 24 +++++- src/synclean/models/enums.py | 16 +++- src/synclean/models/pagination.py | 10 ++- src/synclean/models/user.py | 117 ++++++++++++++++++++++++------ src/synclean/models/user_media.py | 42 +++++++++++ 5 files changed, 178 insertions(+), 31 deletions(-) create mode 100644 src/synclean/models/user_media.py diff --git a/src/synclean/api/users.py b/src/synclean/api/users.py index c04495c..d5dc6de 100644 --- a/src/synclean/api/users.py +++ b/src/synclean/api/users.py @@ -1,19 +1,35 @@ from synclean.api.synapse import SynapseApiClient -from synclean.models.pagination import UserPaginationParams -from synclean.models.user import UserMediaList +from synclean.models.pagination import UserMediaPaginationParams, UserPaginationParams +from synclean.models.user import User, UserList +from synclean.models.user_media import UserMediaList class UserAPI: def __init__(self, client: SynapseApiClient): 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: - pagination = UserPaginationParams() + pagination = UserMediaPaginationParams() params = pagination.to_query_params() response = self.client.request("GET", "/v1/statistics/users/media", params) 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) + + + diff --git a/src/synclean/models/enums.py b/src/synclean/models/enums.py index 5b22a77..73b46b8 100644 --- a/src/synclean/models/enums.py +++ b/src/synclean/models/enums.py @@ -16,13 +16,25 @@ class RoomOrderBy(Enum): HISTORY_VISIBILITY = "history_visibility" STATE_EVENTS = "state_events" -class UserOrderBy(Enum): - """Available user ordering options.""" +class UserMediaOrderBy(Enum): + """Available user ordering options for media.""" USER_ID = "user_id" DISPLAY_NAME = "display_name" MEDIA_LENGTH = "media_length" 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): """Sort direction.""" diff --git a/src/synclean/models/pagination.py b/src/synclean/models/pagination.py index ecf73c8..4dc9c8b 100644 --- a/src/synclean/models/pagination.py +++ b/src/synclean/models/pagination.py @@ -1,6 +1,6 @@ from dataclasses import dataclass 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") @@ -38,4 +38,10 @@ class RoomPaginationParams(PaginationParams[RoomOrderBy]): @dataclass class UserPaginationParams(PaginationParams[UserOrderBy]): """Pagination parameters for users.""" - order_by = UserOrderBy = UserOrderBy.MEDIA_LENGTH \ No newline at end of file + order_by = UserOrderBy = UserOrderBy.NAME + + +@dataclass +class UserMediaPaginationParams(PaginationParams[UserMediaOrderBy]): + """Pagination parameters for users based on media.""" + order_by = UserOrderBy = UserMediaOrderBy.MEDIA_LENGTH \ No newline at end of file diff --git a/src/synclean/models/user.py b/src/synclean/models/user.py index 90d7dbf..6dba5e6 100644 --- a/src/synclean/models/user.py +++ b/src/synclean/models/user.py @@ -1,42 +1,113 @@ from dataclasses import dataclass -from typing import List, Optional, Any, Dict +from datetime import datetime +from typing import Optional, Dict, Any, List @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 - 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) + avatar_url: Optional[str] + creation_ts: int + approved: bool + erased: bool + last_seen_ts: Optional[int] + locked: bool @classmethod - def from_api_response(cls, data: Dict[str, Any]) -> "UserMedia": - """Create UserMedia from API response.""" + def from_api_response(cls, data: Dict[str, Any]) -> "User": + """Create a User instance from json""" 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"), - media_count=data.get("media_count"), - media_length=data.get("media_length"), - user_id=data.get("user_id") + avatar_url=data.get("avatar_url"), + creation_ts=data.get("creation_ts"), + 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 -class UserMediaList: - users: List[UserMedia] +class UserList: + users: List[User] + next_token: Optional[str] 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 - 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", [])] + def from_api_response(cls, data: Dict[str, Any]) -> "UserList": + """Create a UserList instance from API response""" + users = [User.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") + next_token=data.get("next_token"), + total=data.get("total", 0) ) \ No newline at end of file diff --git a/src/synclean/models/user_media.py b/src/synclean/models/user_media.py new file mode 100644 index 0000000..90d7dbf --- /dev/null +++ b/src/synclean/models/user_media.py @@ -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") + ) \ No newline at end of file