Здравствуйте! В этой статье рассмотрим, как с помощью Python мониторить сайты компаний, парсить отчёты из PDF, извлекать ключевые данные и отправлять обработанные результаты в Telegram.
Публичные компании обязаны публиковать свою отчётность (финансовые результаты, годовые отчёты, пресс-релизы) на своих сайтах, часто в формате PDF. Для трейдера или инвестора скорость получения и анализа этой информации критически важна: тот, кто первым увидит тренд или аномалию в отчёте, может принять лучшее решение и получить преимущество на рынке.
Мы создадим скрипт на Python, который будет скачивать с сайта (разберём на примере Яндекс) отчёт в формате PDF, преобразовывать неструктурированный текст в структурированные данные, извлекать ключевые метрики (выручка, чистая прибыль) и отправлять сжатый инсайт в Telegram-бот.
В рамках проекта решим 4 задачи:
загрузка финансового отчёта в PDF с сайта;
извлечение из файла текста;
анализ текста с помощью LLM для отбора и структурирования отчёта;
публикация в Telegram-канал.
Таким образом, можно автоматизировать работу с неструктурированными данными и превращать их в ценные, удобочитаемые с помощью современных AI-инструментов.
Наше решение мы развернём в движке приложений Amvera, так как это проще использования VPS и есть API к таким LLM, как GPT5 с оплатой рублёвой картой.
В примере кода все персональные данные для доступа к API и Telegram заменены на фиктивные.
Структура проекта и зависимости
Наш проект состоит из четырёх последовательных скриптов. Архитектура и ключевые библиотеки:
Модуль 1: 1_Download.py
Задача: Загрузка целевого PDF-файла с официального сайта.
Используемые зависимости:
requests: Для выполнения HTTP-запросов к веб-сайту.BeautifulSoup4 (bs4): Для парсинга HTML и поиска ссылок на нужный PDF-файл.urllib.parse: Для корректного объединения базового URL и относительных путей.
Модуль 2: 2_Parser_PDF.py
Задача: Конвертация скачанного PDF в сырой текст.
Используемые зависимости:
PyMuPDF (fitz): Мощная библиотека для точного извлечения текста из PDF-документов, включая документы со сложной разметкой.
Модуль 3: 3_Analytics.py
Задача: Анализ извлечённого текста с помощью AI.
Используемые зависимости:
aiohttp: Для асинхронных HTTP-запросов к LLM предоставляемых в Amvera.asyncio: Для организации асинхронного выполнения кода, что критически важно для избежания долгих блокировок при работе с нейросетью.logging: Для детального логирования каждого этапа процесса.
Модуль 4: 4_Send.py
Задача: Публикация результата в Telegram.
Используемые зависимости:
telethon: Асинхронный клиент для Telegram API, предоставляющий полный контроль над отправкой сообщений.python-dotenv: Для загрузки чувствительных переменных окружения (API-ключи, токены) из файла.env.
Конфигурация и окружение
Все параметры работы скриптов вынесены в переменные или в начало файлов – это удобно для настройки без правки основной логики. Критически важные данные хранятся в переменных окружения в Amvera в файле .env. Но можно сохранять и непосредственно в секретах сервиса.
Перед запуском на Amvera необходимо настроить все эти переменные в веб-интерфейсе панели управления.
Детальный разбор кода
1. Загрузка PDF
Скрипт имитирует работу браузера, чтобы получить HTML-код страницы. Затем он ищет все ссылки (<a> теги) и фильтрует их, чтобы найти ссылку на PDF-файл с финансовой отчётностью. Критерии поиска: ссылка должна содержать ключевые слова (financials, IFRS, financial) и иметь расширение .pdf.
# Получаем содержимое страницы
response = requests.get(url, headers=headers)
response.raise_for_status()
# Парсим HTML
soup = BeautifulSoup(response.text, 'html.parser')
# Ищем все ссылки на странице и фильтруем их
all_links = soup.find_all('a', href=True)
pdf_links = []
for link in all_links:
href = link['href']
if ('financials' in href or 'IFRS' in href or 'financial' in href) and href.endswith('.pdf'):
pdf_links.append(href)Если прямая ссылка не найдена, скрипт пытается найти её по тексту ссылки (Отчётность, Financial Statements). После обнаружения PDF скачивается и сохраняется на диск.
2. Парсинг PDF в текст
Этот скрипт предельно прост. Он использует библиотеку PyMuPDF (fitz) для открытия PDF-файла и итерации по всем его страницам. Текст с каждой страницы извлекается и записывается в один большой текстовый файл.
def pdf_to_text(pdf_path, txt_path):
doc = fitz.open(pdf_path)
text = ""
for page_num in range(len(doc)):
page = doc.load_page(page_num)
text += page.get_text() + "\n"
# . сохранение текста в файл .Этот этап подготавливает данные для самого сложного и интересного шага – анализа AI.
3. Анализ текста с помощью LLM LLaMA через Amvera API
Это ядро всего проекта. Скрипт отправляет объединённый текст в модель LLM Amvera с очень детальным и строгим промптом (системным сообщением). Для примера в нашем коде используется модель llama8b.
Промпт в работе с языковыми моделями, это не просто запрос, это детальное техническое задание, которое определяет формат, стиль, структуру и содержание ответа.
Гибкость этого подхода в том, что, не меняя ни строчки кода, а лишь редактируя системный промпт, вы можете полностью изменить задачу модели: с анализа новостей переключиться на финансовую аналитику, поиск рисков или составление выжимки по проектам.
Пример 1: Финансовый дайджест для инвесторов (вариант)
Цель: Создать сжатый отчёт о ключевых финансовых показателях за период.
Промпт:
Ты финансовый аналитик. Проанализируй предоставленный текст отчётности компании «Яндекс».
Сосредоточься на следующих ключевых метриках:
- Выручка за отчётный период (указать динамику к прошлому периоду)
- Чистая прибыль/убыток (динамика)
- Основные статьи расходов (на что больше всего потратили)
- Ключевые драйверы роста или падения выручки
- Крупные сделки, поглощения или продажа активов, если были
- Прогнозы руководства на следующий квартал/год, если они указаны.
Структура ответа:
1. Ключевые показатели: [Сводная таблица с цифрами]
2. Что произошло: [Краткий аналитический комментарий 3-4 предложения]
3. Куда движется компания: [Упомянуть про стратегию, инвестиции и прогнозы]
Используй только данные из отчёта. Излагай сухо, фактологично, без воды. Все цифры сопровождай единицами измерения (млрд р., млн $ США).
Пример 2: Поиск рисков и упоминаний регуляторов (вариант)
Цель: Выявить потенциальные риски для бизнеса, упоминания регулятивных органов и судебных разбирательств.
Промпт:
Внимательно изучи текст отчёта компании «Яндекс». Выступи в роли юриста-аналитика.
Выбери и перечисли всю информацию, касающуюся:
1. Регулятивных рисков: Упоминания ФАС, Роскомнадзора, ФНС, новых законов или подзаконных актов, которые оказывают или могут оказать влияние на бизнес.
2. Судебных разбирательств: Текущие или потенциальные иски, размер требований, стороны конфликта.
3. Геополитических рисков: Упоминания санкций, ограничений на международных рынках, проблем с поставками оборудования/ПО.
4. Прочих рисков: Упоминания кибератак, утечек данных, проблем с поставщиками.
Структура ответа:
- Категория риска: [Например, Регулятивное разбирательство]
- Факт: [Краткое описание, например: «ФАС возбудила дело о нарушении антимонопольного законодательства»]
- Сумма/Размер: [если указан, например: «Размер штрафа может составить до 2% от выручки на рынке»]
- Статус: [если указан, например: «Рассмотрение назначено на декабрь 2025 года»]
Излагай максимально кратко, только факты из документа. Не давай собственных оценок.
Промт можно редактировать под свои задачи, указывая цель, структуру и собственные правила. Чтобы легко обновлять его простым перезапуском проекта на удалённом сервере, мы используем переменную SYSTEM_PROMPT в файле Analytics.py.
Это превращает ваш статичный конвейер в универсальный и гибкий инструмент для анализа любых текстовых данных, где для смены задачи достаточно просто изменить текстовую инструкцию в конфигурации.
4. Отправка в Telegram
Финальный скрипт отвечает за публикацию обработанного текста. Он использует Telethon для асинхронной работы с Telegram API.
Чтение результата: Скрипт ищет файл
result.txt(или другой указанный файл) в текущей директории.Разбивка на сообщения: Поскольку Telegram имеет лимит в ~4000 символов на сообщение, скрипт умно разбивает длинный текст на части. Скрипт старается разбивать текст по абзацам (два переноса строки
\n\n), а если абзац слишком длинный, то по словам, чтобы не обрывать предложения.
def split_long_paragraph(paragraph: str, max_len: int = 4000) -> list:
"""Разбивает длинный абзац на части по словам"""
chunks = []
while paragraph:
if len(paragraph) <= max_len:
chunks.append(paragraph)
break
split_index = paragraph.rfind(' ', 0, max_len)
# . логика разбивки .
return chunks
Отправка с задержкой: Сообщения отправляются с небольшой задержкой (
await asyncio.sleep(1)), чтобы избежать лимитов флуд-контроля Telegram.
Результат лога на сайте
Полный код
import re
import asyncio
import os
import logging
import requests
import httpx
import json
import fitz
from datetime import datetime, timedelta
from urllib.parse import urljoin
from telethon import TelegramClient
from telethon.sessions import StringSession
from dotenv import load_dotenv
from bs4 import BeautifulSoup
import pytz
# ========== ЗАГРУЗКА ПЕРЕМЕННЫХ ИЗ .env ==========
load_dotenv()
# ========== ЛОГИРОВАНИЕ ==========
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# ========== КОНФИГУРАЦИЯ ==========
API_ID = int(os.getenv('API_ID'))
API_HASH = os.getenv('API_HASH')
BOT_TOKEN = os.getenv('BOT_TOKEN')
CHANNEL_LINK = os.getenv('CHANNEL_LINK', "https://t.me/+8bLwTjyymSdjMDQy")
TIMEZONE = pytz.timezone('Europe/Moscow')
BOT_SESSION_STRING = os.getenv('BOT_SESSION_STRING')
# Конфигурация LLM
LLM_BASE_URL = "https://kong-proxy.yc.amvera.ru/api/v1/models/llama"
LLM_API_KEY = "345475782567879655"
LLM_MODEL = "llama8b"
# Пути к файлам
DATA_FOLDER = os.getenv('DATA_FOLDER', "/data")
import os
os.makedirs(DATA_FOLDER, exist_ok=True)
def download_pdf():
"""Скачивание PDF файла с финансовой отчетностью"""
logger.info("? Запуск скачивания PDF файла")
# Настройки
url = 'https://ir.yandex.ru/financial-releases'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
try:
# Получаем содержимое страницы
response = requests.get(url, headers=headers)
response.raise_for_status()
# Парсим HTML
soup = BeautifulSoup(response.text, 'html.parser')
# Ищем все ссылки на странице
all_links = soup.find_all('a', href=True)
# Фильтруем ссылки, которые ведут на PDF файлы с финансовой отчетностью
pdf_links = []
for link in all_links:
href = link['href']
if ('financials' in href or 'IFRS' in href or 'financial' in href) and href.endswith('.pdf'):
pdf_links.append(href)
if not pdf_links:
# Если не нашли прямых ссылок, попробуем найти по тексту
for link in all_links:
if link.get_text().strip() in ['Отчетность', 'Financial Statements']:
pdf_links.append(link['href'])
# Выбираем первую найденную ссылку
pdf_url = pdf_links[0]
# Если ссылка относительная, преобразуем в абсолютную
if not pdf_url.startswith('http'):
pdf_url = urljoin(url, pdf_url)
logger.info(f"Найдена ссылка на отчетность: {pdf_url}")
# Скачиваем файл
pdf_response = requests.get(pdf_url, headers=headers)
pdf_response.raise_for_status()
# Определяем имя файла
filename = os.path.basename(pdf_url)
if not filename.endswith('.pdf'):
filename = f"Yandex_Financial_Report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
# Сохраняем файл в папку data
filepath = os.path.join(DATA_FOLDER, filename)
with open(filepath, 'wb') as f:
f.write(pdf_response.content)
logger.info(f'Файл успешно сохранен: {filepath}')
logger.info(f'Размер файла: {len(pdf_response.content)} байт')
return filepath
except Exception as e:
logger.error(f'Произошла ошибка при скачивании PDF: {str(e)}')
raise
def pdf_to_text(pdf_path):
"""Преобразование PDF в текст"""
logger.info(f"? Конвертация PDF в текст: {pdf_path}")
# Формируем полные пути к файлам
txt_filename = "output.txt"
txt_path = os.path.join(DATA_FOLDER, txt_filename)
try:
# Открываем PDF файл
doc = fitz.open(pdf_path)
text = ""
# Извлекаем текст из всех страниц
for page_num in range(len(doc)):
page = doc.load_page(page_num)
text += page.get_text() + "\n"
# Сохраняем текст в файл
with open(txt_path, "w", encoding="utf-8") as txt_file:
txt_file.write(text)
doc.close()
logger.info(f"Текст успешно сохранен в {txt_path}")
return txt_path
except Exception as e:
logger.error(f"Ошибка при конвертации PDF: {e}")
raise
def process_text():
"""Обработка текста с помощью LLM"""
logger.info("? Обработка текста с помощью LLM")
# Пути к файлам
INPUT_FILE = os.path.join(DATA_FOLDER, "output.txt")
OUTPUT_FILE = os.path.join(DATA_FOLDER, "result.txt")
# Чтение содержимого файла
try:
with open(INPUT_FILE, 'r', encoding='utf-8') as f:
content = f.read().strip()
except FileNotFoundError:
logger.error(f"Файл {INPUT_FILE} не найден")
raise
except Exception as e:
logger.error(f"Ошибка при чтении файла: {e}")
raise
if not content:
logger.error("Файл пуст")
raise ValueError("Файл пуст")
logger.info(f"Прочитано из файла: {len(content)} символов")
# Формирование промпта
prompt = f" Составь резюме, приведи основные экономические показатели сравнивая их по годам, дай прогноз.:\n\n{content}"
# Отправка запроса к LLM
headers = {
"X-Auth-Token": f"Bearer {LLM_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": LLM_MODEL,
"messages": [
{
"role": "user",
"text": prompt
}
]
}
try:
logger.info("Отправка запроса к LLM...")
with httpx.Client(timeout=120.0) as client:
response = client.post(LLM_BASE_URL, headers=headers, json=payload)
response.raise_for_status()
data = response.json()
# Извлекаем ответ из правильной структуры
summary = data["result"]["alternatives"][0]["message"]["text"]
# Сохранение результата
with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
f.write(summary)
logger.info(f"Результат сохранен в файл: {OUTPUT_FILE}")
logger.info(f"Длина ответа: {len(summary)} символов")
return OUTPUT_FILE
except Exception as e:
logger.error(f"Ошибка при запросе к LLM: {e}")
raise
def split_long_paragraph(paragraph: str, max_len: int = 4000) -> list:
"""Разбивает длинный абзац на части по словам"""
chunks = []
while paragraph:
if len(paragraph) <= max_len:
chunks.append(paragraph)
break
# Ищем место для разбиения (последний пробел перед лимитом)
split_index = paragraph.rfind(' ', 0, max_len)
if split_index == -1:
# Если нет пробелов - вынужденное разбиение
split_index = max_len
chunk = paragraph[:split_index].strip()
if chunk:
chunks.append(chunk)
paragraph = paragraph[split_index:].strip()
return chunks
async def send_to_channel(content: str):
"""Отправка текстового контента в канал с сохранением структуры абзацев"""
bot_client = TelegramClient(
StringSession(BOT_SESSION_STRING) if BOT_SESSION_STRING else 'session_bot.session',
API_ID, API_HASH
)
await bot_client.start(bot_token=BOT_TOKEN)
try:
entity = await bot_client.get_entity(CHANNEL_LINK)
# Разбиваем текст на абзацы (разделитель - 2+ переноса строки)
paragraphs = re.split(r'\n{2,}', content)
# Формируем сообщения, сохраняя абзацы
messages = []
current_message = ""
for paragraph in paragraphs:
paragraph = paragraph.strip()
if not paragraph:
continue
# Если абзац слишком длинный, разбиваем по словам
if len(paragraph) > 4000:
chunks = split_long_paragraph(paragraph)
for chunk in chunks:
# Добавляем два переноса для сохранения форматирования
chunk_text = chunk + "\n\n"
# Проверяем, влезет ли абзац в текущее сообщение
if len(current_message) + len(chunk_text) > 4000 and current_message:
messages.append(current_message.strip())
current_message = ""
current_message += chunk_text
else:
# Добавляем два переноса для сохранения форматирования
paragraph_text = paragraph + "\n\n"
# Проверяем, влезет ли абзац в текущее сообщение
if len(current_message) + len(paragraph_text) > 4000 and current_message:
messages.append(current_message.strip())
current_message = ""
current_message += paragraph_text
# Добавляем последнее сообщение
if current_message.strip():
messages.append(current_message.strip())
# Отправляем сформированные сообщения
for i, message in enumerate(messages, 1):
logger.info(f"? Отправка части {i}/{len(messages)} ({len(message)} символов)")
await bot_client.send_message(
entity, message, link_preview=False, parse_mode='markdown'
)
await asyncio.sleep(1) # Задержка между сообщениями
except Exception as e:
logger.exception("Ошибка при отправке текста в канал")
raise
finally:
await bot_client.disconnect()
async def main():
"""Главная функция, объединяющая все этапы"""
logger.info(" Запуск полного процесса")
# Проверка обязательных переменных
if not all([API_ID, API_HASH, BOT_TOKEN]):
logger.error(" Отсутствуют обязательные переменные: API_ID, API_HASH или BOT_TOKEN")
return
try:
# 1. Скачивание PDF
pdf_path = download_pdf()
# 2. Конвертация PDF в текст
txt_path = pdf_to_text(pdf_path)
# 3. Обработка текста с помощью LLM
result_path = process_text()
# 4. Отправка результата в Telegram
with open(result_path, 'r', encoding='utf-8') as f:
content = f.read()
if content:
logger.info(f"? Подготовка текстового содержимого ({len(content)} символов)")
await send_to_channel(content)
else:
logger.warning(" Файл result.txt пуст")
except Exception as e:
logger.exception(" Произошла ошибка в процессе выполнения")
if __name__ == "__main__":
asyncio.run(main())
Запуск в облаке для работы 24/7
Запуск мы произведём в Amvera. Это проще, чем использование VPS, за счёт встроенного CI/CD (можно развернуть перетягиванием файлов в интерфейсе, или через Git). Помимо этого не требует иностранной карты для использования LLM.
Развёртывание этого конвейера на Amvera состоит из нескольких шагов.
1. Подготовка репозитория
Ваш репозиторий должен содержать все четыре файла скриптов и файл requirements.txt.
Содержимое requirements.txt:
requests==2.31.0
beautifulsoup4==4.12.2
PyMuPDF==1.23.8
aiohttp==3.9.1
telethon==1.33.1
python-dotenv==1.0.0
pytz==2023.3.post1
aiogram
requests
httpx==0.27.22. Настройка переменных окружения
В панели управления Amvera в разделе настроек вашего сервиса необходимо задать все переменные:
LLM_API_KEY (обязательно)
API_ID (обязательно)
API_HASH (обязательно)
BOT_TOKEN (обязательно)
CHANNEL_LINK (куда будем отправлять результат)
BOT_SESSION_STRING (опционально)
DATA_FOLDER (по умолчанию /data)
3. Сборка и деплой
После настройки переменных и расписания:
Нажмите «Собрать проект». Amvera установит все зависимости из
requirements.txt.После успешной сборки проект будет запущен.
Отслеживайте работу и ошибки в реальном времени в разделе «Логи».
Итоговое сообщение с результатом анализа
В результате в папке data должно выглядеть следующим образом:
Заключение
Мы создали шаблон для системы мониторинга финансовой отчётности. Он демонстрирует технологии связки классического веб-скрейпинга, парсинга PDF и современных мощных языковых моделей для создания полностью автоматизированной системы анализа и публикации данных.
Такой подход можно адаптировать для множества задач: мониторинг изменений в законодательных документах, анализ ежегодных отчётов компаний, создание дайджестов новостей из различных источников.
Надеюсь, этот пример вдохновит вас на создание собственных мощных и умных автоматизаций.
Исходный код проекта, вы можете найти на GitHub по ссылке.
crushilov
Спасибо за статью, давно думал в эту сторону.
Подскажите, пожалуйста, на сколько качественные данные в результате получаете?