16.09.2025 09:17
VadimMichaylov 0 237 Источник

В этой статье рассмотрим, как с помощью Python собирать и обрабатывать новости с сайта, имеющего RSS.

RSS – это простой XML-формат, в котором содержатся заголовки, описания, ссылки и даты публикации. Современные сайты часто предоставляют RSS-ленты для удобного чтения новостей в сторонних приложениях. Благодаря этому практически любой язык программирования позволяет «подписаться» на поток новостей и обрабатывать его по своему усмотрению.

В нашей статье мы создадим скрипт на Python, который за заданный период (например, за последние 4 часа) соберёт все записи из нескольких лент сайта BBC, отфильтрует их по ключевому слову «Трамп» и опубликует итоговый подбор в наш Telegram-канал. Далее рассмотрим код, вы легко сможете адаптировать его под любую другую ленту или ключевое слово.

Всё это мы развернём в облаке Amvera, заполнив всего несколько полей конфигурации, и выполнив три команды в IDE.

В примере кода все персональные данные для доступа к Telegram (API_ID, API_HASH, BOT_TOKEN и ссылки на канал) заменены на демонстрационные.

Структура проекта и зависимости

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

  • requests
    Для выполнения HTTP-запросов к RSS-лентам (загрузка XML).

  • xml.etree.ElementTree
    Встроенный в Python модуль для разбора XML-документов.

  • pytz
    Работа с часовыми поясами и приведением дат к единому стандарту (UTC).

  • email.utils.parsedate_to_datetime
    Преобразование строки pubDate из RSS в объект datetime.

  • telethon
    Асинхронный клиент для работы с Telegram API: авторизация, отправка сообщений.

  • asyncio
    Организация асинхронного кода для последовательной загрузки лент и отправки сообщений.

  • logging
    Логи важны для понимания, на каком этапе что-то пошло не так.

Конфигурация и окружение

Все параметры работы скрипта вынесены в начало файла – это удобно для настройки без правки логики:

DATA_FOLDER = '/data'  # Папка для хранения результатов, если нужно

RSS_FEEDS = [
    'https://feeds.bbci.co.uk/news/rss.xml',
    'https://feeds.bbci.co.uk/news/world/rss.xml',
    'https://feeds.bbci.co.uk/news/business/rss.xml',
    'https://feeds.bbci.co.uk/news/technology/rss.xml',
]

# Период фильтрации: последние 4 часа
PERIOD = timedelta(hours=4)
TIMEZONE = pytz.utc  # Telegram (в примере данные фиктивные – замените на свои)

API_ID = int(os.getenv('API_ID', 1234567))
API_HASH = os.getenv('API_HASH', 'ваш_api_hash')
BOT_TOKEN = os.getenv('BOT_TOKEN', 'ваш_bot_token')
CHANNEL_LINK = os.getenv('CHANNEL_LINK', 'https://t.me/ваш\\_канал')
BOT_SESSION_STRING = os.getenv('BOT_SESSION_STRING')  # session string для Telethon

# HTTP-заголовки, чтобы сервер думал, что к нему заходит обычный браузер
HEADERS = {'User-Agent': 'Mozilla/5.0 (compatible; BBCNewsScraper/1.0)'}

DATA_FOLDER — куда можно сохранять скачанный XML или логи (при желании расширить функционал).

RSS_FEEDS — список URL‑адресов фидов BBC. Вы можете добавить свои собственные или оставить только нужные.

PERIOD и TIMEZONE — определяют, за какой промежуток времени мы собираем новости и в каком часовом поясе их сравниваем.

Перед запуском на Amvera (или в любом другом месте) нужно настроить реальные значения через веб-интерфейс или CLI Amvera.

Логирование

Чтобы понимать, что происходит внутри нашего скрипта на удалённом сервере, мы подключаем стандартный модуль logging. Это позволит не «тонуть» в стене вывода – все важные события будут сохранены в логи.

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger('bbc_news_scraper')

  • level=logging.INFO — будем получать все сообщения уровня INFO и выше (WARNING, ERROR).

  • format='%(asctime)s - %(levelname)s - %(message)s' — в каждой строке лога будет время, уровень и текст сообщения.

  • logger = logging.getLogger('bbc_news_scraper') — именованный логгер, чтобы при масштабировании проекта можно было разделять логи по модулям.

