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

Maehdakvan · 25 окт 2024

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

Рад представить вам мой новый 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` и добавьте ваши данные:
# Тут уже указано официальное приложение 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
Код парсера:
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` в формате ссылки. Этот файл можно использовать для дальнейшего анализа или управления чатами.

Пример ***** парсера:
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 отключен. Работа завершена.
Разработка на заказ:

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