Парсинг и анализ Telegram-групп в целях выявления и сбора оперативно значимой информации

от

Вся информация представлена исключительно в демонстрационных целях

В соответствии с пунктом 1 части 1 статьи 10.6 Федерального закона от 27.07.2006 № 149-ФЗ «Об информации, информационных технологиях и о защите информации» (далее – ФЗ «Об информации») социальной сетью является сайт и (или) страница сайта в сети «Интернет», и (или) информационная система, и (или) программа для электронных вычислительных машин, которые предназначены и (или) используются их пользователями для предоставления и (или) распространения посредством созданных ими персональных страниц информации на государственном языке Российской Федерации, государственных языках республик в составе Российской Федерации или иных языках народов Российской Федерации, на которых может распространяться реклама, направленная на привлечение внимания потребителей, находящихся на территории Российской Федерации, и доступ к которым в течение суток составляет более пятисот тысяч пользователей сети «Интернет», находящихся на территории Российской Федерации.

Определение принадлежности информационного ресурса к социальным сетям может быть осуществлено на официальном сайте Федеральной службы по надзору в сфере связи, информационных технологий и массовых коммуникаций (https://530-fz.rkn.gov.ru/) в разделе «Проверка ресурсов» через специальную форму.

Telegram – система мгновенного обмена сообщений. С 2021 года Роскомнадзором включена в перечень социальных сетей. По количеству ежедневных посещений (50,9 миллионов человек) занимает второе место среди аналогичных цифровых площадок.

Telegram как платформа обеспечивает высокую скорость передачи данных, широкий охват аудитории и относительную анонимность, что делает его эффективным инструментом для оперативного обмена информацией. В соответствии с Федеральным законом от 12.08.1995 № 144-ФЗ «Об оперативно-розыскной деятельности», оперативно значимая информация может быть получена из открытых источников, к которым относятся и публичные Telegram-каналы и группы. При этом сбор и использование такой информации должны осуществляться в рамках действующего законодательства, с соблюдением требований конфиденциальности и защиты персональных данных (ст. 7 Федерального закона от 27.07.2006 № 152-ФЗ «О персональных данных»).

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

Если группа открытая (публичная) или у вас имеется доступ к данной группе, то парсинг и последующий анализ данных может осуществляться средствами Python (ниже представлен скрипт для реализации парсинга).

Установка Python и настройка IDE рассматривались нами ранее.

Для работы скрипта необходимо:

— создать новый проект, активировать виртуальное окружение и установить необходимые зависимости;

— зарегистрировать свое приложение https://my.telegram.org/auth?to=apps и получить api_id и api_hash;

— в скрипте заменить YOUR_API_ID, YOUR_API_HASH и YOUR_PHONE на свои данные;

— настроить список ключевых слов под свои нужды;

— запустить скрипт;

— при первом запуске нужно будет ввести код подтверждения из Telegram.

Скрипт будет сохранять в базу (файл telegram_post.db) только те сообщения, которые содержат хотя бы одно слово из списка keywords.

Рекомендации по улучшению работы скрипта:

— чтобы анализировать конкретные группы, можно изменить обработчик событий, добавив фильтр:

client.add_event_handler(handle_new_post, events.NewMessage(chats=['group_username']))

— для обработки больших объемов данных можно добавить пагинацию или ограничение по дате;

— для получения максимального эффекта, рекомендуется применять лемматизацию текста.

Код скрипта:

import sqlite3
from datetime import datetime
from telethon import TelegramClient, events
from telethon.tl.types import PeerChannel

# Конфигурация
api_id = 'YOUR_API_ID'  # Замените на ваш API ID
api_hash = 'YOUR_API_HASH'  # Замените на ваш API Hash
phone = 'YOUR_PHONE'  # Ваш номер телефона, связанный с Telegram
session_name = 'session_name'  # Имя файла сессии

# Список ключевых слов для поиска
keywords = [
    'купить', 'продать', 'заказать', 
    'доставка', 'оптом', 'розница',
    'скидка', 'акция', 'промокод'
]

# Подключаемся к SQLite
conn = sqlite3.connect('telegram_posts.db')
cursor = conn.cursor()

# Создаем таблицу, если она не существует
cursor.execute('''
CREATE TABLE IF NOT EXISTS posts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER,
    message TEXT,
    group_id INTEGER,
    group_name TEXT,
    post_date TEXT,
    keywords_found TEXT
)
''')
conn.commit()

# Функция для проверки наличия ключевых слов
def check_keywords(text):
    found = [word for word in keywords if word.lower() in text.lower()]
    return ', '.join(found) if found else None

# Функция обработки новых сообщений
async def handle_new_post(event):
    # Получаем информацию о сообщении
    message = event.message
    text = message.text or message.raw_text
    
    if not text:
        return
    
    # Проверяем на ключевые слова
    found_keywords = check_keywords(text)
    
    if found_keywords:
        # Получаем информацию о группе
        if isinstance(message.peer_id, PeerChannel):
            group_id = message.peer_id.channel_id
            group = await event.get_chat()
            group_name = group.title
        else:
            group_id = None
            group_name = None
        
        # Получаем информацию о пользователе
        user_id = message.from_id.user_id if message.from_id else None
        
        # Записываем в базу данных
        cursor.execute('''
        INSERT INTO posts (user_id, message, group_id, group_name, post_date, keywords_found)
        VALUES (?, ?, ?, ?, ?, ?)
        ''', (
            user_id,
            text,
            group_id,
            group_name,
            message.date.isoformat(),
            found_keywords
        ))
        conn.commit()
        
        print(f'Найдено совпадение в группе {group_name} от пользователя {user_id}')

# Основная функция
async def main():
    # Подключаемся к клиенту Telegram
    client = TelegramClient(session_name, api_id, api_hash)
    await client.start(phone)
    
    print("Клиент запущен. Начинаю мониторинг сообщений...")
    
    # Добавляем обработчик новых сообщений
    client.add_event_handler(handle_new_post, events.NewMessage())
    
    # Запускаем клиент
    await client.run_until_disconnected()

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