Telegram Parser Чатов | Инструмент для автоматизации поиска чатов

Привет, разработчики!

Рад представить вам мой новый Telegram Парсер Чатов! Этот инструмент позволяет эффективно искать и анализировать чаты по заданным тегам и окончаниям.

Я занимаюсь разработкой программного обеспечения и готов создать что-то уникальное для ваших проектов (ну это так, к слову).

Особенности парсера:

• Асинхронный поиск чатов с использованием Telethon API
• Фильтрация по тегам и окончаниям с комбинированным поиском
• Продвинутое логирование действий и ошибок с ротацией *****
• Сохранение уникальных тегов чатов в удобном формате ссылки на телеграм
• Фильтрация по минимальному количеству подписчиков
• Проверка доступности комментариев в группах
• Определение языка контента с настраиваемым порогом (CIS/INT)
• Поддержка различных типов чатов (супергруппы, группы)
• Настраиваемые параметры поиска (количество проверяемых постов, порог языковой детекции)
• Обработка ошибок и пропуск проблемных чатов без прерывания работы
• Прогресс выполнения с отображением текущего/общего количества запросов

Установка и настройка:

Требования:

• Python 3.8+
• Библиотеки: asyncio, telethon, loguru
• Telegram API credentials (api_id и api_hash)

Необходимые зависимости:

python -m pip install pip asyncio telethon loguru --upgrade

Код

python -m pip install pip asyncio telethon loguru --upgrade

Настройка конфигурации:

Создайте файл `config.py` и добавьте ваши данные:
​

Настройка config.py

# Тут уже указано официальное приложение Telegram Desktop, если шаришь могёшь поменять
api_id = 611335
api_hash = 'd524b414d21f4d37f08684c1df41ac9c'

session_file = 'session.session' # Путь к файлу сессии

tags_file = 'tags.txt' # Путь к файлу с тегами
endings_file = 'ends.txt' # Путь к файлу окончаний
results_file = 'results.txt' # Путь к файлу результатов

min_subscribers = 1000 # Фильтр по минимальным подписчикам
posts_to_check = 5 # Количество постов для открытости комментариев
require_comments_enabled = True # Проверять ли открытость комментариев

lang = 'CIS' # Детектить по русскому языку CIS или INT чтоб другие языки
lang_detection_ratio = 0.8 # сейчас 80% (степень уверенности в языке, если 80% русский то значит русский)

Python

# Тут уже указано официальное приложение Telegram Desktop, если шаришь могёшь поменять

api_id = 611335

api_hash = 'd524b414d21f4d37f08684c1df41ac9c'



session_file = 'session.session' # Путь к файлу сессии



tags_file = 'tags.txt' # Путь к файлу с тегами

endings_file = 'ends.txt' # Путь к файлу окончаний

results_file = 'results.txt' # Путь к файлу результатов



min_subscribers = 1000 # Фильтр по минимальным подписчикам

posts_to_check = 5 # Количество постов для открытости комментариев

require_comments_enabled = True # Проверять ли открытость комментариев



lang = 'CIS'  # Детектить по русскому языку CIS или INT чтоб другие языки

lang_detection_ratio = 0.8  # сейчас 80% (степень уверенности в языке, если 80% русский то значит русский)

Использование парсера:

python parser.py

Код

python parser.py

Код парсера:
​

parser.py

import asyncio
from telethon import TelegramClient
from telethon.tl.functions.contacts import SearchRequest
from telethon.tl.functions.channels import GetFullChannelRequest
from loguru import logger
import os
import re

from config import (
api_id,
api_hash,
session_file,
tags_file,
endings_file,
results_file,
min_subscribers,
posts_to_check,
require_comments_enabled,
lang,
lang_detection_ratio
)

logger.add("parser.log", rotation="1 MB")
logger.info("Начало работы парсера.")

client = TelegramClient(session_file, api_id, api_hash)

async def read_lines(file_path):
if not os.path.exists(file_path):
logger.error(f"Файл {file_path} не найден.")
return []
with open(file_path, 'r', encoding='utf-8') as f:
lines = [line.strip() for line in f if line.strip()]
logger.info(f"Прочитано {len(lines)} строк из {file_path}.")
return lines

async def check_comments_enabled(channel_entity):
try:
posts = await client.get_messages(channel_entity, limit=posts_to_check)

if not posts:
return False

comment_enabled_count = 0
for post in posts:
if not hasattr(post, 'replies') or post.replies is None:
continue

if post.replies.comments:
comment_enabled_count += 1

return comment_enabled_count >= len(posts) / 2
except Exception as e:
logger.error(f"Ошибка при проверке комментариев: {e}")
return False

async def detect_language(channel_entity):
try:
posts = await client.get_messages(channel_entity, limit=10)

if not posts:
return None

russian_pattern = re.compile('[а-яА-ЯёЁ]')

total_chars = 0
russian_chars = 0

for post in posts:
if not post.message:
continue

message_text = post.message
total_chars += len(message_text)

