API Reference

class Watchr(BaseModel):
    """Central class that orchestrates the polling, comparing, and alerting process.

    Attributes:
        alerter: The instance is called with the scraped text and is responsible to send out the alert.
        comparer: The instance is called with the scraped text and is responsible to compare the text with
            the previous state to check if there are any changes and an alert is necessary.
    """

    comparer: AbstractComparer = Field(default_factory=FSComparer)
    alerter: AbstractAlerter = Field(default_factory=PrintAlerter)

    _poller: Callable[[Playwright], str] | None = PrivateAttr(default=None)

    @property
    def poller(self) -> Callable[[Playwright], str]:
        """The poller function that scrapes the text from the website."""
        if self._poller is None:
            message = "Please set a poller function using the `set_poller` decorator."
            logger.error(message)
            raise ValueError(message)
        return self._poller

    def set_poller(
        self,
        f: Callable[[Playwright], str],
    ) -> Callable[[Playwright], str]:
        """Decorator to set the poller function.

        The wrapper returns the function unchanged, but stores it in the `_poller` attribute.

        Args:
            f: A function that takes a `Playwright` instance and returns the scraped text.

        Returns:
            The function that was passed in.
        """
        self._poller = f
        return f

    def __call__(
        self,
    ) -> None:
        """Main method that orchestrates the polling, comparing, and alerting process.

        When called the poller function is called to scrape the text from the website. The text is then
        compared to the previous state. If there are changes, an alert is sent out.
        """
        with sync_playwright() as playwright:
            result = self.poller(playwright)
            logger.debug(result)

        if self.comparer(result) == Status.NO_CHANGES:
            logger.info("No changes detected.")
            return

        logger.info("Changes detected! Sending alert.")
        self.alerter(result)

class Status(Enum):
    """Enum representing the status of a comparison.

    Convenience enum because I got confused with booleans.

    Attributes:
        CHANGED: The text has changed.
        NO_CHANGES: The text has not changed.
    """

    CHANGED = auto()
    NO_CHANGES = auto()

class AbstractComparer(BaseModel, ABC):
    """Decide if a text has changed."""

    @abstractmethod
    def __call__(self, text: str) -> Status:
        """Decide if a text has changed.

        A new comparer must implement this method. The method should compare the text with the previous
        state and return the status of the comparison.
        """
        ...

class DummyComparer(AbstractComparer):
    """A comparer that always returns [`Status.CHANGED`][web_watchr.compare.Status]."""

    def __call__(self, text: str) -> Status:
        """Always return [`Status.CHANGED`][web_watchr.compare.Status]."""
        return Status.CHANGED

class FSComparer(AbstractComparer):
    """A comparer that compares the text with the content of a file.

    Attributes:
        cache_dir: The directory where the cache file is stored. Defaults to `~/.local/share/web_watchr/cache`.
        send_on_missing: Whether to send a notification if the cache file is missing.
        identifier: The identifier used for the cache file.
    """

    cache_dir: Path = Path("~/.local/share/web_watchr/cache").expanduser()
    send_on_missing: bool = False
    identifier: str = "fs_comparer"

    _cache_path: Path | None = None

    @property
    def cache_path(self) -> Path:
        """The path to the cache file.

        Creates the cache directory if it does not exist.
        """
        if self._cache_path is None:
            self._cache_path = self.cache_dir / f"{self.identifier}.txt"
            if not self.cache_dir.exists():
                logger.debug(f"Creating cache directory: {self.cache_dir}")
                self.cache_dir.mkdir(parents=True)

        return self._cache_path

    def __call__(self, text: str) -> Status:
        """Compare the text with the content of the cache file.

        Args:
            text: The text to compare with the cache file.

        Returns:
            [`Status.CHANGED`][web_watchr.compare.Status] if the text is different from the cache file.
            [`Status.NO_CHANGES`][web_watchr.compare.Status] if the text is the same as the cache file.
            If the cache file is missing, it returns [`Status.CHANGED`][web_watchr.compare.Status] if `send_on_missing` is `True`.
        """
        if not self.cache_path.exists():
            logger.debug(f"Cache file not found: {self.cache_path}. Saving new state.")
            self._save(new_state=text)
            return Status.CHANGED if self.send_on_missing else Status.NO_CHANGES

        if not self._equal_to_cache(new_state=text):
            self._save(new_state=text)
            return Status.CHANGED

        return Status.NO_CHANGES

    def _load(self) -> str:
        with open(self.cache_path, "r") as f:
            return f.read()

    def _equal_to_cache(self, *, new_state: str) -> bool:
        old_state = self._load()
        return old_state == new_state

    def _save(self, *, new_state: str):
        with open(self.cache_path, "w") as f:
            f.write(new_state)

class AbstractAlerter(BaseModel, ABC):
    """Send an alert."""

    @abstractmethod
    def __call__(self, text: str) -> None:
        """Send an alert.

        A new alerter must implement this method. The method should send the alert with the given text.

        Args:
            text: The text of the alert.
        """
        ...

class PrintAlerter(AbstractAlerter):
    """Print the alert to the standard output."""

    def __call__(self, text: str) -> None:
        """Print the alert to the standard output.

        Args:
            text: The text of the alert.
        """
        print(text)

class TelegramAlerter(AbstractAlerter):
    """Sends a message to a telegram chat.

    For this alerter to work, you need to
    [create a telegram bot and get its token](https://core.telegram.org/bots/tutorial#obtain-your-bot-token).
    Furthermore, the bot needs to be added to the chat you want to send messages to and you
    need to retrieve the chat ID of the chat (e.g., like [this](https://stackoverflow.com/a/32572159/9685500)).

    Attributes:
        token: The token of the telegram bot.
        chat_id: The chat ID to send the message to.
    """

    token: str
    chat_id: str

    _bot: Bot | None = None

    @property
    def bot(self) -> Bot:
        """Telegram bot instance"""
        if self._bot is None:
            self._bot = Bot(token=self.token)
        return self._bot

    def __call__(self, text: str) -> None:
        """Send a message to a telegram chat.

        Args:
            text: The text of the message.
        """
        logger.debug(f"Sending message '{text}' to telegram.")
        run(self.bot.send_message(chat_id=self.chat_id, text=text))