117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
from typing import Optional, List
|
|
|
|
from synclean.api.users import UserAPI
|
|
from synclean.models.enums import UserMediaStatsOrderBy, Direction, UserOrderBy, UserMediaOrderBy
|
|
from synclean.models.pagination import UserMediaStatsPaginationParams, UserPaginationParams, UserMediaPaginationParams
|
|
from synclean.models.user import UserList, User
|
|
from synclean.models.user_media import UserMediaList, MediaItem
|
|
from synclean.models.user_media_stats import UserMediaStatsList, UserMediaStats
|
|
|
|
|
|
class UserService:
|
|
def __init__(self, user_api: UserAPI):
|
|
self.user_api = user_api
|
|
|
|
def get_users_list_by_media(
|
|
self,
|
|
limit: int,
|
|
offset: int,
|
|
order_by: str,
|
|
direction: str,
|
|
search: Optional[str]
|
|
) -> UserMediaStatsList | None:
|
|
order_by_enum = UserMediaStatsOrderBy(order_by)
|
|
direction_enum = Direction(direction)
|
|
|
|
pagination_params = UserMediaStatsPaginationParams(limit, offset, order_by_enum, direction_enum, search)
|
|
|
|
users = self.user_api.get_users_list_by_media(pagination_params)
|
|
if users:
|
|
return users
|
|
return None
|
|
|
|
def get_users_list(
|
|
self,
|
|
limit: int,
|
|
offset: int,
|
|
order_by: str,
|
|
direction: str,
|
|
search: Optional[str]
|
|
) -> UserList | None:
|
|
order_by_enum = UserOrderBy(order_by)
|
|
direction_enum = Direction(direction)
|
|
|
|
pagination_params = UserPaginationParams(limit, offset, order_by_enum, direction_enum, search)
|
|
users = self.user_api.get_users_list(pagination_params)
|
|
if users:
|
|
return users
|
|
return None
|
|
|
|
def get_user_details_by_id(self, user_id: str) -> User | None:
|
|
user_details = self.user_api.get_user_details_by_name(user_id)
|
|
if user_details:
|
|
return user_details
|
|
|
|
print("No user found")
|
|
return None
|
|
|
|
def get_user_media_stats_by_id(self, user_id: str) -> UserMediaStats | None:
|
|
if user_id.startswith('@'):
|
|
user_id = user_id[1:]
|
|
if ':' in user_id:
|
|
user_id = user_id.split(':')[0]
|
|
|
|
user_media = self.user_api.get_user_media_stats_by_id(user_id)
|
|
if user_media:
|
|
return user_media
|
|
return None
|
|
|
|
def get_user_media_by_id(
|
|
self,
|
|
user_id: str,
|
|
limit: int,
|
|
offset: int,
|
|
order_by: str,
|
|
direction: str
|
|
) -> UserMediaList | None:
|
|
order_by_enum = UserMediaOrderBy(order_by)
|
|
direction_enum = Direction(direction)
|
|
|
|
pagination = UserMediaPaginationParams(limit, offset, order_by_enum, direction_enum)
|
|
|
|
user_media = self.user_api.get_media_by_user_id(user_id, pagination)
|
|
if user_media:
|
|
return user_media
|
|
return None
|
|
|
|
def get_user_media_by_type(
|
|
self,
|
|
user_id: str,
|
|
limit: int,
|
|
offset: int,
|
|
order_by: str,
|
|
direction: str,
|
|
media_type: str
|
|
) -> dict[str, List[MediaItem]] | None:
|
|
order_by_enum = UserMediaOrderBy(order_by)
|
|
direction_enum = Direction(direction)
|
|
|
|
pagination = UserMediaPaginationParams(limit, offset, order_by_enum, direction_enum)
|
|
|
|
user_media = self.user_api.get_all_user_media(user_id, pagination)
|
|
if user_media:
|
|
filtered_media = [media for media in user_media if media.media_type.startswith(media_type)]
|
|
if filtered_media:
|
|
total_size = sum(media.media_length for media in filtered_media)
|
|
media_by_type = {
|
|
"media": filtered_media,
|
|
"total_size": total_size,
|
|
"total_count": len(filtered_media)
|
|
}
|
|
return media_by_type
|
|
else:
|
|
return None
|
|
return None
|
|
|
|
def delete_user_media(self, avatar: str):
|
|
pass
|