Загрузка RSS-листа

Основная задача парсера — получить XML по URL-у RSS. Здесь на помощь приходит библиотека requests:

import requests

HEADERS = {'User-Agent': 'Mozilla/5.0 (compatible; BBCNewsScraper/1.0)'}

def fetch_feed_xml(url):
    """
    Загружает RSS-фид по HTTP и возвращает сырое содержимое XML.
    В случае ошибки логируем ее и возвращаем None.
    """
    try:
        resp = requests.get(url, headers=HEADERS, timeout=10)
        resp.raise_for_status()  # бросит исключение, если статус ≠200
        return resp.content
    except requests.RequestException as e:
        logger.error(f"Ошибка получения {url}: {e}")
        return None

  • timeout=10 защищает от «зависших» запросов.

  • resp.raise_for_status() сразу «поймает» HTTP-ошибки (404, 500 и т. д.).

При любом исключении мы логируем сообщение через logger.error, чтобы в логах было видно, какой именно фид не отвечает.

Парсинг RSS и фильтрация по времени и ключевому слову

После того как мы получили «сырое» содержимое RSS, нужно из него вытащить только те элементы, которые нам интересны — то есть статьи последних 4 часов со словом «Трамп» в заголовке. Для этого используем xml.etree.ElementTree и стандартную утилиту для разбора дат.

import xml.etree.ElementTree as ET
from email.utils import parsedate_to_datetime

def parse_feed_items(xml_content: bytes, cutoff: datetime, keyword: str) -> list[dict]:
    """
    Разбирает XML, извлекает  и фильтрует по дате и ключевому слову.
    Возвращает список словарей {'title', 'link', 'published'}.
    """
    items = []
    try:
        root = ET.fromstring(xml_content)
    except ET.ParseError as e:
        logger.error(f"Ошибка разбора XML: {e}")
        return items

    # Ищем все элементы 
    for item in root.findall('.//item'):
        title_el = item.find('title')
        link_el = item.find('link')
        pubdate_el = item.find('pubDate')

        if not (title_el and link_el and pubdate_el):
            continue

        # Преобразуем pubDate → datetime с часовым поясом UTC
        try:
            pub_dt = parsedate_to_datetime(pubdate_el.text)
            pub_dt = pub_dt.astimezone(TIMEZONE)
        except Exception as e:
            logger.warning(f"Не удалось распарсить дату: {e}")
            continue

        # Отфильтруем по времени и по слову “Трамп”
        if pub_dt >= cutoff and keyword.lower() in title_el.text.lower():
            items.append({
                'title': title_el.text.strip(),
                'link': link_el.text.strip(),
                'published': pub_dt.strftime('%Y-%m-%d %H:%M:%S %Z')
            })
    return items

Мы ищем все `` внутри RSS и проверяем, что у каждого есть заголовок, ссылка и дата.

С помощью parsedate_to_datetime из email.utils конвертируем строку вида Fri, 07 Aug 2025 12:34:56 GMT в datetime с поясом UTC.

Сравниваем дату публикации с cutoff (текущим временем минус 4 ч).

Дополнительноеусловие — в заголовке должно встречаться слово «Трамп» (без учета регистра).

Так мы получаем именно те новости, которые свежие и релевантные нашей теме.

Отправка сообщений в Telegram

Когда список свежих статей готов, формируем сообщение и отправляем его в ваш канал с помощью Telethon:

from telethon import TelegramClient
from telethon.sessions import StringSession

