193 lines
6.3 KiB
Python
193 lines
6.3 KiB
Python
from requests.exceptions import HTTPError
|
|
|
|
from itd.users import get_user, update_profile, follow, unfollow, get_followers, get_following
|
|
from itd.etc import get_top_clans, get_who_to_follow, get_platform_status
|
|
from itd.comments import get_comments, add_comment, delete_comment, like_comment, unlike_comment
|
|
from itd.hashtags import get_hastags, get_posts_by_hastag
|
|
from itd.notifications import get_notifications, mark_as_read, mark_all_as_read, get_unread_notifications_count
|
|
from itd.posts import create_post, get_posts, get_post, edit_post, delete_post, pin_post, repost, view_post
|
|
from itd.reports import report
|
|
from itd.search import search
|
|
from itd.request import refresh_auth
|
|
|
|
|
|
def refresh_on_error(func):
|
|
def wrapper(self, *args, **kwargs):
|
|
try:
|
|
return func(self, *args, **kwargs)
|
|
except HTTPError as e:
|
|
if '401' in str(e):
|
|
self.refresh_auth()
|
|
return func(self, *args, **kwargs)
|
|
raise e
|
|
return wrapper
|
|
|
|
|
|
class Client:
|
|
def __init__(self, token: str | None, cookies: str | None = None):
|
|
self.cookies = cookies
|
|
|
|
if token:
|
|
self.token = token.replace('Bearer ', '')
|
|
elif self.cookies:
|
|
self.refresh_auth()
|
|
else:
|
|
raise ValueError('Provide token or cookie')
|
|
|
|
@refresh_on_error
|
|
def refresh_auth(self):
|
|
if self.cookies:
|
|
self.token = refresh_auth(self.cookies)
|
|
else:
|
|
print('no cookies!')
|
|
|
|
@refresh_on_error
|
|
def get_user(self, username: str) -> dict:
|
|
return get_user(self.token, username)
|
|
|
|
@refresh_on_error
|
|
def get_me(self) -> dict:
|
|
return self.get_user('me')
|
|
|
|
@refresh_on_error
|
|
def update_profile(self, username: str | None = None, display_name: str | None = None, bio: str | None = None) -> dict:
|
|
return update_profile(self.token, bio, display_name, username)
|
|
|
|
@refresh_on_error
|
|
def follow(self, username: str) -> dict:
|
|
return follow(self.token, username)
|
|
|
|
@refresh_on_error
|
|
def unfollow(self, username: str) -> dict:
|
|
return unfollow(self.token, username)
|
|
|
|
@refresh_on_error
|
|
def get_followers(self, username: str) -> dict:
|
|
return get_followers(self.token, username)
|
|
|
|
@refresh_on_error
|
|
def get_following(self, username: str) -> dict:
|
|
return get_following(self.token, username)
|
|
|
|
|
|
@refresh_on_error
|
|
def get_who_to_follow(self) -> dict:
|
|
return get_who_to_follow(self.token)
|
|
|
|
@refresh_on_error
|
|
def get_top_clans(self) -> dict:
|
|
return get_top_clans(self.token)
|
|
|
|
@refresh_on_error
|
|
def get_platform_status(self) -> dict:
|
|
return get_platform_status(self.token)
|
|
|
|
|
|
@refresh_on_error
|
|
def add_comment(self, post_id: str, content: str, reply_comment_id: str | None = None):
|
|
return add_comment(self.token, post_id, content, reply_comment_id)
|
|
|
|
@refresh_on_error
|
|
def get_comments(self, post_id: str, limit: int = 20, cursor: int = 0, sort: str = 'popular'):
|
|
return get_comments(self.token, post_id, limit, cursor, sort)
|
|
|
|
@refresh_on_error
|
|
def like_comment(self, id: str):
|
|
return like_comment(self.token, id)
|
|
|
|
@refresh_on_error
|
|
def unlike_comment(self, id: str):
|
|
return unlike_comment(self.token, id)
|
|
|
|
@refresh_on_error
|
|
def delete_comment(self, id: str):
|
|
return delete_comment(self.token, id)
|
|
|
|
|
|
@refresh_on_error
|
|
def get_hastags(self, limit: int = 10):
|
|
return get_hastags(self.token, limit)
|
|
|
|
@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)
|
|
|
|
|
|
@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)
|
|
|
|
@refresh_on_error
|
|
def mark_as_read(self, id: str):
|
|
return mark_as_read(self.token, id)
|
|
|
|
@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)
|
|
|
|
|
|
@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)
|
|
|
|
@refresh_on_error
|
|
def get_posts(self, username: str | None = None, limit: int = 20, cursor: int = 0, sort: str = '', tab: str = ''):
|
|
return get_posts(self.token, username, limit, cursor, sort, tab)
|
|
|
|
@refresh_on_error
|
|
def get_post(self, id: str):
|
|
return get_post(self.token, id)
|
|
|
|
@refresh_on_error
|
|
def edit_post(self, id: str, content: str):
|
|
return edit_post(self.token, id, content)
|
|
|
|
@refresh_on_error
|
|
def delete_post(self, id: str):
|
|
return delete_post(self.token, id)
|
|
|
|
@refresh_on_error
|
|
def pin_post(self, id: str):
|
|
return pin_post(self.token, id)
|
|
|
|
@refresh_on_error
|
|
def repost(self, id: str, content: str | None = None):
|
|
return repost(self.token, id, content)
|
|
|
|
@refresh_on_error
|
|
def view_post(self, id: str):
|
|
return view_post(self.token, id)
|
|
|
|
|
|
@refresh_on_error
|
|
def report(self, id: str, type: str = 'post', reason: str = 'other', description: str = ''):
|
|
return report(self.token, id, type, reason, description)
|
|
|
|
@refresh_on_error
|
|
def report_user(self, id: str, reason: str = 'other', description: str = ''):
|
|
return report(self.token, id, 'user', reason, description)
|
|
|
|
@refresh_on_error
|
|
def report_post(self, id: str, reason: str = 'other', description: str = ''):
|
|
return report(self.token, id, 'post', reason, description)
|
|
|
|
@refresh_on_error
|
|
def report_comment(self, id: str, reason: str = 'other', description: str = ''):
|
|
return report(self.token, id, 'comment', reason, description)
|
|
|
|
|
|
@refresh_on_error
|
|
def search(self, query: str, user_limit: int = 5, hashtag_limit: int = 5):
|
|
return search(self.token, query, user_limit, hashtag_limit)
|
|
|
|
@refresh_on_error
|
|
def search_user(self, query: str, limit: int = 5):
|
|
return search(self.token, query, limit, 0)
|
|
|
|
@refresh_on_error
|
|
def search_hashtag(self, query: str, limit: int = 5):
|
|
return search(self.token, query, 0, limit) |