coordinator.py

Go to the documentation of this file.

"""Data update coordinator for RSS/Atom feeds."""
 
from __future__ import annotations
 
from calendar import timegm
from datetime import datetime
import html
from logging import getLogger
from time import gmtime, struct_time
from typing import TYPE_CHECKING
from urllib.error import URLError
 
import feedparser
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.storage import Store
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
 
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, EVENT_FEEDREADER
 
DELAY_SAVE = 30
STORAGE_VERSION = 1
 
 
_LOGGER = getLogger(__name__)
 
 
class FeedReaderCoordinator(
    DataUpdateCoordinator[list[feedparser.FeedParserDict] | None]
):
    """Abstraction over Feedparser module."""
 
    config_entry: ConfigEntry
 
    def __init__(
        self,
        hass: HomeAssistant,
        url: str,
        max_entries: int,
        storage: StoredData,
    ) -> None:
        """Initialize the FeedManager object, poll as per scan interval."""
        super().__init__(
            hass=hass,
            logger=_LOGGER,
            name=f"{DOMAIN} {url}",
            update_interval=DEFAULT_SCAN_INTERVAL,
        )
        self.urlurl = url
        self.feed_authorfeed_author: str | None = None
        self.feed_versionfeed_version: str | None = None
        self._max_entries_max_entries = max_entries
        self._storage_storage = storage
        self._last_entry_timestamp_last_entry_timestamp: struct_time | None = None
        self._event_type_event_type = EVENT_FEEDREADER
        self._feed_feed: feedparser.FeedParserDict | None = None
        self._feed_id_feed_id = url
 
    @callback
    def _log_no_entries(self) -> None:
        """Send no entries log at debug level."""
        _LOGGER.debug("No new entries to be published in feed %s", self.urlurl)
 
    async def _async_fetch_feed(self) -> feedparser.FeedParserDict:
        """Fetch the feed data."""
        _LOGGER.debug("Fetching new data from feed %s", self.urlurl)
 
        def _parse_feed() -> feedparser.FeedParserDict:
            return feedparser.parse(
                self.urlurl,
                etag=None if not self._feed_feed else self._feed_feed.get("etag"),
                modified=None if not self._feed_feed else self._feed_feed.get("modified"),
            )
 
        feed = await self.hasshass.async_add_executor_job(_parse_feed)
 
        if not feed:
            raise UpdateFailed(f"Error fetching feed data from {self.url}")
 
        # The 'bozo' flag really only indicates that there was an issue
        # during the initial parsing of the XML, but it doesn't indicate
        # whether this is an unrecoverable error. In this case the
        # feedparser lib is trying a less strict parsing approach.
        # If an error is detected here, log warning message but continue
        # processing the feed entries if present.
        if feed.bozo != 0:
            if isinstance(feed.bozo_exception, URLError):
                raise UpdateFailed(
                    f"Error fetching feed data from {self.url} : {feed.bozo_exception}"
                )
 
            # no connection issue, but parsing issue
            _LOGGER.warning(
                "Possible issue parsing feed %s: %s",
                self.urlurl,
                feed.bozo_exception,
            )
        return feed
 
    async def async_setup(self) -> None:
        """Set up the feed manager."""
        feed = await self._async_fetch_feed_async_fetch_feed()
        self.loggerlogger.debug("Feed data fetched from %s : %s", self.urlurl, feed["feed"])
        if feed_author := feed["feed"].get("author"):
            self.feed_authorfeed_author = html.unescape(feed_author)
        self.feed_versionfeed_version = feedparser.api.SUPPORTED_VERSIONS.get(feed["version"])
        self._feed_feed = feed
 
    async def _async_update_data(self) -> list[feedparser.FeedParserDict] | None:
        """Update the feed and publish new entries to the event bus."""
        assert self._feed_feed is not None
        # _last_entry_timestamp is not set during async_setup, but we have already
        # fetched data, so we can use them, instead of fetch again
        if self._last_entry_timestamp_last_entry_timestamp:
            self._feed_feed = await self._async_fetch_feed_async_fetch_feed()
 
        # Using etag and modified, if there's no new data available,
        # the entries list will be empty
        _LOGGER.debug(
            "%s entri(es) available in feed %s",
            len(self._feed_feed.entries),
            self.urlurl,
        )
        if not self._feed_feed.entries:
            self._log_no_entries_log_no_entries()
            return None
 
        if TYPE_CHECKING:
            assert isinstance(self._feed_feed.entries, list)
 
        self._filter_entries_filter_entries()
        self._publish_new_entries_publish_new_entries()
 
        _LOGGER.debug("Fetch from feed %s completed", self.urlurl)
 
        if self._last_entry_timestamp_last_entry_timestamp:
            self._storage_storage.async_put_timestamp(self._feed_id_feed_id, self._last_entry_timestamp_last_entry_timestamp)
 
        return self._feed_feed.entries
 
    @callback
    def _filter_entries(self) -> None:
        """Filter the entries provided and return the ones to keep."""
        assert self._feed_feed is not None
        if len(self._feed_feed.entries) > self._max_entries_max_entries:
            _LOGGER.debug(
                "Processing only the first %s entries in feed %s",
                self._max_entries_max_entries,
                self.urlurl,
            )
            self._feed_feed.entries = self._feed_feed.entries[0 : self._max_entries_max_entries]
 
    @callback
    def _update_and_fire_entry(self, entry: feedparser.FeedParserDict) -> None:
        """Update last_entry_timestamp and fire entry."""
        # Check if the entry has a updated or published date.
        # Start from a updated date because generally `updated` > `published`.
        if time_stamp := entry.get("updated_parsed") or entry.get("published_parsed"):
            self._last_entry_timestamp_last_entry_timestamp = time_stamp
        else:
            _LOGGER.debug(
                "No updated_parsed or published_parsed info available for entry %s",
                entry,
            )
        entry["feed_url"] = self.urlurl
        self.hasshass.bus.async_fire(self._event_type_event_type, entry)
        _LOGGER.debug("New event fired for entry %s", entry.get("link"))
 
    @callback
    def _publish_new_entries(self) -> None:
        """Publish new entries to the event bus."""
        assert self._feed_feed is not None
        new_entry_count = 0
        firstrun = False
        self._last_entry_timestamp_last_entry_timestamp = self._storage_storage.get_timestamp(self._feed_id_feed_id)
        if not self._last_entry_timestamp_last_entry_timestamp:
            firstrun = True
            # Set last entry timestamp as epoch time if not available
            self._last_entry_timestamp_last_entry_timestamp = dt_util.utc_from_timestamp(0).timetuple()
        # locally cache self._last_entry_timestamp so that entries published at identical times can be processed
        last_entry_timestamp = self._last_entry_timestamp_last_entry_timestamp
        for entry in self._feed_feed.entries:
            if firstrun or (
                (
                    time_stamp := entry.get("updated_parsed")
                    or entry.get("published_parsed")
                )
                and time_stamp > last_entry_timestamp
            ):
                self._update_and_fire_entry_update_and_fire_entry(entry)
                new_entry_count += 1
            else:
                _LOGGER.debug("Already processed entry %s", entry.get("link"))
        if new_entry_count == 0:
            self._log_no_entries_log_no_entries()
        else:
            _LOGGER.debug("%d entries published in feed %s", new_entry_count, self.urlurl)
 
 