async def send_to_telegram_channel(message: str):
    """
    Асинхронно отправляет текст в Telegram-канал.
    Все личные данные (API_ID, API_HASH, BOT_TOKEN, SESSION) в примере заменены.
    """
    try:
        bot_client = TelegramClient(
            StringSession(BOT_SESSION_STRING) if BOT_SESSION_STRING else 'session_bot',
            API_ID, API_HASH
        )
        await bot_client.start(bot_token=BOT_TOKEN)
        entity = await bot_client.get_entity(CHANNEL_LINK)
        # link_preview=False – чтобы не подтягивались большие картинки
        await bot_client.send_message(entity, message, link_preview=False)
        logger.info("Сообщение успешно отправлено в Telegram")
    except Exception as e:
        logger.error(f"Ошибка при отправке в Telegram: {e}")
    finally:
        if bot_client.is_connected():
            await bot_client.disconnect()

  • StringSession или обычная сессия — хранит информацию о вашем боте.

  • start(bot_token=...) делает авторизацию по токену.

  • get_entity автоматически находит нужный канал по ссылке.

Полный код (скелет скрипта)

import os
import requests
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta
import pytz
from email.utils import parsedate_to_datetime
import logging
import asyncio
from telethon import TelegramClient
from telethon.sessions import StringSession
import time

# ========== CONFIGURATION ==========
DATA_FOLDER = '/data'
RSS_FEEDS = [
    'https://feeds.bbci.co.uk/news/rss.xml',
    'https://feeds.bbci.co.uk/news/world/rss.xml',
    'https://feeds.bbci.co.uk/news/business/rss.xml',
    'https://feeds.bbci.co.uk/news/technology/rss.xml',
]

# Telegram configuration
API_ID = int(os.getenv('API_ID', 2122521))
API_HASH = os.getenv('API_HASH', 'd27621d9925562342baefc013e')
BOT_TOKEN = os.getenv('BOT_TOKEN', '759406329:32522CPhmU')
CHANNEL_LINK = os.getenv('CHANNEL_LINK', 'https://t.me/+vR323w55y')
BOT_SESSION_STRING = os.getenv('BOT_SESSION_STRING')

HEADERS = {'User-Agent': 'Mozilla/5.0 (compatible; BBCNewsScraper/1.0)'}
TIMEZONE = pytz.utc
PERIOD = timedelta(hours=4) # Период - 4 часа
MOSCOW_TZ = pytz.timezone('Europe/Moscow')

# ========== LOGGING ==========
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('bbc_news_scraper')

# ========== UTILS ==========
def ensure_dir(path):
    os.makedirs(path, exist_ok=True)

# ========== FETCH FEED ==========
def fetch_feed_xml(url):
    try:
        resp = requests.get(url, headers=HEADERS, timeout=10)
        resp.raise_for_status()
        return resp.content
    except requests.RequestException as e:
        logger.error(f"Ошибка получения {url}: {e}")
        return None

# ========== PARSE FEED ITEMS ==========
def parse_feed_items(xml_content, cutoff):
    items = []
    try:
        root = ET.fromstring(xml_content)
    except ET.ParseError as e:
        logger.error(f"Ошибка разбора XML: {e}")
        return items

    for item in root.findall('.//item'):
        title_el = item.find('title')
        link_el = item.find('link')
        pubdate_el = item.find('pubDate')

        if title_el is None or link_el is None or pubdate_el is None:
            continue

        try:
            pub_dt = parsedate_to_datetime(pubdate_el.text)
            pub_dt = pub_dt.astimezone(TIMEZONE)
        except Exception:
            continue

        if pub_dt >= cutoff:
            items.append({
                'title': title_el.text.strip(),
                'link': link_el.text.strip(),
                'published': pub_dt.strftime('%Y-%m-%d %H:%M:%S %Z')
            })
    return items

# ========== TELEGRAM SENDER ==========
async def send_to_telegram_channel(message):
    """Отправка сообщения в телеграм-канал"""
    try:
        bot_client = TelegramClient(
            StringSession(BOT_SESSION_STRING) if BOT_SESSION_STRING else 'session_bot',
            API_ID, API_HASH
        )
        await bot_client.start(bot_token=BOT_TOKEN)
        entity = await bot_client.get_entity(CHANNEL_LINK)
        await bot_client.send_message(entity, message, link_preview=False)
        logger.info("Сообщение отправлено в Telegram")
    except Exception as e:
        logger.error(f"Ошибка отправки в Telegram: {e}")
    finally:
        if bot_client.is_connected():
            await bot_client.disconnect()

