Быстрый старт¶
Установка¶
pip install aioitd
Без авторизации¶
Апи хештегов доступно без авторизации.
from aioitd import AsyncITDClient
import asyncio
async def main():
async with AsyncITDClient() as client:
hashtags = await client.search_hashtags('a')
for hashtag in hashtags:
print(f"#{hashtag.name}: {hashtag.posts_count}")
asyncio.run(main())
Авторизация¶
Для остальных функций необходим refresh_token
Чтобы получить refresh_token:
- Войди в аккаунт на итд.com
- Откройть devtools
- Среди табов выбрать network (сеть)
- Перезагрузить страницу
- Найти запрос
refresh - В куки запроса скопировать
refresh_token
Или:
- Войти в аккаунт на итд.com
- Откройть devtools
- Среди табов выбрать application (приложение)
- Перезагрузить страницу
- Выбрать куки и найти куку
refersh_token
После передайте refresh_token в AsyncITDClient:
from aioitd import AsyncITDClient
import asyncio
refresh_token = "ca1291a4a990b985a57b880ed3fb863eef80ac3990acc68e8f106a783e3af402"
async def main():
async with AsyncITDClient(refresh_token) as client:
user = await client.get_user("nowkie")
print(
f"id: {user.id}\n"
f"username: {user.username}\n"
f"follower_count: {user.followers_count}"
)
asyncio.run(main())
Создание поста¶
from aioitd import AsyncITDClient, File, Post
import asyncio
from uuid import UUID
refresh_token = "ВАШ ТОКЕН"
async def main():
async with AsyncITDClient(refresh_token) as client:
images_path = ["sun.png", "снег.gif", "python.jpg"]
files_ids: list[UUID] = []
for path in images_path:
with open(path, 'rb') as f:
file: File = await client.upload_file(f)
files_ids.append(file.id)
post: Post = await client.create_post("ТЕКСТ ПОСТА", attachment_ids=files_ids)
asyncio.run(main())
Поменять баннер¶
async def main():
async with AsyncITDClient(refresh_token) as client:
with open(r"file.gif", 'rb') as file:
image = await client.upload_file(file)
await client.update_profile(banner_id=image.id)
Уведомления по SSE¶
from aioitd import NotificationEvent, ConnectedEvent
async def main():
async with AsyncITDClient(refresh_token) as client:
async with client.connect_notifications() as events:
async for event in events:
if isinstance(event, ConnectedEvent):
print("Прослушка начата")
elif isinstance(event, NotificationEvent):
print(event.actor)
asyncio.run(main())
RateLimitError¶
RateLimitError ограничение по количеству запросов.
retray_after — через сколько можно повторить запрос.
while True:
try:
await client.post("content")
except RateLimitError as ex:
await asyncio.sleep(ex.retry_after)