class StoredData:
    """Represent a data storage."""
 
    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize data storage."""
        self._data_data: dict[str, struct_time] = {}
        self.hasshass = hass
        self._store: Store[dict[str, str]] = Store(hass, STORAGE_VERSION, DOMAIN)
        self.is_initializedis_initialized = False
 
    async def async_setup(self) -> None:
        """Set up storage."""
        if (store_data := await self._store.async_load()) is not None:
            # Make sure that dst is set to 0, by using gmtime() on the timestamp.
            self._data_data = {
                feed_id: gmtime(datetime.fromisoformat(timestamp_string).timestamp())
                for feed_id, timestamp_string in store_data.items()
            }
        self.is_initializedis_initialized = True
 
    def get_timestamp(self, feed_id: str) -> struct_time | None:
        """Return stored timestamp for given feed id."""
        return self._data_data.get(feed_id)
 
    @callback
    def async_put_timestamp(self, feed_id: str, timestamp: struct_time) -> None:
        """Update timestamp for given feed id."""
        self._data_data[feed_id] = timestamp
        self._store.async_delay_save(self._async_save_data_async_save_data, DELAY_SAVE)
 
    @callback
    def _async_save_data(self) -> dict[str, str]:
        """Save feed data to storage."""
        return {
            feed_id: dt_util.utc_from_timestamp(timegm(struct_utc)).isoformat()
            for feed_id, struct_utc in self._data_data.items()
        }