# ========== MAIN ==========
async def main():
    logger.info("Запуск сбора новостей BBC за последние 4 часа")
    now = datetime.now(TIMEZONE)
    cutoff = now - PERIOD
    news_items = []

    # Сбор новостей из всех RSS-лент
    for url in RSS_FEEDS:
        xml = fetch_feed_xml(url)
        if not xml:
            continue
        parsed = parse_feed_items(xml, cutoff)
        logger.info(f"{url}: найдено {len(parsed)} статей за последние 4 часа")
        news_items.extend(parsed)

    # Формирование сообщения для Telegram
    if news_items:
        message = "? *Последние новости BBC за 4 часа:*

"
        for item in news_items:
            message += f"• [{item['title']}]({item['link']})
"
        # Отправка сообщения в Telegram
        await send_to_telegram_channel(message)
        logger.info(f"Отправлено {len(news_items)} новостей в Telegram")
    else:
        logger.info("Нет новостей для отправки")
        await send_to_telegram_channel("ℹ️ За последние 4 часа новостей от BBC не обнаружено")

def calculate_next_run():
    """Вычисляет время следующего запуска (00:00, 04:00, 08:00 и т.д. по Москве)"""
    now = datetime.now(MOSCOW_TZ)
    current_hour = now.hour
    # Вычисляем ближайший час, кратный 4
    target_hour = ((current_hour // 4) * 4 + 4) % 24
    next_run = now.replace(hour=target_hour, minute=0, second=0, microsecond=0)
    # Если ближайшее время уже прошло сегодня, планируем на завтра
    if next_run < now:
        next_run += timedelta(days=1)
    logger.info(f"Следующий запуск в: {next_run.strftime('%Y-%m-%d %H:%M:%S %Z%z')}")
    return next_run

async def scheduler():
    """Планировщик, запускающий задачу каждые 4 часа"""
    while True:
        next_run = calculate_next_run()
        sleep_seconds = (next_run - datetime.now(MOSCOW_TZ)).total_seconds()
        logger.info(f"Ожидание {sleep_seconds / 3600:.2f} часов до следующего запуска")
        await asyncio.sleep(sleep_seconds)
        try:
            await main()
        except Exception as e:
            logger.error(f"Ошибка в основном задании: {e}")

if __name__ == '__main__':
    asyncio.run(scheduler())

Запуск скрипта в Amvera

Самое время развернуть код на удаленном сервере. Скрипт мы запустим в Amvera, это даст нам возможность простых обновлений проекта на сервере через встроенный CI/CD одной командой, бесплатное логирование с семантическим поиском и возможность не платить за остановленные проекты.

Развертывание

  1. Регистрация и подготовка аккаунта: перейдите по ссылке, зарегистрируйтесь и подтвердите почту. При регистрации обычно бонусы на баланс в размере 111 руб. Их нам хватит на тест.

  2. Создание нового проекта: в консоли Amvera/Проекты/Новый проект. Тип сервиса: Приложение.

  3. Загрузка кода в репозиторий проекта: откройте вкладку Репозиторий и загрузите main.py (или ваш основной скрипт) и requirements.txt.

requirements.txt пример:

requests==2.32.4
pytz==2025.7.post1
telethon==1.35.0
python-dotenv==1.1.1

Настройка сборки и команды запуска

  • В Конфигурации укажите окружение Python.

  • В поле scriptName выберите main.py (или имя вашего скрипта).

  • Создайте переменные окружения (API_ID, API_HASH, BOT_TOKEN, CHANNEL_LINK, BOT_SESSION_STRING и т.д.) через интерфейс Amvera.

Сборка и запуск

  • Нажмите «Сохранить» и затем «Собрать проект».

Пример лога запущенного сервиса
Пример лога запущенного сервиса

  • После успешной сборки Amvera автоматически запустит ваш скрипт. Проверить логи можно в разделе Логи.

Пример получаемых по RSS новостей
Пример получаемых по RSS новостей

Заключение

В этой статье мы показали, как легко и быстро настроить сбор новостей из RSS-лент BBC с фильтрацией по ключевому слову и автоматическую отправку в Telegram-канал.

Надеюсь, этот пример вдохновит вас на собственные проекты по мониторингу новостей или любого другого контента в сети.

Комментарии (0)