SavePearlHarbor

Агрегация и парсинг XML RSS ленты на Python

—

от автора

admin

В этой статье рассмотрим, как с помощью 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 logginglogging.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 requestsHEADERS = {'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 ETfrom email.utils import parsedate_to_datetimedef 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 TelegramClientfrom telethon.sessions import StringSessionasync 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 osimport requestsimport xml.etree.ElementTree as ETfrom datetime import datetime, timedeltaimport pytzfrom email.utils import parsedate_to_datetimeimport loggingimport asynciofrom telethon import TelegramClientfrom telethon.sessions import StringSessionimport 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 configurationAPI_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.utcPERIOD = 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_runasync 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.4pytz==2025.7.post1telethon==1.35.0python-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-канал.

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

ссылка на оригинал статьи https://habr.com/ru/articles/947318/