Предисловие
В этой статье будет реализован парсер для Telegram, который будет собирать zip- и rar-архивы из каналов. Для управления файлами будет создана веб-панель, с помощью которой можно фильтровать архивы по заданным критериям, таким как названия папок и файлов, указанные в конфигурации. Также будет добавлена возможность получать уведомления о новых собранных файлах. Работа программы будет протестирована на каналах, публикующих логи, а также на канале, публикующем исходники различных проектов.Подготовка проекта
Первое, что нужно сделать — это создать проект и подготовить его основную структуру.А именно:
- главный Python-файл с названием main.py;
- папка static с CSS-файлом для стилей будущей панели;
- папка templates с index.html (веб-часть будущей панели);
- в папку проекта добавить папку с WinRAR (это понадобится для разархивации архивов из Telegram-каналов, если они в формате rar, поскольку не было найдено ни одной библиотеки, которая может это сделать самостоятельно);
- также внутри папки проекта нужно создать файл config.json (в нем будут храниться ключевые слова, по которым будет происходить поиск файлов внутри архивов).
Python:
import io
import os
import shutil
import zipfile
import rarfile
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from flask import Flask, jsonify, render_template, send_file, request
from flask_sqlalchemy import SQLAlchemy
from pyrogram import Client
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
import json
import progressbar
from aiogram import BotТак как панель будет на вебе, по классике я буду использовать Flask в связке с SQLAlchemy (с базой данных SQLite), поэтому его нужно инициализировать.
Python:
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///files.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)Далее нужно указать инициализацию базы данных внутри контекста приложения Flask, а также запуск самого Flask-приложения.
Python:
if __name__ == "__main__":
with app.app_context():
db.create_all()
app.run(debug=True, use_reloader=False)Далее укажем необходимые переменные.
Python:
rarfile.UNRAR_TOOL = os.path.abspath(r'WinRAR\UnRAR.exe') # Путь до unrar.exe
folder_extractions = ""
max_size =
channels = ["@channel", "@channel1", "@channe2", "@channe3"]
api_id =
api_hash = ""
session_string = ""
tg_client = Client(name="my_session", session_string=session_string, api_id=api_id, api_hash=api_hash)- rarfile.UNRAR_TOOL отвечает за путь к .exe файлу необходимого WinRAR для разархивации будущих логов.
- folder_extractions отвечает за путь к папке, куда будут разархивироваться логи.
- max_size отвечает за максимальный допустимый размер логов в байтах.
- channels — это список каналов, по которым будет проходиться программа.
- api_id, api_hash, session_string понадобятся для авторизации в аккаунте Telegram, с которого и будет происходить парсинг логов.
- tg_client создает и инициализирует новый клиент Telegram с последующей авторизацией, используя api_id, api_hash, и session_string.
Как получить session_string
Для того, чтобы получить сессию, придется написать небольшой скрипт. Использовать он будет библиотеку Pyrogram.
Python:
from pyrogram import Client
api_id = 123213123
api_hash = ""
with Client("my_account", api_id, api_hash) as client:
try:
client.start()
except:
pass
print("Сессия:", client.export_session_string())Получение api id и api hash
Для того чтобы получить необходимые данные, нужно перейти по ссылке: https://my.telegram.org/apps.Затем ввести номер телефона от аккаунта, который будет использован.
Далее нужно заполнить все поля, кроме последнего (оно не обязательно).
После заполнения нужно нажать на кнопку Create; после этого откроется страница, где будут находиться API ID и API hash.
После того как в код вставлены API ID и API hash, можно запустить скрипт для получения сессии. В консоли потребуется указать номер телефона, пройти 2FA (если оно есть) и ввести код из SMS.
P.S. Лучше не пытайтесь получить сессию с аккаунта, зарегистрированного на виртуальный номер. Такие аккаунты сразу же блокируются безвозвратно (проверено на моем опыте). В живых остаются лишь аккаунты, созданные на реальную SIM-карту.
Создание таблиц в базе данных
Теперь рассмотрим создание таблиц в базе данных. Таблиц будет всего три:- FileRecord
- FolderRecord
- Logs
Модель базы данных для хранения и записи всех скачанных архивов из Telegram будет использоваться для проверки на дубликаты файлов по размеру и имени файла.
Python:
class FileRecord(db.Model):
id = db.Column(db.Integer, primary_key=True)
file_id = db.Column(db.String, unique=True, nullable=False)
file_name = db.Column(db.String, nullable=False)
file_size = db.Column(db.Integer, nullable=False)
download_date = db.Column(db.DateTime, default=datetime.utcnow)
def __repr__(self):
return f'<FileRecord {self.file_name} ({self.file_size} bytes)>'- id — первичный ключ, который используется для уникализации каждой строки в базе данных (то есть это ключ, позволяющий создать уникальное значение id для каждой записи).
- file_id — file_id из Telegram.
- file_name — имя вложения из Telegram или имя архива.
- file_size — размер архива.
- download_date — дата загрузки архива из Telegram-канала.
Модель базы данных для хранения информации о папках и дополнительных данных будет использоваться для отображения всех файлов и их директорий на странице с логами, создавая более подробный отчет.
Python:
class FolderRecord(db.Model):
id = db.Column(db.Integer, primary_key=True)
logs_id = db.Column(db.Integer, db.ForeignKey('logs.id'), nullable=False)
file_record_id = db.Column(db.Integer, db.ForeignKey('file_record.id'),
nullable=False)
folder_name = db.Column(db.String, nullable=False)
rel_path = db.Column(db.String, nullable=True)
def __repr__(self):
return f'<FolderRecord {self.folder_name}, txt: {self.txt}, folder: {self.folder}>'- id — первичный ключ для уникализации каждой строки в базе данных.
- logs_id — внешний ключ на таблицу file_record на столбец id для связи между таблицами.
- folder_name — название папки с логом.
- rel_path — относительный путь до папки.
Модель базы данных для хранения информации о логах и вывода на странице с логами будет использоваться для отображения различных заданных параметров по фильтру.
Python:
class Logs(db.Model):
id = db.Column(db.Integer, primary_key=True)
file_record_id = db.Column(db.Integer, db.ForeignKey('file_record.id'),
nullable=False)
folder_name = db.Column(db.String, nullable=False)
count_param = db.Column(db.String)
full_path = db.Column(db.String)- id — первичный ключ для уникализации каждой строки в базе данных.
- file_record_id — внешний ключ на таблицу FileRecord на столбец id.
- folder_name — название папки с логом.
- full_path — полный путь до лога.
Парсинг архивов из Telegram
Далее начнем реализовывать логику для скачивания архивов из каналов.Как это устроено:
- Берется канал из массива в переменной channels.
- Получаем историю сообщений канала.
- Ищем в этой истории сообщения, содержащие вложения в форматах zip или rar.
- Затем в найденных сообщениях проверяем, чтобы файл с таким же именем и размером не был записан в базу данных. Также проверяем, чтобы файл не превышал допустимого максимального размера, указанного в переменной max_size, и дополнительно проверяем, чтобы время и дата создания поста в канале не были раньше времени и даты запуска софта.
- Вызываем инициализацию метода ProgressBar из библиотеки progressbar, и далее происходит запуск анимации.
- После этого происходит скачивание вложения с вызовом функции progress (будет написана позже) для обновления анимации прогресс-бара.
- После скачивания происходит запись данных о файле в базу данных.
- В конце вызывается функция для разархивации архива, о которой будет написано позже.
Для начала рассмотрим, как выглядит история сообщений из канала.
Парсинг сообщений
Теперь рассмотрим саму функцию для парсинга более подробно.
Python:
async def download_smallest_archive():
try:
async with tg_client:
for channel in channels:
async for message in tg_client.get_chat_history(channel):
if message.document and message.document.mime_type in ["application/zip", "application/vnd.rar"]:
file = message.document
if FileRecord.query.filter_by(file_name=file.file_name,
file_size=file.file_size).first() or file.file_size > max_size or file.date < time_started:
continue
print(f"Файл качается {file.file_name}")
bar = progressbar.ProgressBar(maxval=file.file_size, widgets=['Loading: ', progressbar.AnimatedMarker()]).start()
file_path = await tg_client.download_media(file.file_id, file_name=file.file_name, progress=progress, progress_args=(file.file_size, bar))
file_record = FileRecord(
file_id=file.file_id,
file_name=file.file_name,
file_size=file.file_size,
download_date=file.date.astimezone()
)
db.session.add(file_record)
db.session.commit()
print(f"Файл скачался {file.file_name}")
await extract_archive(file_path, f"{folder_extractions}/{os.path.basename(file_path)}", file_record.id)
print(f"Файл разархивировался {file.file_name}")
except Exception as err:
print(err)Далее проверяем файл на уникальность по file_name и file_size (эта проверка проводится до скачивания, а не по хешу после скачивания). Также проверяется максимальный допустимый размер файла и дата, чтобы отсеять старые сообщения и избежать загрузки слишком больших файлов, например, 5-гигабайтных логов.
Перед загрузкой файла инициализируем и запускаем анимацию для отображения процесса загрузки. Затем передаем в функцию для обновления (бар) идентификатор файла, его имя и дополнительные параметры, такие как максимальный размер и объект прогресс-бара. Запускаем функцию загрузки tg_client.download_media и попутно запускаем функцию обновления прогресс-бара под названием progress (будет показана позже).
После загрузки файла добавляем запись в таблицу file_record и сохраняем изменения в базе данных. Затем разархивируем файл с помощью функции extract_archive, передавая путь до архива, место для распаковки и идентификатор файла, чтобы пометить его как разархивированный.
Функция для обновления прогрессбара
Python:
async def progress(current, total, file_size=None, bar=None):
try:
print(f"current: {current}")
print(f"full_size: {file_size}")
if current < file_size: # Что бы избежать ошибки ZeroDivisionError и переполнения
bar.update(current) # обновляет анимацию загрузки bar
except ZeroDivisionError:
pass # Игнорировать ошибку если скачано 0 байтНастройка планировщика задач (scheduler)
Теперь можно было бы рассмотреть функцию разархивации, но перед этим рассмотрим использование scheduler (планировщик, позволяющий запланировать запуск функций с конкретным интервалом).Можно было бы использовать бесконечный цикл для функции парсинга, но тогда Flask будет занимать основной поток, и функция не будет запускаться, так как поток занят. Чтобы это исправить, можно запустить Flask в отдельном потоке, а в основном вызывать асинхронную функцию парсинга. Однако в данном случае мне показалось, что использование scheduler — более подходящий вариант, так как на форумах люди советуют не выносить Flask в отдельный поток, а использовать его контекстное меню.
Настройка параметров запуска планировщика задач и его задач.
Python:
scheduler = BackgroundScheduler()
scheduler.add_job(
scheduled_download,
'interval',
max_instances=1,
minutes=1,
next_run_time=datetime.now()
)
scheduler.start()scheduler.add_job — это добавление новой задачи в планировщик с конкретными параметрами, такими как:
- максимальное количество задач,
- интервал между каждым запуском,
- параметр для первичного запуска без ожидания.
Как видно, внутри задачи указана функция scheduled_download. В данной функции будет вызываться функция парсинга, и она будет вызываться в контексте приложения Flask.
Python:
def scheduled_download():
with app.app_context():
tg_client.loop.run_until_complete(download_smallest_archive())Разархивация файлов
Теперь можем рассмотреть функцию для разархивации архивов из каналов.И первое, что указано внутри функции — это определение типа файла: zip или rar, а также разархивация соответствующим методом.
Python:
async def extract_archive(file_path, dest_folder, file_record_id):
os.makedirs(dest_folder, exist_ok=True)
try:
if file_path.endswith('.zip'):
with zipfile.ZipFile(file_path, 'r') as archive:
archive.extractall(dest_folder)
elif file_path.endswith('.rar'):
with rarfile.RarFile(file_path, 'r') as archive:
archive.extractall(path=dest_folder)
else:
raise ValueError("Неподдерживаемый формат файла. Поддерживаются только zip и rar.")
except (rarfile.BadRarFile, rarfile.PasswordRequired, rarfile.NeedFirstVolume, rarfile.RarCannotExec) as e:
print(f"Ошибка с файлом: {e}")
returnНебольшая загвоздка
папок и корректной записи данных в базу данных.Дело в том, что разные пользователи архивируют по-своему: у кого-то в архиве сразу находятся файлы логов формата с TXT файлами и несколько папок, а у кого-то — множество папок, каждая из которых содержит файлы логов и другие данные. Чаще всего встречается второй вариант, и для него была написана логика записи имени основной папки каждого лога из архива.
Однако, если использовать тот же метод для архива, который разархивировался в папку, содержащую сразу файлы логов, в базу данных записывался первый попавшийся файл в качестве основного названия папки лога. В связи с этим я применил, на мой взгляд, максимально возможный костыль: проверяется наличие файла с расширением .txt непосредственно в разархивированной папке. Если такой файл присутствует, предполагается, что это уже папка с данными лога. В таком случае создается новая папка внутри разархивированной, и все файлы перемещаются в нее, тем самым стандартизируя структуру путей как для архивов с множеством логов, так и для архивов с одним логом.
Возможно объяснение покажется запутанным, поэтому покажу оба вида папок с логами.
Когда один лог в архиве:
Когда много логов в архиве:
Проверка структуры файла
А вот и сама логика проверки типа структуры папки и создания дополнительной папки:
Python:
if bool([f for f in os.listdir(dest_folder) if f.endswith('.txt')]):
# Создаем новую папку внутри dest_folder с уникальным именем (без расширения .zip или .rar)
archive_name = os.path.splitext(os.path.basename(file_path))[0]
new_folder_path = os.path.join(dest_folder, archive_name)
# Перемещаем все файлы и папки из dest_folder в новую папку
for item in os.listdir(dest_folder):
old_path = os.path.join(dest_folder, item)
new_path = os.path.join(new_folder_path, item)
# Проверяем, что не пытаемся переместить папку в саму себя
if old_path != new_path:
# Убедимся, что новая папка не существует, иначе пропустим перемещение
if not os.path.exists(new_folder_path):
os.makedirs(new_folder_path)
# Если это файл, перемещаем его
if os.path.isfile(old_path):
shutil.move(old_path, new_path)
# Если это директория, перемещаем всю директорию
elif os.path.isdir(old_path):
shutil.move(old_path, new_path)Далее мы получаем ключевые слова (параметры) из конфига, которые будем использовать в дальнейшем при обработке файлов и подсчете статистики по фильтрам искомых файлов:
Python:
with open("config.json", 'r') as file:
keywords = json.load(file).get("name_file", [])Вот и сама обработка
Python:
for dir_log in os.listdir(dest_folder):
count_list = [] # массив для счетчик с нашими параметрами из конфига
for keyword in keywords:
# Подсчитаем количество файлов и папок, содержащих ключевое слово в названии
count = 0
for dirpath, dirnames, filenames in os.walk(os.path.join(dest_folder, dir_log)):
# Проверка в самой папке (dirpath) — если название папки содержит ключевое слово
if keyword in os.path.basename(dirpath):
count += 1
# Проверка в папках (dirnames)
if any(keyword in dirname for dirname in dirnames):
count += 1
# Проверка в файлах (filenames)
if any(keyword in filename for filename in filenames):
count += 1
count_list.append(str(count)) # добавляем этот счетчик в архивПроверяем на первое ключевое слово каждую из этих переменных (dirpath, dirname, filename), обновляя счетчик.
В конце мы собираем наш массив со счетчиками с помощью count_list.append(str(count)).
Так же и с остальными ключевыми словами (параметрами из конфига), пока список не закончится.
После прохождения цикла нам нужно добавить запись в базу данных о логе.
Вот функция, которая, используя полученные переменные, создает запись.
Python:
log = Logs(full_path=os.path.abspath(os.path.join(dest_folder, dir_log)), folder_name=dir_log,
file_record_id=file_record_id, count_param=",".join(count_list))
db.session.add(log)
db.session.commit()Далее, чтобы сделать уведомления, соберем наше сообщение и создадим клавиатуру.
Сама клавиатура будет inline, чтобы прикрепляться к сообщению, а не появляться в GUI переписки как reply."
Python:
ikb = InlineKeyboardMarkup(
inline_keyboard=[
[
InlineKeyboardButton(text=f"Скачать", url=f"{host}/download_folder/{log_id}"),
],
])inline_keyboard нужен для добавления списка клавиш.
InlineKeyboardButton нужен для инициализации самого объекта клавиши, в котором указываем текст и ссылку, поэтому параметры — text и url, а не callback.
Для запуска самой рассылки используем admins_tg_id, где список наших ID, которым делаем рассылку, и функцию send_message для отправки сообщения с клавиатурой.
Python:
for admin_tg_id in admins_tg_id:
await bot.send_message(chat_id=admin_tg_id, text=text, reply_markup=ikb)Добавление записей в таблицу FolderRecord для будущего вывода каждого файла в панели.
Python:
for dirpath, _, filenames in os.walk(os.path.join(dest_folder, dir_log)):
for filename in filenames:
rel_path = os.path.relpath(os.path.join(dirpath, filename), dest_folder)
folder_record = FolderRecord(
file_record_id=file_record_id,
folder_name=rel_path.split(os.sep)[0],
rel_path=rel_path,
logs_id=log_id
)
db.session.add(folder_record)
db.session.commit()Веб часть
Маршрут основной страницы
Теперь можно приступить к работе с веб-панелью. Первым делом будет реализован маршрут до страницы index.html.
Python:
@app.route("/index")
def index():
with open("config.json", 'r') as file:
config = json.load(file)
config_columns = config.get("name_file", [])
all_logs = Logs.query.all()
return render_template('index.html', all_logs=all_logs, config_columns=config_columns)Маршрут для скачивания логов
Далее рассмотрим еще один маршрут, чтобы в дальнейшем добавить в панели кнопку для скачивания конкретного лога.
Python:
@app.route("/download_folder/<int:log_id>")
def download_folder(log_id):
# Получаем лог по ID
log = Logs.query.get(log_id)
if not log:
return "Log not found", 404
folder_path = log.full_path
print(folder_path)
if not os.path.exists(folder_path):
return "Folder not found", 404
folder_name = os.path.basename(folder_path) # Название папки из полного пути
zip_filename = f"{folder_name}.zip"
zip_filepath = os.path.join(folder_zipper_panel, zip_filename) # путь где создание архива
if not os.path.exists(folder_zipper_panel):
os.makedirs(folder_zipper_panel)
# Создание архива
with zipfile.ZipFile(zip_filepath, 'w', zipfile.ZIP_DEFLATED) as zipf:
for dirpath, dirnames, filenames in os.walk(folder_path):
for filename in filenames:
file_path = os.path.join(dirpath, filename)
zipf.write(file_path,
os.path.relpath(file_path, folder_path)) # Добавляем файл в архив с относительным путем
# Отправляем архив в браузер
return send_file(zip_filepath, as_attachment=True)Работа с html
Теперь нам нужно перейти в шаблон файл index.html (он же templates) для отображения всей информации на странице с возможностью скачивания файлов.А именно, таблица со столбцами:
- названия папки лога,
- параметры из конфига (в моем случае появится 4 столбца),
- а также кнопка скачать и ее форма для отправки запроса.
HTML:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>File Records</title>
<link rel="stylesheet" href="{{ url_for('static', filename='index.css') }}">
</head>
<body>
<h1>File Records</h1>
<table border="1">
<thead>
<tr>
<th>Folder Name</th>
{% for column in config_columns %}
<th>{{ column }}</th>
{% endfor %}
<th>Actions</th>
</tr>
</thead>
</table>
</body>
</html>Теперь рассмотрим логику отображения данных самих логов внутри tbody.
HTML:
<tbody>
{% for log in all_logs %}
<tr>
<td>{{ log.folder_name }}</td>
{% set counts = log.count_param.split(',') %}
{% for count in counts %}
<td>{{ count }}</td>
{% endfor %}
<td>
<form action="{{ url_for('download_folder', log_id=log.id) }}">
<button type="submit">Download</button>
</form>
</td>
</tr>
<tr>
<td colspan="{{ config_columns | length + 2 }}">
<details>
<summary>Показать файлы</summary>
<ul>
<div class="full_info">
{% for record in log.folder_records %}
{{ record.rel_path }}
{% endfor %}
</div>
</ul>
</details>
</td>
</tr>
{% endfor %}
</tbody>P.S. Форма action нужна для перехода на url_for('download_folder') с указанием нашего id лога и отправки для скачивания. Далее делаем кнопку "Показать все файлы лога" и перечисляем их из log.folder_records (именно для этого создавалась таблица folder_records).
Стили для панели:
CSS:
body{
display: flex;
flex-direction: column;
align-items: center;
background: #17212b;
color: #f5f5f5;
font-family: Regular, sans-serif
}
h1 {
text-align: center;
color: #f5f5f5;
margin: 2% 0% 0%;
}
table {
width: 100%;
margin: 20px 0;
border-collapse: collapse;
box-shadow: 0 0 10px rgba(0, 0, 0, 0.3);
background-color: #1e1e1e;
}
th, td {
padding: 12px;
text-align: center;
border: 1px solid #35414d;
}
th {
background-color: #366693;
color: #f5f5f5;
}
td {
background-color: #202b36;
color: #f5f5f5;
}
tr:nth-child(even) td {
background-color: #202b36;
}
tr:hover {
background-color: #444;
}
button{
background: #4c9ce2;
color: #f5f5f5;
border: none;
padding: 5%;
border-radius: 5px;
width: 100px;
}
.full_info{
}Вот как по итогу выглядит веб-панель:
Вот как выглядят уведомления в Telegram:
На этом с написанием проекта закончено. Также я настроил его для работы с Docker, чтобы вы могли быстро развернуть его на своём VDS или хостинге с K8s (Kubernetes).
Как установить проект через Docker
Для этого, как ни странно, понадобится сам Docker, который можно скачать по ссылке: https://www.docker.com/Далее нужно скачать сам проект, который я выложил на GitHub: https://github.com/overlordgamedev/Telegram-File-Parser/tree/main
Затем нужно зайти в скачанную папку с проектом, открыть командную строку в этой папке и ввести команду:
docker-compose up -d
После этой команды начнется скачивание образа Docker с Python и необходимых библиотек. По окончанию скачивания можно будет перейти по ссылке 127.0.0.1:5000 или по общедоступной ссылке на ваш сервер.
Вывод
Я старался создать софт, который будет полезен для данного форума, и считаю, что мне это удалось. Этот софт можно использовать на приватных облаках с логами, что лучше, чем просто включить уведомления в канале. В таком случае вы избегаете лишних уведомлений с рекламными постами и получаете статистику по каждому логу. Кроме того, софт универсален и не ограничивается только работой с логами — им можно сортировать любые архивы из каналов по любым названиям папок и файлов.Из сложностей в разработке хочу отметить, как корректно работать со структурой папок и логов внутри них, а также правильно сохранять данные в базу данных и отображать их в веб-панели. Как всегда, если у вас есть замечания по реализации какой-либо логики, с удовольствием прочту ваши предложения в комментариях.
Разработано OverlordGameDev специально для форума xss.pro.
Ссылка на статью в виде документа - https://docs.google.com/document/d/1-HhuQ1tewoHjuNmI3soeU1eMLaVfH8lFxQPhWfFXXEI/edit?usp=sharing
