Парсер + | Telegram чатов

Всем привет
Хочу поделиться с вами моим скриптом, который позволяет парсить телеграм-чаты по ключевым словам, языку
и количеством участников
​

код

import re
from telethon.sync import TelegramClient
from telethon.tl.functions.contacts import SearchRequest
from telethon.tl.functions.channels import GetFullChannelRequest
from telethon.tl.types import PeerChannel
from telethon.tl.functions.messages import GetHistoryRequest
from colorama import Fore
from langdetect import detect
from telethon.tl.types import PeerChannel, Channel
from langdetect import DetectorFactory
import asyncio

api_id = ''
api_hash = ""

client = TelegramClient('search', api_id, api_hash)

keywords = []

with open('settings.txt', 'r', encoding='utf-8') as file:
content = file.read()

min_participants_match = re.search(r'мин\. участников\s*=\s*"([^"]+)"', content)
language_match = re.findall(r'язык\s*=\s*"([^"]+)"', content)

MIN_PARTICIPANTS = int(min_participants_match.group(1)) if min_participants_match else None
languages = language_match if language_match else []


print(MIN_PARTICIPANTS)
print(languages)

NUM_MESSAGES_TO_ANALYZE = 5

def chat_by_messages(messages):
if not messages:
return False
combined_messages = " ".join([msg.message for msg in messages if hasattr(msg, 'message') and msg.message])
try:
DetectorFactory.seed = 0
detected_language = detect(combined_messages)
if detected_language not in (languages):
return False
else:
print(f" Язык сообщений: {detected_language}")
except Exception as e:
return False
return True

async def search_chats(keywords):
for keyword in keywords:
try:
result = await client(SearchRequest(q=keyword, limit=5))

for chat in result.chats:
await get_channel(chat.id)

except Exception as e:
pass


def save_username(username):
try:
with open("channels.txt", "r", encoding="utf-8") as f:
existing_usernames = f.read().splitlines()
replace_username = username.replace('@', 'https://t.me/')
except FileNotFoundError:
existing_usernames = []

if replace_username not in existing_usernames:
with open("channels.txt", "a", encoding="utf-8") as f:
f.write(replace_username + "\n")
print(Fore.GREEN + f"Сохранен юзернейм: {username}")
else:
print(Fore.WHITE + f"Юзернейм {username} уже существует в файле.")

async def get_channel(chat_id):
try:
full_channel = await client(GetFullChannelRequest(PeerChannel(chat_id)))
channel = full_channel.chats[0]

if isinstance(channel, Channel) and not channel.broadcast:

if hasattr(full_channel.full_chat, 'participants_count') and full_channel.full_chat.participants_count >= MIN_PARTICIPANTS:
history = await client(GetHistoryRequest(
peer=channel,
offset_id=0,
offset_date=None,
add_offset=0,
limit=NUM_MESSAGES_TO_ANALYZE,
max_id=0,
min_id=0,
hash=0
))

messages = history.messages

if chat_by_messages(messages):
if hasattr(channel, 'username') and channel.username:
username = "@" + channel.username
save_username(username)
else:
pass
else:
pass
else:
pass
except Exception as e:
print(Fore.RED + f"Ошибка при получении информации о канале {chat_id}: {e}")


async def main():
await client.start()
with open('words.txt', 'r') as file:
keyword = file.readlines()
keyword = [word.strip() for word in keyword]
for i in keyword:
keywords.append(i)
await search_chats(keywords)
await client.disconnect()

if __name__ == '__main__':
asyncio.run(main())

Python

import re
from telethon.sync import TelegramClient
from telethon.tl.functions.contacts import SearchRequest
from telethon.tl.functions.channels import GetFullChannelRequest
from telethon.tl.types import PeerChannel
from telethon.tl.functions.messages import GetHistoryRequest
from colorama import Fore
from langdetect import detect
from telethon.tl.types import PeerChannel, Channel
from langdetect import DetectorFactory
import asyncio

