itlxrd/itd-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add models part 3

Browse Source
This commit is contained in:
firedotguy
2026-02-01 17:20:37 +03:00
parent 2a9f7da9a9
commit ba78457de5
13 changed files with 267 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
itd/client.py
View File

@@ -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']

12
itd/exceptions.py
View File

@@ -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'
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'

5
itd/models/_text.py
View File

@@ -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')

4
itd/models/comment.py
View File

@@ -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

9
itd/models/file.py
View File

@@ -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

2
itd/models/notification.py
View File

@@ -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

7
itd/models/pagination.py
View File

@@ -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')
has_more: bool = Field(True, alias='hasMore')
next_cursor: UUID | None = Field(None, alias='nextCursor')

30
itd/models/post.py
View File

@@ -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

17
itd/models/user.py
View File

@@ -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')

21
itd/request.py
View File

@@ -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

3
itd/routes/hashtags.py
View File

@@ -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})

15
itd/routes/notifications.py
View File

@@ -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')

23
itd/routes/posts.py
View File

@@ -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
# todo post restore
# todo post like