- Регистрация
- Янв 5, 2024
- Сообщения
- 69
- Реакции
- 108
Всем шалом!
Работать будем с - python
Python скрипт для получения информации и постов из телеграм канала
Код:
sudo apt-get install python3 && python3-pip
pip3 install asyncio aiohttp bs4
Библиотека Teleview:
- Асинхронная Python библиотека для получения информации о канале, его постах, и комментариев
- Зависимости: asyncio, aiohttp, bs4
Возможности библиотеки:
- Получения информации о канале: название, описание, кол-во подписчиков, фотография канала
- Поиск постов по ID, получение 20 актуальных постов, получение всех постов в канале
- Получение информации о постах: текст, дата, кол-во просмотров, прикрепленные медиа (видео-сообщений, так и видео-файлов и голосовых сообщений) итд...
- Поиск комментариев по ID и получение актуальных комментариев
- Получение информации о комментарии: текст, дата, автор и так далее и так далее...
function getChannel
Асинхронная функция для получения информации о канале
Принимает:- str url - ссылка или ЮЗ канала
- class Post: если канал найден
- None: если канал НЕ найден
- Из библиотеки
function getPost
Асинхронная функция для получения поста в канале по ID
Принимает:- class Channel | str channel - Обьект канала или ссылка на канал
- int id - ID поста
- class Post: если пост найден
- None: если пост НЕ найден
- Из библиотеки
- Из класса Channel | аргумент channel не требуется
function getComments
Асинхронная функция для получения актуальных комментариев "под" постом
Принимает:- class Post post - Обьект поста в канале
- int limit - лимит по количеству комментариев
- list[class Comment]: если комментарие найдены
- None: если комменариев нет
- Из библиотеки
- Из класса Post | аргумент post не требуется
function getComment
Асинхронная функция для получения комментария по ID
Принимает:- class Post post - Обьект поста в канале
- int id - ID комментария
- class Comment: если комментарий найден
- None: если комменариев нет или комментарий с таким ID отсуствует
- Из библиотеки
- Из класса Post | аргумент post не требуется
function getLatest
Асинхронная функция для получения 20 последних постов в канале
Ничего не принимаетРезультат:
- list[class Post]: Список постов от нового к старому
- Из класса Channel
function isPublic
Асинхронная функция которая узнает публичный ли канал или нет
Ничего не принимаетРезультат:
- bool: публичный канал или нет
- Из класса Channel
function isPrivate
Асинхронная функция которая узнает приватный ли канал или нет
Ничего не принимаетРезультат:
- bool: приватный канал или нет
- Из класса Channel
function toDict
Асинхронная функция для выдачи данных, которые содержатся в классе, в виде словаря
Ничего не принимаетРезультат:
- dict: Словарь с информацией о классе
- Из любого класса
function getAllPosts
Асинхронная функция которая получает все посты в канале
Предупреждение- Выполнение функции может занять длительное время!
- Может преодолеть лимит запросов!
- class Channel channel - обьект Channel
- list[class Post]: список всех постов от старого к новому
- Из библиотеки
- Из класса Channel | аргумент channel не требуется
class Channel
- prop url [str] - Ссылка на канал
- prop name [str] - Название канала
- prop description [str] - Описание канала
- prop subscribers [int] - Количество подписчиков на канал
- prop picture [str | None] - Ссылка на картинку канала
- prop private [bool] - Приватный канал или нет
- async func getPost - Функция для получения поста в канале по ID
- async func getLatest - Функция для получения последних 20 постов в канале
- async func getAllPosts - Функция для получения всех постов в канале
- async func isPublic - Узнать приватный канал или нет
- async func isPrivate - Узнать публичный канал или нет
- async func toDict - Функция для отображения всех данных в формате словаря
class Post
- prop channel [class Channel] - обьект канала
- prop url [str] - ссылка на пост
- prop id [int] - ID поста
- prop text [str | None] - текст поста
- prop media [list[Media]] - список Медиа обьектов (фото, видео итд)
- prop views [int] - количество просмотров
- prop datetime [str] - строка с датой и временем отправки поста в GMT+0
- async func getComments - Функция для получения последних комментариев "под" постом
- async func getComment - Функция для получения комментария по ID
- async func toDict - Функция для отображения всех данных в формате словаря
class Media
- prop url [str] - ссылка на загрузку медиа-контента
- prop type [str] - тип медиа-контента (photo, video, round_video, voice / jpg, mp4, mp4, ogg )
- async func toDict - Функция для отображения всех данных в формате словаря
class Comment
- prop post [class Post] - обьект поста
- prop id [int] - ID комментария
- prop text [str] - текст комментария
- prop reply [int | None] - ID комментария на который был произведен ответ
- prop author [class Author] - класс где хранится информация об авторе комментария
- prop datetime [str] - строка с датой и временем отправки комментария в GMT+0
- async func toDict - Функция для отображения всех данных в формате словаря
class Author
- prop name [str] - имя пользователя
- prop username [str | None] - ЮЗ пользователя
- prop photo [str | None] - ссылка на фотографию профиля
- async func toDict - Функция для отображения всех данных в формате словаря
Установка, настройка :
Код:
Создаёте папку teleview для библиотеки и в ней:
__init__.py
Код:
from .corotune import *
from .models import *
version = '1.2.1'
source = 'http://commudazrdyhbullltfdy222krfjhoqzizks5ejmocpft3ijtxq5khqd.onion'
random_UA = True
request_header = {
'User-Agent': '1'
}
if random_UA:
import random
request_header['User-Agent'] = random.choice(['Windows','MacOS','Linux','iPhone', 'Android'])+'; '+random.choice(['Chrome 100', 'Firefox 93', 'Safari 14', 'Opera 30', 'Edge'])
Код:
corotune.py
Код:
""" --- Асинхронная функция которая получает все посты в канале ---
ВНИМАНИЕ: ВЫПОЛНЕНИЕ МОЖЕТ ЗАНЯТЬ ДЛИТЕЛЬНОЕ ВРЕМЯ!
ВНИМАНИЕ: МОЖЕТ ПРЕОДОЛЕТЬ ЛИМИТ ПО ЗАПРОСАМ!
Принимает:
* channel: обьект Channel
Результат:
* list[Post]: список всех постов от старого к новому
"""
posts = {}
latest = await channel.getLatest()
for p in latest: posts[p.id] = p
print(list(posts.keys()))
while True:
pids = list(posts.keys())
pids.sort()
last_pid = pids[0]
url = f'https://t.me/s/{channel.url.split("/")[-1]}/{last_pid}'
async with aiohttp.ClientSession() as client:
async with client.get(url, headers=request_header) as resp: bs = bs4.BeautifulSoup(await resp.text('utf-8'), 'html.parser')
loaded_posts = bs.findAll(class_='tgme_widget_message')
for post in loaded_posts:
post_object = runtime.BStoPost(post, channel)
posts[post_object.id] = post_object
if 1 in posts.keys(): break
pids = list(posts.keys())
pids.sort()
result = []
for id in pids: result.append(posts[id])
return result
Код:
models.py
Код:
from . import runtime, corotune
class Media:
def __init__(self, url, type) -> None:
self.url = url
self.type = type
async def toDict(self) -> dict:
return {
'url': self.url,
'type': self.type
}
class Author:
def __init__(self, name, username, photo) -> None:
self.name = name
self.username = username
self.photo = photo
async def toDict(self) -> dict:
return {
'name': self.name,
'username': self.username,
'photo': self.photo
}
class Comment:
def __init__(self, post, id, text, reply, author, datetime) -> None:
self.post = post
self.id = id
self.text = text
self.reply = reply
self.author = author
self.datetime = datetime
async def toDict(self) -> dict:
return {
'post': self.post.url,
'id': self.id,
'text': self.text,
'reply': self.reply,
'author': await self.author.toDict(),
'datetime': self.datetime
}
class Post:
def __init__(self, channel, url, id, text, media, views, datetime, data) -> None:
self.channel = channel
self.url = url
self.id = id
self.text = text
self.media = media
self.views = int(views)
self.datetime = datetime
self.__data = data
async def getComments(self, limit = 10) -> list[Comment] | None:
"""--- Асинхронная функция для получения последних комментариев "под" постом ---
Принимает:
* limit: лимит по количеству постов. максимум 100
Результат:
* list[Comment]: если комментарии есть
* None: если их под постом нету
"""
return await corotune.getComments(self, limit)
async def getComment(self, id) -> Comment | None:
"""--- Асинхронная функция для поиска комментария по ID ---
Принимает:
* id: ID поста
Результат:
* Comment: если комментарий найден
* None: если под постом нету комментариев или комментарий не найден
"""
return await corotune.getComment(self, id)
async def toDict(self) -> dict:
return {
'channel': self.channel.url,
'url': self.url,
'id': self.id,
'text': self.text,
'media': [await media.toDict() for media in self.media],
'views': self.views,
'datetime': self.datetime
}
class Channel:
def __init__(self, url, name, description, subscribers, picture, data) -> None:
self.private = False
self.url = url
self.name = name
self.description = description
self.subscribers = int(subscribers)
self.picture = picture
self.__data = data
self.__posts = {}
async def getPost(self, id: int) -> Post | None:
"""--- Асинхронная функция для поиска поста в канале ---
Принимает:
* id: айди поста
Результат:
* Post: если пост найден
* None: если пост НЕ найден
"""
return await corotune.getPost(self, id)
async def getLatest(self) -> list[Post]:
""" --- Получает 20 актуальных постов, возращая от актуального к старому ---
Ничего не принимает
Результат:
* list[Post]: список постов от актуального к старому
"""
return [runtime.BStoPost(post, self) for post in self.__data.findAll(class_='tgme_widget_message')]
async def getAllPosts(self) -> list[Post]:
""" --- Асинхронная функция которая получает все посты в канале ---
ВНИМАНИЕ: ВЫПОЛНЕНИЕ МОЖЕТ ЗАНЯТЬ ДЛИТЕЛЬНОЕ ВРЕМЯ!
ВНИМАНИЕ: МОЖЕТ ПРЕОДОЛЕТЬ ЛИМИТ ПО ЗАПРОСАМ!
Ничего не принимает
Результат:
* list[Post]: список всех постов от старого к новому
"""
if not self.__posts: self.__posts = await corotune.getAllPosts(self)
return self.__posts
async def isPublic(self) -> bool:
""" --- Асинхронная функция которая узнает публичный ли канал или нет ---
Ничего не принимает
Результат:
* bool: Публичный канал или нет
"""
return not self.private
async def isPrivate(self) -> bool:
""" --- Асинхронная функция которая узнает приватный ли канал или нет ---
Ничего не принимает
Результат:
* bool: Приватный канал или нет
"""
return self.private
async def toDict(self) -> dict:
return {
'url': self.url,
'name': self.name,
'description': self.description,
'subscribers': self.subscribers,
'picture': self.picture,
'private': self.private
}
Код:
runtime.py
Код:
from .models import *
import bs4, aiohttp
def BStoChannel(bs, name) -> Channel:
info = bs.find(class_='tgme_channel_info')
try: description = info.find(class_='tgme_channel_info_description').text # описание
except: description = None
subs_str = info.find(class_='tgme_channel_info_counter').find(class_='counter_value').text.replace('.','') # получение кол-ва подписчиков
if 'K' in subs_str: subscribers = float(subs_str[:-1])*1000//1 # конвертация из str в int
elif 'M' in subs_str: subscribers = float(subs_str[:-1])*1000*1000//1
else: subscribers = int(subs_str)
try: picture = info.find(class_='tgme_page_photo_image').find('img').get('src') # фото канала
except: picture = None
return Channel(f'https://t.me/{name}', info.find(class_='tgme_channel_info_header_title').text, description, subscribers, picture, bs)
def BStoPost(bs, channel) -> Post:
try: text = bs4.BeautifulSoup(str(bs.find(class_='tgme_widget_message_text')).replace('<br/>','\n'), 'html.parser').text # получаем текст сообщения, форматируя <br/> в \n
except AttributeError: text = None
id = bs.get('data-post').split('/')[-1]
try:
views = bs.find(class_='tgme_widget_message_views').text
if 'K' in views: views = float(views[:-1])*1000//1
elif 'M' in views: views = float(views[:-1])*1000*1000//1
else: views = int(views)
except: views = 0
media = []
photos = bs.findAll(class_='tgme_widget_message_photo_wrap')
videos = bs.findAll(class_='tgme_widget_message_video')
if photos or videos:
for photo in photos: media.append(Media(photo.get('style').split("background-image:url('")[1].split("')")[0], 'photo/jpg'))
for video in videos: media.append(Media(video.get('src'), 'video/mp4'))
roundvideo = bs.findAll(class_='tgme_widget_message_roundvideo')
if roundvideo: media.append(Media(roundvideo[0].get('src'), 'round_video/mp4'))
voicemsg = bs.findAll(class_='tgme_widget_message_voice')
if voicemsg: media.append(Media(voicemsg[0].get('src'), 'voice/ogg'))
try: datetime = bs.find(class_='datetime').get('datetime')
except: datetime = bs.find(class_='time').get('datetime')
return Post(channel, f'{channel.url}/{id}', int(id), text, media, views, datetime, bs)
def BStoComments(msg, post) -> list[Comment]:
try: reply = int(msg.find(class_='tgme_widget_message_reply').get('data-reply-to'))
except: reply = None
return Comment(post, int(msg.get('data-post-id')), bs4.BeautifulSoup(str(msg.find(class_='js-message_text')).replace('<br/>','\n'), 'html.parser').text, reply, BStoAuthor(msg), msg.find(class_='tgme_widget_message_date').find('time').get('datetime'))
def BStoComment(msg, post) -> Comment:
try: reply = int(msg.find(class_='tgme_widget_message_reply').get('href').split('/')[-1])
except: reply = None
return Comment(post, int(msg.get('data-post-id')), bs4.BeautifulSoup(str(msg.find(class_='js-message_text')).replace('<br/>','\n'), 'html.parser').text, reply, BStoAuthor(msg), msg.find(class_='tgme_widget_message_date').find('time').get('datetime'))
def BStoAuthor(msg) -> Author:
try: author_username = msg.find(class_='tgme_widget_message_user').find('a').get('href').split('/')[-1]
except: author_username = None
try: author_photo = msg.find(class_='tgme_widget_message_user_photo').find('a').find('i').find('img').get('src')
except: author_photo = None
return Author(msg.find(class_='tgme_widget_message_author_name').text, author_username, author_photo)
Код:
import teleview, asyncio
async def main():
# --- Получения информации о канале --- #
channel = await teleview.getChannel('contest')
print(f'Название канала: {channel.name}')
print(f'Описание канала: {channel.description}')
if channel.private: print('Это приватный канал')
# или await channel.isPrivate()
else: print('Это публичный канал')
# --- Поиск поста по ID --- #
post = await channel.getPost(198)
# или await teleview.getPost(channel, 198)
# или await teleview.getPost('contest', 198)
print(f'* Текст поста с ID {post.id}: {post.text}')
# --- Получение 20 актуальных постов --- #
latest_posts = await channel.getLatest()
for latest_post in latest_posts:
print(f'* Текст поста с ID {latest_post.id}: {latest_post.text}')
# --- Поиск комментария по ID --- #
comment = await post.getComment(141108)
# или await teleview.getComment(post, 141108)
print(f'- Текст комментария с ID {comment.id}: {comment.text}')
# --- Получение 15 актуальных комментариев --- #
latest_comments = await post.getComments(limit = 15)
# или await teleview.getComments(post, limit = 15)
for latest_comment in latest_comments:
print(f'- Текст комментария с ID {latest_comment.id}: {latest_comment.text}')
asyncio.run(main())
Удачи и главное успехов!
На чай :
ETH:
0x72FD965183eF965831aa2D11554223357BE46F63
BTC :
bc1qnuz27ww36vsvupqcte9wdsnt06ts4hn90ahyrt
USDT TRC20 :
TSPsBxnYjbQSeXMjV2C7zT6VQAoKXjQpoC