api_id = ''
api_hash = ""

client = TelegramClient('search', api_id, api_hash)

keywords = []

with open('settings.txt', 'r', encoding='utf-8') as file:
    content = file.read()

min_participants_match = re.search(r'мин\. участников\s*=\s*"([^"]+)"', content)
language_match = re.findall(r'язык\s*=\s*"([^"]+)"', content)

MIN_PARTICIPANTS = int(min_participants_match.group(1)) if min_participants_match else None
languages = language_match if language_match else []


print(MIN_PARTICIPANTS)
print(languages)

NUM_MESSAGES_TO_ANALYZE = 5

def chat_by_messages(messages):
    if not messages:
        return False
    combined_messages = " ".join([msg.message for msg in messages if hasattr(msg, 'message') and msg.message])
    try:
        DetectorFactory.seed = 0
        detected_language = detect(combined_messages)
        if detected_language not in (languages):
            return False
        else:
            print(f"  Язык сообщений: {detected_language}")
    except Exception as e:
        return False
    return True

async def search_chats(keywords):
    for keyword in keywords:
        try:
            result = await client(SearchRequest(q=keyword, limit=5))

            for chat in result.chats:
                await get_channel(chat.id)

        except Exception as e:
            pass


def save_username(username):
    try:
        with open("channels.txt", "r", encoding="utf-8") as f:
            existing_usernames = f.read().splitlines()
            replace_username = username.replace('@', 'https://t.me/')
    except FileNotFoundError:
        existing_usernames = []

    if replace_username not in existing_usernames:
        with open("channels.txt", "a", encoding="utf-8") as f:
            f.write(replace_username + "\n")
        print(Fore.GREEN + f"Сохранен юзернейм: {username}")
    else:
        print(Fore.WHITE + f"Юзернейм {username} уже существует в файле.")

async def get_channel(chat_id):
    try:
        full_channel = await client(GetFullChannelRequest(PeerChannel(chat_id)))
        channel = full_channel.chats[0]

        if isinstance(channel, Channel) and not channel.broadcast:

            if hasattr(full_channel.full_chat, 'participants_count') and full_channel.full_chat.participants_count >= MIN_PARTICIPANTS:
                history = await client(GetHistoryRequest(
                    peer=channel,
                    offset_id=0,
                    offset_date=None,
                    add_offset=0,
                    limit=NUM_MESSAGES_TO_ANALYZE,
                    max_id=0,
                    min_id=0,
                    hash=0
                ))

                messages = history.messages

                if chat_by_messages(messages):
                    if hasattr(channel, 'username') and channel.username:
                        username = "@" + channel.username
                        save_username(username)
                    else:
                        pass
                else:
                    pass
            else:
                pass
    except Exception as e:
        print(Fore.RED + f"Ошибка при получении информации о канале {chat_id}: {e}")


async def main():
    await client.start()
    with open('words.txt', 'r') as file:
        keyword = file.readlines()
        keyword = [word.strip() for word in keyword]
        for i in keyword:
            keywords.append(i)
    await search_chats(keywords)
    await client.disconnect()

if __name__ == '__main__':
    asyncio.run(main())

функционал

1. Установите Python и библиотеку Telethon: выполните команду pip install telethon.
2. Создайте новую папку для скрипта
3. В этой папке создайте файл: main.py ,words.txt,channels.txt,settings.txt
4. Скопируйте приведённый выше код и вставьте его в файл main.py :
5. Получите свой api_id и api_hash на сайте my telegram org и вставьте их в соответствующие переменные в main.py .
6. В words.txt напишите ключевые слова на разных строках для поиска.
7. channels.txt оставьте пустым, а в settings.txt запишите как на видео
8. Запустите файл main.py через командную строку, используя команду python main.py .
9.Введите номер и подтвердите кодом

демонстрация

​