send_alrb_decisions.py

from typing import Dict, Set
import os
import re
import requests
from bs4 import BeautifulSoup
from sqlalchemy import create_engine, select, String, Column, MetaData
from sqlalchemy.orm import Session, declarative_base

CONNSTR = os.environ.get("DB_CONNECTION_STRING")
telegram_key = os.environ.get("TELEGRAM_KEY")

meta = MetaData(schema="windmill")
engine = create_engine(CONNSTR)
session = Session(engine)
Base = declarative_base(metadata=meta)


class Decisions(Base):
    __tablename__ = "alrb_decisions"

    decision_id = Column(String, primary_key=True)

    def __repr__(self) -> str:
        return f"Decision({self.decision_id!r})"


def escape_markdown(text: str, version: int = 1, entity_type: str = None) -> str:
    """Helper function to escape telegram markup symbols. From telegream-api-bot

    .. versionchanged:: 20.3
        Custom emoji entity escaping is now supported.

    Args:
        text (:obj:`str`): The text.
        version (:obj:`int` | :obj:`str`): Use to specify the version of telegrams Markdown.
            Either ``1`` or ``2``. Defaults to ``1``.
        entity_type (:obj:`str`, optional): For the entity types
            :tg-const:`telegram.MessageEntity.PRE`, :tg-const:`telegram.MessageEntity.CODE` and
            the link part of :tg-const:`telegram.MessageEntity.TEXT_LINK` and
            :tg-const:`telegram.MessageEntity.CUSTOM_EMOJI`, only certain characters need to be
            escaped in :tg-const:`telegram.constants.ParseMode.MARKDOWN_V2`. See the `official API
            documentation <https://core.telegram.org/bots/api#formatting-options>`_ for details.
            Only valid in combination with ``version=2``, will be ignored else.
    """
    if int(version) == 1:
        escape_chars = r"_*`["
    elif int(version) == 2:
        if entity_type in ["pre", "code"]:
            escape_chars = r"\`"
        elif entity_type in ["text_link", "custom_emoji"]:
            escape_chars = r"\)"
        else:
            escape_chars = r"\_*[]()~`>#+-=|{}.!"
    else:
        raise ValueError("Markdown version must be either 1 or 2!")

    return re.sub(f"([{re.escape(escape_chars)}])", r"\\\1", text)


def get_alrb_decisions() -> Dict[str, str]:
    """Go to the ALRB Decisions webpage, grab all the links, and return a list of the PDF decisions"""

    res = requests.get(
        "http://www.alrb.gov.ab.ca/decisions/decisions.html", allow_redirects=True
    )
    soup = BeautifulSoup(res.content, "html.parser")
    # %%
    decisions = {}
    for link in soup.find_all("a", attrs={"href": re.compile("pdf")}):
        name = link.parent.previous_sibling.previous_sibling.get_text()
        id = link.get("href")
        decisions[id] = name

    return decisions


def get_known_decisions() -> Set[str]:
    """Get all the PDFs that were already sent via Telegram API"""

    stmt = select(Decisions)

    return {decision_id for decision_id in session.scalars(stmt)}


def communicate_new_decisions() -> None:
    """Compare all PDFs that were already sent via Telegram with the ALRB and send the new ones"""

    alrb_decisions = get_alrb_decisions()
    known_decisions = {decision.decision_id for decision in get_known_decisions()}
    new_decisions = set(alrb_decisions.keys()) - known_decisions

    for decision in new_decisions:
        res = requests.post(
            "https://api.telegram.org/bot"
            + telegram_key
            + "/sendMessage?chat_id=-414892076",
            data={
                "text": f"ALRB Decision: [{0}](http://www.alrb.gov.ab.ca/decisions/{decision})".format(
                    escape_markdown(alrb_decisions[decision], 2)
                ),
                "parse_mode": "MarkdownV2",
            },
        )
        if res.status_code == 200:
            session.add(Decisions(decision_id=decision))
            session.commit()
        else:
            break


if __name__ == "__main__":
    communicate_new_decisions()