diff --git a/itd/client.py b/itd/client.py index f2d4414..714f5cd 100644 --- a/itd/client.py +++ b/itd/client.py @@ -2,7 +2,7 @@ from uuid import UUID from _io import BufferedReader from typing import cast -from requests.exceptions import HTTPError +from requests.exceptions import HTTPError, ConnectionError from itd.routes.users import get_user, update_profile, follow, unfollow, get_followers, get_following, update_privacy from itd.routes.etc import get_top_clans, get_who_to_follow, get_platform_status @@ -17,13 +17,16 @@ from itd.routes.auth import refresh_token, change_password, logout from itd.routes.verification import verificate, get_verification_status from itd.models.comment import Comment +from itd.models.notification import Notification +from itd.models.post import Post, NewPost from itd.models.clan import Clan +from itd.models.hashtag import Hashtag from itd.models.user import User, UserProfileUpdate, UserPrivacy, UserFollower, UserWhoToFollow from itd.models.pagination import Pagination from itd.models.verification import Verification, VerificationStatus from itd.request import set_cookies -from itd.exceptions import NoCookie, NoAuthData, SamePassword, InvalidOldPassword, NotFound, ValidationError, UserBanned, PendingRequestExists +from itd.exceptions import NoCookie, NoAuthData, SamePassword, InvalidOldPassword, NotFound, ValidationError, UserBanned, PendingRequestExists, Forbidden, UsernameTaken def refresh_on_error(func): @@ -35,6 +38,9 @@ def refresh_on_error(func): self.refresh_auth() return func(self, *args, **kwargs) raise e + except ConnectionError: + self.refresh_auth() + return func(self, *args, **kwargs) return wrapper @@ -166,6 +172,8 @@ class Client: res = update_profile(self.token, bio, display_name, username, banner_id) if res.status_code == 422 and 'found' in res.json(): raise ValidationError(*list(res.json()['found'].items())[0]) + if res.json().get('error', {}).get('code') == 'USERNAME_TAKEN': + raise UsernameTaken() res.raise_for_status() return UserProfileUpdate.model_validate(res.json()) @@ -354,6 +362,7 @@ class Client: Raises: ValidationError: Ошибка валидации + NotFound: Пост не найден Returns: Comment: Комментарий @@ -379,6 +388,7 @@ class Client: Raises: ValidationError: Ошибка валидации + NotFound: Пользователь или комментарий не найден Returns: Comment: Комментарий @@ -397,56 +407,191 @@ class Client: @refresh_on_error def get_comments(self, post_id: UUID, limit: int = 20, cursor: int = 0, sort: str = 'popular') -> tuple[list[Comment], Pagination]: + """Получить список комментариев + + Args: + post_id (UUID): UUID поста + limit (int, optional): Лимит. Defaults to 20. + cursor (int, optional): Курсор (сколько пропустить). Defaults to 0. + sort (str, optional): Сортировка. Defaults to 'popular'. + + Raises: + NotFound: Пост не найден + + Returns: + list[Comment]: Список комментариев + Pagination: Пагинация + """ res = get_comments(self.token, post_id, limit, cursor, sort) if res.json().get('error', {}).get('code') == 'NOT_FOUND': raise NotFound('Post') res.raise_for_status() data = res.json()['data'] - return [Comment.model_validate(comment) for comment in data['comments']], Pagination(page=(cursor // limit) or 1, limit=limit, total=data['total'], hasMore=data['hasMore']) + return [Comment.model_validate(comment) for comment in data['comments']], Pagination(page=(cursor // limit) or 1, limit=limit, total=data['total'], hasMore=data['hasMore'], nextCursor=None) @refresh_on_error - def like_comment(self, id: UUID): - return like_comment(self.token, id) + def like_comment(self, id: UUID) -> int: + """Лайкнуть комментарий + + Args: + id (UUID): UUID комментария + + Raises: + NotFound: Комментарий не найден + + Returns: + int: Количество лайков + """ + res = like_comment(self.token, id) + if res.json().get('error', {}).get('code') == 'NOT_FOUND': + raise NotFound('Comment') + res.raise_for_status() + + return res.json()['likesCount'] @refresh_on_error - def unlike_comment(self, id: UUID): - return unlike_comment(self.token, id) + def unlike_comment(self, id: UUID) -> int: + """Убрать лайк с комментария + + Args: + id (UUID): UUID комментария + + Raises: + NotFound: Комментарий не найден + + Returns: + int: Количество лайков + """ + res = unlike_comment(self.token, id) + if res.json().get('error', {}).get('code') == 'NOT_FOUND': + raise NotFound('Comment') + res.raise_for_status() + + return res.json()['likesCount'] @refresh_on_error - def delete_comment(self, id: UUID): - return delete_comment(self.token, id) + def delete_comment(self, id: UUID) -> None: + """Удалить комментарий + + Args: + id (UUID): UUID комментария + + Raises: + NotFound: Комментарий не найден + Forbidden: Нет прав на удаление + """ + res = delete_comment(self.token, id) + if res.status_code == 204: + return + if res.json().get('error', {}).get('code') == 'NOT_FOUND': + raise NotFound('Comment') + if res.json().get('error', {}).get('code') == 'FORBIDDEN': + raise Forbidden('delete comment') + res.raise_for_status() @refresh_on_error - def get_hastags(self, limit: int = 10): - return get_hastags(self.token, limit) + def get_hastags(self, limit: int = 10) -> list[Hashtag]: + """Получить список популярных хэштэгов + + Args: + limit (int, optional): Лимит. Defaults to 10. + + Returns: + list[Hashtag]: Список хэштэгов + """ + res = get_hastags(self.token, limit) + res.raise_for_status() + + return [Hashtag.model_validate(hashtag) for hashtag in res.json()['data']['hashtags']] @refresh_on_error - def get_posts_by_hashtag(self, hashtag: str, limit: int = 20, cursor: int = 0): - return get_posts_by_hastag(self.token, hashtag, limit, cursor) + def get_posts_by_hashtag(self, hashtag: str, limit: int = 20, cursor: UUID | None = None) -> tuple[Hashtag | None, list[Post], Pagination]: + """Получить посты по хэштэгу + + Args: + hashtag (str): Хэштэг (без #) + limit (int, optional): Лимит. Defaults to 20. + cursor (UUID | None, optional): Курсор (UUID последнего поста, после которого брать данные). Defaults to None. + + Returns: + Hashtag | None: Хэштэг + list[Post]: Посты + Pagination: Пагинация + """ + res = get_posts_by_hastag(self.token, hashtag, limit, cursor) + res.raise_for_status() + data = res.json()['data'] + + return Hashtag.model_validate(data['hashtag']), [Post.model_validate(post) for post in data['posts']], Pagination.model_validate(data['pagination']) @refresh_on_error - def get_notifications(self, limit: int = 20, cursor: int = 0, type: str | None = None): - return get_notifications(self.token, limit, cursor, type) + def get_notifications(self, limit: int = 20, offset: int = 0) -> tuple[list[Notification], Pagination]: + """Получить уведомления + + Args: + limit (int, optional): Лимит. Defaults to 20. + offset (int, optional): Сдвиг. Defaults to 0. + + Returns: + list[Notification]: Уведомления + Pagination: Пагинация + """ + res = get_notifications(self.token, limit, offset) + res.raise_for_status() + + return ( + [Notification.model_validate(notification) for notification in res.json()['notifications']], + Pagination(page=(offset // limit) + 1, limit=limit, hasMore=res.json()['hasMore'], nextCursor=None) + ) @refresh_on_error - def mark_as_read(self, id: str): - return mark_as_read(self.token, id) + def mark_as_read(self, id: UUID) -> bool: + """Прочитать уведомление + + Args: + id (UUID): UUID уведомления + + Returns: + bool: Успешно (False - уже прочитано) + """ + res = mark_as_read(self.token, id) + res.raise_for_status() + + return res.json()['success'] @refresh_on_error - def mark_all_as_read(self): - return mark_all_as_read(self.token) - - @refresh_on_error - def get_unread_notifications_count(self): - return get_unread_notifications_count(self.token) + def mark_all_as_read(self) -> None: + """Прочитать все уведомления""" + res = mark_all_as_read(self.token) + res.raise_for_status() @refresh_on_error - def create_post(self, content: str, wall_recipient_id: int | None = None, attach_ids: list[str] = []): - return create_post(self.token, content, wall_recipient_id, attach_ids) + def get_unread_notifications_count(self) -> int: + """Получить количество непрочитанных уведомлений + + Returns: + int: Количество + """ + res = get_unread_notifications_count(self.token) + res.raise_for_status() + + return res.json()['count'] + + + @refresh_on_error + def create_post(self, content: str, wall_recipient_id: UUID | None = None, attach_ids: list[UUID] = []) -> NewPost: + res = create_post(self.token, content, wall_recipient_id, attach_ids) + if res.json().get('error', {}).get('code') == 'NOT_FOUND': + raise NotFound('Wall recipient') + if res.status_code == 422 and 'found' in res.json(): + raise ValidationError(*list(res.json()['found'].items())[0]) + res.raise_for_status() + + return NewPost.model_validate(res.json()) @refresh_on_error def get_posts(self, username: str | None = None, limit: int = 20, cursor: int = 0, sort: str = '', tab: str = ''): @@ -513,7 +658,7 @@ class Client: @refresh_on_error def upload_file(self, name: str, data: BufferedReader): - return upload_file(self.token, name, data) + return upload_file(self.token, name, data).json() def update_banner(self, name: str) -> UserProfileUpdate: id = self.upload_file(name, cast(BufferedReader, open(name, 'rb')))['id'] diff --git a/itd/exceptions.py b/itd/exceptions.py index 68c4d8b..f1cc39f 100644 --- a/itd/exceptions.py +++ b/itd/exceptions.py @@ -48,4 +48,14 @@ class RateLimitExceeded(Exception): def __init__(self, retry_after: int): self.retry_after = retry_after def __str__(self): - return f'Rate limit exceeded - too much requests. Retry after {self.retry_after} seconds' \ No newline at end of file + return f'Rate limit exceeded - too much requests. Retry after {self.retry_after} seconds' + +class Forbidden(Exception): + def __init__(self, action: str): + self.action = action + def __str__(self): + return f'Forbidden to {self.action}' + +class UsernameTaken(Exception): + def __str__(self): + return 'Username is already taken' \ No newline at end of file diff --git a/itd/models/_text.py b/itd/models/_text.py index 49238a7..96aa99e 100644 --- a/itd/models/_text.py +++ b/itd/models/_text.py @@ -3,15 +3,10 @@ from datetime import datetime from pydantic import BaseModel, Field, field_validator -from itd.models.user import UserPost -from itd.models.file import Attach - class TextObject(BaseModel): id: UUID content: str - author: UserPost - attachments: list[Attach] = [] created_at: datetime = Field(alias='createdAt') diff --git a/itd/models/comment.py b/itd/models/comment.py index efcd499..1330225 100644 --- a/itd/models/comment.py +++ b/itd/models/comment.py @@ -2,12 +2,16 @@ from pydantic import Field from itd.models._text import TextObject from itd.models.user import UserPost +from itd.models.file import Attach class Comment(TextObject): + author: UserPost + likes_count: int = Field(0, alias='likesCount') replies_count: int = Field(0, alias='repliesCount') is_liked: bool = Field(False, alias='isLiked') + attachments: list[Attach] = [] replies: list['Comment'] = [] reply_to: UserPost | None = None # author of replied comment, if this comment is reply \ No newline at end of file diff --git a/itd/models/file.py b/itd/models/file.py index b732b7e..f7f34e7 100644 --- a/itd/models/file.py +++ b/itd/models/file.py @@ -12,15 +12,18 @@ class File(BaseModel): size: int -class Attach(BaseModel): +class PostAttach(BaseModel): id: UUID type: AttachType = AttachType.IMAGE url: str thumbnail_url: str | None = Field(None, alias='thumbnailUrl') + width: int | None = None + height: int | None = None + + +class Attach(PostAttach): filename: str mime_type: str = Field(alias='mimeType') size: int - width: int | None = None - height: int | None = None duration: int | None = None order: int = 0 \ No newline at end of file diff --git a/itd/models/notification.py b/itd/models/notification.py index 7ea3ea9..4dca9fe 100644 --- a/itd/models/notification.py +++ b/itd/models/notification.py @@ -11,7 +11,7 @@ class Notification(BaseModel): type: NotificationType target_type: NotificationTargetType | None = Field(None, alias='targetType') # none - follows, other - NotificationTragetType.POST - target_id: int | None = Field(None, alias='targetId') # none - follows + target_id: UUID | None = Field(None, alias='targetId') # none - follows preview: str | None = None # follow - none, comment/reply - content, repost - original post content, like - post content, wall_post - wall post content diff --git a/itd/models/pagination.py b/itd/models/pagination.py index df81f56..e140b8b 100644 --- a/itd/models/pagination.py +++ b/itd/models/pagination.py @@ -1,7 +1,10 @@ +from uuid import UUID + from pydantic import BaseModel, Field class Pagination(BaseModel): - page: int = 1 + page: int | None = 1 limit: int = 20 total: int | None = None - has_more: bool = Field(True, alias='hasMore') \ No newline at end of file + has_more: bool = Field(True, alias='hasMore') + next_cursor: UUID | None = Field(None, alias='nextCursor') \ No newline at end of file diff --git a/itd/models/post.py b/itd/models/post.py index dfe33dd..f07c36c 100644 --- a/itd/models/post.py +++ b/itd/models/post.py @@ -1,29 +1,47 @@ +from uuid import UUID + from pydantic import Field -from itd.models.user import UserPost +from itd.models.user import UserPost, UserNewPost from itd.models._text import TextObject +from itd.models.file import PostAttach +from itd.models.comment import Comment -class PostShort(TextObject): +class _PostShort(TextObject): likes_count: int = Field(0, alias='likesCount') comments_count: int = Field(0, alias='commentsCount') reposts_count: int = Field(0, alias='repostsCount') views_count: int = Field(0, alias='viewsCount') +class PostShort(_PostShort): + author: UserPost + + class OriginalPost(PostShort): is_deleted: bool = Field(False, alias='isDeleted') -class Post(PostShort): +class _Post(_PostShort): is_liked: bool = Field(False, alias='isLiked') is_reposted: bool = Field(False, alias='isReposted') is_viewed: bool = Field(False, alias='isViewed') is_owner: bool = Field(False, alias='isOwner') - comments: list = [] + attachments: list[PostAttach] = [] + comments: list[Comment] = [] original_post: OriginalPost | None = None - wall_recipient_id: int | None = None - wall_recipient: UserPost | None = None + wall_recipient_id: UUID | None = Field(None, alias='wallRecipientId') + wall_recipient: UserPost | None = Field(None, alias='wallRecipient') + + +class Post(_Post, PostShort): + pass + + +class NewPost(_Post): + author: UserNewPost + diff --git a/itd/models/user.py b/itd/models/user.py index 0c717ae..a35ff30 100644 --- a/itd/models/user.py +++ b/itd/models/user.py @@ -19,22 +19,23 @@ class UserProfileUpdate(BaseModel): updated_at: datetime | None = Field(None, alias='updatedAt') - model_config = {'populate_by_name': True} - -class UserNotification(BaseModel): - id: UUID +class UserNewPost(BaseModel): username: str | None = None display_name: str = Field(alias='displayName') avatar: str - model_config = {'populate_by_name': True} - - -class UserPost(UserNotification): verified: bool = False +class UserNotification(UserNewPost): + id: UUID + + +class UserPost(UserNotification, UserNewPost): + pass + + class UserWhoToFollow(UserPost): followers_count: int = Field(0, alias='followersCount') diff --git a/itd/request.py b/itd/request.py index a63cf10..909ed31 100644 --- a/itd/request.py +++ b/itd/request.py @@ -1,6 +1,7 @@ from _io import BufferedReader from requests import Session +from requests.exceptions import JSONDecodeError from itd.exceptions import InvalidToken, InvalidCookie, RateLimitExceeded @@ -29,10 +30,13 @@ def fetch(token: str, method: str, url: str, params: dict = {}, files: dict[str, if method == "get": res = s.get(base, timeout=120 if files else 20, params=params, headers=headers) else: - res = s.request(method.upper(), base, timeout=20, json=params, headers=headers, files=files) + res = s.request(method.upper(), base, timeout=120 if files else 20, json=params, headers=headers, files=files) - if res.json().get('error', {}).get('code') == 'RATE_LIMIT_EXCEEDED': - raise RateLimitExceeded(res.json()['error'].get('retryAfter', 0)) + try: + if res.json().get('error', {}).get('code') == 'RATE_LIMIT_EXCEEDED': + raise RateLimitExceeded(res.json()['error'].get('retryAfter', 0)) + except JSONDecodeError: + pass print(res.text) return res @@ -75,9 +79,12 @@ def auth_fetch(cookies: str, method: str, url: str, params: dict = {}, token: st # print(res.text) if res.text == 'UNAUTHORIZED': raise InvalidToken() - if res.json().get('error', {}).get('code') == 'RATE_LIMIT_EXCEEDED': - raise RateLimitExceeded(res.json()['error'].get('retryAfter', 0)) - if res.json().get('error', {}).get('code') in ('SESSION_NOT_FOUND', 'REFRESH_TOKEN_MISSING'): - raise InvalidCookie() + try: + if res.json().get('error', {}).get('code') == 'RATE_LIMIT_EXCEEDED': + raise RateLimitExceeded(res.json()['error'].get('retryAfter', 0)) + if res.json().get('error', {}).get('code') in ('SESSION_NOT_FOUND', 'REFRESH_TOKEN_MISSING'): + raise InvalidCookie() + except JSONDecodeError: + pass return res diff --git a/itd/routes/hashtags.py b/itd/routes/hashtags.py index e6e9a63..72404c4 100644 --- a/itd/routes/hashtags.py +++ b/itd/routes/hashtags.py @@ -1,7 +1,8 @@ +from uuid import UUID from itd.request import fetch def get_hastags(token: str, limit: int = 10): return fetch(token, 'get', 'hashtags/trending', {'limit': limit}) -def get_posts_by_hastag(token: str, hashtag: str, limit: int = 20, cursor: int = 0): +def get_posts_by_hastag(token: str, hashtag: str, limit: int = 20, cursor: UUID | None = None): return fetch(token, 'get', f'hashtags/{hashtag}/posts', {'limit': limit, 'cursor': cursor}) diff --git a/itd/routes/notifications.py b/itd/routes/notifications.py index 8ad6759..de0a741 100644 --- a/itd/routes/notifications.py +++ b/itd/routes/notifications.py @@ -1,16 +1,15 @@ +from uuid import UUID + from itd.request import fetch -def get_notifications(token: str, limit: int = 20, cursor: int = 0, type: str | None = None): - data = {'limit': str(limit), 'cursor': str(cursor)} - if type: - data['type'] = type - return fetch(token, 'get', 'notifications', data) +def get_notifications(token: str, limit: int = 20, offset: int = 0): + return fetch(token, 'get', 'notifications', {'limit': limit, 'offset': offset}) -def mark_as_read(token: str, id: str): - return fetch(token, 'post', f'notification/{id}/read') +def mark_as_read(token: str, id: UUID): + return fetch(token, 'post', f'notifications/{id}/read') def mark_all_as_read(token: str): - return fetch(token, 'post', f'notification/read-all') + return fetch(token, 'post', f'notifications/read-all') def get_unread_notifications_count(token: str): return fetch(token, 'get', 'notifications/count') \ No newline at end of file diff --git a/itd/routes/posts.py b/itd/routes/posts.py index b2a0705..97edcd2 100644 --- a/itd/routes/posts.py +++ b/itd/routes/posts.py @@ -1,11 +1,13 @@ +from uuid import UUID + from itd.request import fetch -def create_post(token: str, content: str, wall_recipient_id: int | None = None, attach_ids: list[str] = []): +def create_post(token: str, content: str, wall_recipient_id: UUID | None = None, attach_ids: list[UUID] = []): data: dict = {'content': content} if wall_recipient_id: - data['wallRecipientId'] = wall_recipient_id + data['wallRecipientId'] = str(wall_recipient_id) if attach_ids: - data['attachmentIds'] = attach_ids + data['attachmentIds'] = list(map(str, attach_ids)) return fetch(token, 'post', 'posts', data) @@ -20,28 +22,29 @@ def get_posts(token: str, username: str | None = None, limit: int = 20, cursor: return fetch(token, 'get', 'posts', data) -def get_post(token: str, id: str): +def get_post(token: str, id: UUID): return fetch(token, 'get', f'posts/{id}') -def edit_post(token: str, id: str, content: str): +def edit_post(token: str, id: UUID, content: str): return fetch(token, 'put', f'posts/{id}', {'content': content}) -def delete_post(token: str, id: str): +def delete_post(token: str, id: UUID): return fetch(token, 'delete', f'posts/{id}') -def pin_post(token: str, id: str): +def pin_post(token: str, id: UUID): return fetch(token, 'post', f'posts/{id}/pin') -def repost(token: str, id: str, content: str | None = None): +def repost(token: str, id: UUID, content: str | None = None): data = {} if content: data['content'] = content return fetch(token, 'post', f'posts/{id}/repost', data) -def view_post(token: str, id: str): +def view_post(token: str, id: UUID): return fetch(token, 'post', f'posts/{id}/view') def get_liked_posts(token: str, username: str, limit: int = 20, cursor: int = 0): return fetch(token, 'get', f'posts/user/{username}/liked', {'limit': limit, 'cursor': cursor}) -# todo post restore \ No newline at end of file +# todo post restore +# todo post like \ No newline at end of file