Привет, разработчики! Рад представить вам мой новый 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 Code 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 Code 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 Code tech news gaming ends.txt group chat community Code 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 отключен. Работа завершена. Code 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 отключен. Работа завершена. Разработка на заказ: Если вам нужно что-то особенное или у вас есть идеи для проектов, я с радостью помогу! За адекватную цену я могу разработать любые инструменты и решения, адаптированные под ваши нужды. Свяжитесь со мной для обсуждения деталей и получения индивидуального предложения.
qpEHIKC, python -m pip install pip asyncio telethon loguru --upgrade Code python -m pip install pip asyncio telethon loguru --upgrade
zxcFlezyi, если использовать его прямо по назначению будет польза - парсинг чатов для того что бы позже спарсить от туда аудиторию и заинвайтить, или это уже хуйня не актуально, никто так не делает?)
https://github.com/k1p1k-code/ParserChatsTelegramKeywords оцени мой парсер там еще aiogram бот прикручин