matches = russian_pattern.findall(message_text)
russian_chars += len(matches)

if total_chars == 0:
return None

russian_ratio = russian_chars / total_chars

if russian_ratio >= lang_detection_ratio:
return 'CIS'
else:
return 'INT'

except Exception as e:
logger.error(f"Ошибка при определении языка: {e}")
return None

async def main():
await client.start()
logger.info("Клиент Telegram успешно запущен.")

tags = await read_lines(tags_file)
endings = await read_lines(endings_file)

if not tags or not endings:
logger.error("Не удалось прочитать теги или окончания. Завершение работы.")
return

chat_ids = set()
detected_lang = lang

total_queries = len(tags) * len(endings)
current_query = 0

for tag in tags:
for end in endings:
current_query += 1
search_query = f"{tag}{end}"
logger.info(f"Поиск {current_query}/{total_queries}: '{search_query}'")
try:
result = await client(SearchRequest(
q=search_query,
limit=100
))

if not result.chats:
logger.warning(f"По запросу '{search_query}' чатов не найдено.")
continue

for chat in result.chats:
try:
if hasattr(chat, 'broadcast') and chat.broadcast:
continue
elif (hasattr(chat, 'megagroup') and chat.megagroup) or hasattr(chat, 'title'):
if not chat.username:
continue

full_chat = await client(GetFullChannelRequest(channel=chat))
participants_count = full_chat.full_chat.participants_count

if participants_count < min_subscribers:
logger.info(f"Пропущен '{chat.title}': мало подписчиков ({participants_count} < {min_subscribers})")
continue

if require_comments_enabled:
comments_enabled = await check_comments_enabled(chat)
if not comments_enabled:
logger.info(f"Пропущен '{chat.title}': комментарии отключены")
continue

channel_lang = await detect_language(chat)
if channel_lang:
detected_lang = channel_lang

if lang != detected_lang:
logger.info(f"Пропущен '{chat.title}': несоответствующий язык (обнаружен {detected_lang}, требуется {lang})")
continue

chat_ids.add(chat.username)
logger.info(f"Найдена группа: {chat.title} (Username: {chat.username}, Подписчиков: {participants_count}, Язык: {detected_lang})")
else:
logger.warning(f"Неизвестный тип чата: {chat}")
except Exception as e:
logger.error(f"Ошибка при обработке чата '{getattr(chat, 'title', 'Неизвестно')}': {e}")
continue

except Exception as e:
logger.error(f"Ошибка при поиске '{search_query}': {e}")

if chat_ids:
with open(results_file, 'w', encoding='utf-8') as f:
for chat_id in chat_ids:
f.write(f"t.me/{chat_id}\n")
logger.info(f"Сохранено {len(chat_ids)} уникальных каналов/групп в {results_file}.")
else:
logger.warning("Не найдено ни одного подходящего чата.")

await client.disconnect()
logger.info("Клиент Telegram отключен. Работа завершена.")

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

Python

import asyncio

from telethon import TelegramClient

from telethon.tl.functions.contacts import SearchRequest

from telethon.tl.functions.channels import GetFullChannelRequest

from loguru import logger

import os

import re



from config import (

    api_id,

    api_hash,

    session_file,

    tags_file,

    endings_file,

    results_file,

    min_subscribers,

    posts_to_check,

    require_comments_enabled,

    lang,

    lang_detection_ratio

)



logger.add("parser.log", rotation="1 MB")

logger.info("Начало работы парсера.")



client = TelegramClient(session_file, api_id, api_hash)



async def read_lines(file_path):

    if not os.path.exists(file_path):

        logger.error(f"Файл {file_path} не найден.")

        return []

    with open(file_path, 'r', encoding='utf-8') as f:

        lines = [line.strip() for line in f if line.strip()]

    logger.info(f"Прочитано {len(lines)} строк из {file_path}.")

    return lines



async def check_comments_enabled(channel_entity):

    try:

        posts = await client.get_messages(channel_entity, limit=posts_to_check)



        if not posts:

            return False



        comment_enabled_count = 0

        for post in posts:

            if not hasattr(post, 'replies') or post.replies is None:

                continue



            if post.replies.comments:  

                comment_enabled_count += 1



        return comment_enabled_count >= len(posts) / 2

    except Exception as e:

        logger.error(f"Ошибка при проверке комментариев: {e}")

        return False



async def detect_language(channel_entity):

    try:

        posts = await client.get_messages(channel_entity, limit=10)



        if not posts:

            return None



        russian_pattern = re.compile('[а-яА-ЯёЁ]')



        total_chars = 0

        russian_chars = 0



        for post in posts:

            if not post.message:

                continue



            message_text = post.message

            total_chars += len(message_text)



            matches = russian_pattern.findall(message_text)

            russian_chars += len(matches)



        if total_chars == 0:

            return None



        russian_ratio = russian_chars / total_chars



        if russian_ratio >= lang_detection_ratio:

            return 'CIS'

        else:

            return 'INT'



    except Exception as e:

        logger.error(f"Ошибка при определении языка: {e}")

        return None



