from pyrogram import Client
from pyrogram.errors import FloodWait
import asyncio
async def get_post_stats(client: Client, channel_id: int, message_id: int):
"""
Получает статистику поста: просмотры, реакции и количество комментариев.
Args:
client: Экземпляр клиента Pyrogram
channel_id: ID канала
message_id: ID сообщения
Returns:
dict: Словарь со статистикой
"""
try:
# Получаем сообщение
message = await client.get_messages(channel_id, message_id, replies=-1)
# Получаем статистику просмотров
views = message.views if message.views else 0
# Получаем реакции
reactions = {}
if message.reactions:
for reaction in message.reactions.reactions:
reaction_id = reaction.custom_emoji_id if reaction.custom_emoji_id else reaction.emoji
reactions[reaction_id] = reaction.count
comments_count = await client.get_discussion_replies_count(channel_id, message_id)
stats = {
"views": views,
"reactions": reactions,
"forwards": message.forwards or 0,
"comments": comments_count
}
return stats
except FloodWait as e:
print(f"Превышен лимит запросов, ждем {e.value} секунд")
await asyncio.sleep(e.value)
return await get_post_stats(client, channel_id, message_id)
except Exception as e:
print(f"Произошла ошибка: {e}")
return None
# Пример использования
async def main():
# Замените на свои данные
api_id = "0"
api_hash = "0"
channel_id = -1 # ID канала
message_id = 1 # ID сообщения
async with Client("my_account", api_id=api_id, api_hash=api_hash) as app:
stats = await get_post_stats(app, channel_id, message_id)
if stats:
print(f"Просмотры: {stats['views']}")
print(f"Переслали: {stats['forwards']}")
print(f"Реакции: {stats['reactions']}")
print(f"Комментарии: {stats['comments']}")
if __name__ == "__main__":
asyncio.run(main())
Простой