async def main():

    await client.start()

    logger.info("Клиент Telegram успешно запущен.")



    tags = await read_lines(tags_file)

    endings = await read_lines(endings_file)



    if not tags or not endings:

        logger.error("Не удалось прочитать теги или окончания. Завершение работы.")

        return



    chat_ids = set()

    detected_lang = lang  



    total_queries = len(tags) * len(endings)

    current_query = 0



    for tag in tags:

        for end in endings:

            current_query += 1

            search_query = f"{tag}{end}"

            logger.info(f"Поиск {current_query}/{total_queries}: '{search_query}'")

            try:

                result = await client(SearchRequest(

                    q=search_query,

                    limit=100

                ))



                if not result.chats:

                    logger.warning(f"По запросу '{search_query}' чатов не найдено.")

                    continue



                for chat in result.chats:

                    try:

                        if hasattr(chat, 'broadcast') and chat.broadcast:

                            continue

                        elif (hasattr(chat, 'megagroup') and chat.megagroup) or hasattr(chat, 'title'):

                            if not chat.username:

                                continue



                            full_chat = await client(GetFullChannelRequest(channel=chat))

                            participants_count = full_chat.full_chat.participants_count



                            if participants_count < min_subscribers:

                                logger.info(f"Пропущен '{chat.title}': мало подписчиков ({participants_count} < {min_subscribers})")

                                continue



                            if require_comments_enabled:

                                comments_enabled = await check_comments_enabled(chat)

                                if not comments_enabled:

                                    logger.info(f"Пропущен '{chat.title}': комментарии отключены")

                                    continue



                            channel_lang = await detect_language(chat)

                            if channel_lang:

                                detected_lang = channel_lang



                            if lang != detected_lang:

                                logger.info(f"Пропущен '{chat.title}': несоответствующий язык (обнаружен {detected_lang}, требуется {lang})")

                                continue



                            chat_ids.add(chat.username)

                            logger.info(f"Найдена группа: {chat.title} (Username: {chat.username}, Подписчиков: {participants_count}, Язык: {detected_lang})")

                        else:

                            logger.warning(f"Неизвестный тип чата: {chat}")

                    except Exception as e:

                        logger.error(f"Ошибка при обработке чата '{getattr(chat, 'title', 'Неизвестно')}': {e}")

                        continue



            except Exception as e:

                logger.error(f"Ошибка при поиске '{search_query}': {e}")



    if chat_ids:

        with open(results_file, 'w', encoding='utf-8') as f:

            for chat_id in chat_ids:

                f.write(f"t.me/{chat_id}\n")

        logger.info(f"Сохранено {len(chat_ids)} уникальных каналов/групп в {results_file}.")

    else:

        logger.warning("Не найдено ни одного подходящего чата.")



    await client.disconnect()

    logger.info("Клиент Telegram отключен. Работа завершена.")



if __name__ == "__main__":

    asyncio.run(main())

Создание тегов и окончаний:

Создайте файлы `tags.txt` и `ends.txt`, добавляя по одному тегу или окончанию на строку. Пример:

tags.txt

tech
news
gaming

Код

tech

news

gaming

ends.txt

group
chat
community

Код

group

chat

community

Результаты поиска:

Результаты будут сохранены в файл `results.txt` в формате ссылки. Этот файл можно использовать для дальнейшего анализа или управления чатами.


Пример ***** парсера:
​

parser.log

2024-04-27 12:00:00 Начало работы парсера.
2024-04-27 12:00:05 Клиент Telegram успешно запущен.
2024-04-27 12:00:10 Прочитано 3 строк из tags.txt.
2024-04-27 12:00:10 Прочитано 3 строк из ends.txt.
2024-04-27 12:00:15 Поиск 1/9: 'techgroup'
2024-04-27 12:00:20 Найдена группа: Tech Group (Username: techgroup)
...
2024-04-27 12:05:00 Сохранено 10 уникальных тегов чатов в results.txt.
2024-04-27 12:05:05 Клиент Telegram отключен. Работа завершена.

Код

2024-04-27 12:00:00 Начало работы парсера.

2024-04-27 12:00:05 Клиент Telegram успешно запущен.

2024-04-27 12:00:10 Прочитано 3 строк из tags.txt.

2024-04-27 12:00:10 Прочитано 3 строк из ends.txt.

2024-04-27 12:00:15 Поиск 1/9: 'techgroup'

2024-04-27 12:00:20 Найдена группа: Tech Group (Username: techgroup)

...

2024-04-27 12:05:00 Сохранено 10 уникальных тегов чатов в results.txt.

2024-04-27 12:05:05 Клиент Telegram отключен. Работа завершена.

Разработка на заказ:

Если вам нужно что-то особенное или у вас есть идеи для проектов, я с радостью помогу! За адекватную цену я могу разработать любые инструменты и решения, адаптированные под ваши нужды. Свяжитесь со мной для обсуждения деталей и получения индивидуального предложения.

​