[PYTHON] Le calendrier de l'Avent a été publié! Notifier Slack

C'est le 9ème jour du Calendrier de l'Avent aratana 2019.

La société Slack a une chaîne pour le calendrier de l'avent, et quand il est publié sur le calendrier de l'avent aratana, j'essaie d'en informer Slack. (Intervalle d'une heure Je l'ai fait quand je l'avais il y a un an, mais je vais le réécrire et coller le code.

Opération principale

1. Notifier le "Titre, Date, URL" de l'article publié
2. Vérifiez toute la période et notifiez l'article publié (car il peut être publié à une date ultérieure)
3. Notifier chaque message

code

from abc import ABCMeta, abstractmethod
from pathlib import Path
from typing import List, Set

import requests
from bs4 import BeautifulSoup


class AdventCalendarEntry:
    def __init__(self, title: str, url: str, date_str: str) -> None:
        self.title = title
        self.url = url
        self.date_str = date_str

    def empty(self) -> bool:
        if self.title:
            return False

        return True


class AdventCalendarEntryList:
    ITEM_SELECTOR = ".adventCalendarItem"
    DATE_SELECTOR = ".adventCalendarItem_date"
    ENTRY_SELECTOR = ".adventCalendarItem_entry"

    def __init__(self, entry_list: List[AdventCalendarEntry]) -> None:
        self._entry_list = entry_list

    def __iter__(self):
        return iter(self._entry_list)

    @classmethod
    def from_html(cls, html_text: str) -> "AdventCalendarEntryList":
        bs = BeautifulSoup(html_text, "lxml")
        item_list_bs = bs.select(cls.ITEM_SELECTOR)

        entry_list = [cls._make_item_from_item_bs(item_bs) for item_bs in item_list_bs]

        return cls(entry_list)

    @staticmethod
    def _make_item_from_item_bs(item_bs):
        entry_date = item_bs.select_one(AdventCalendarEntryList.DATE_SELECTOR).text

        entry_bs = item_bs.select_one(AdventCalendarEntryList.ENTRY_SELECTOR)
        if not entry_bs:
            #Si non publié
            return AdventCalendarEntry(title="", url="", date_str=entry_date)

        _anchor = entry_bs.select_one("a")
        entry_title = _anchor.text
        entry_url = _anchor.get("href")

        return AdventCalendarEntry(
            title=entry_title, url=entry_url, date_str=entry_date
        )


class AdventCalendarDateCache:
    def __init__(self, cache_path: str) -> None:
        self.cache_path = Path(cache_path)
        self.cached_date_set = self.load()

    def load(self) -> Set[str]:
        if not Path(self.cache_path).exists():
            return set()

        with open(self.cache_path, "r") as f:
            checked_list = [line.strip() for line in f.readlines()]

        return set(checked_list)

    def save(self) -> None:
        with open(self.cache_path, "w") as f:
            lines = "\n".join(list(self.cached_date_set))
            f.writelines(lines)

    def add(self, date_str: str) -> None:
        self.cached_date_set.add(date_str)

    def __contains__(self, date_str: str) -> bool:
        return date_str in self.cached_date_set


class Notification(metaclass=ABCMeta):
    @abstractmethod
    def run(self, title: str, entry: AdventCalendarEntry) -> None:
        pass


class PrintNotification(Notification):
    def run(self, title: str, entry: AdventCalendarEntry) -> None:
        message = (
            f"{title}\n"
            f"＜{entry.date_str}＞\n"
            f"TITLE: {entry.title}\n"
            f"URL  : {entry.url}"
        )

        print(message)


class SlackNotification(Notification):
    def __init__(self, webhook_url: str) -> None:
        self.webhook_url = webhook_url

    def run(self, title: str, entry: AdventCalendarEntry) -> None:
        message = (
            f"<!here>\n"
            f"*{title}*\n"
            f"> ＜{entry.date_str}＞\n"
            f"> TITLE: {entry.title}\n"
            f"> URL  : {entry.url}"
        )

        payload = {"text": message, "as_user": True}

        r = requests.post(self.webhook_url, json=payload)
        r.raise_for_status()


class AdventCalendarNotify:
    def __init__(
        self, message_title: str, cache_path: str, notify: Notification
    ) -> None:
        self.message_title = message_title
        self.cache_path = cache_path
        self.notify = notify

        self.date_cache = AdventCalendarDateCache(cache_path)

    def run(self, url: str) -> None:
        for entry in self.load_entry_list(url):
            if not self.new_entry(entry):
                continue

            self.notify.run(self.message_title, entry)
            self.date_cache.add(entry.date_str)

        self.date_cache.save()

    def load_entry_list(self, url: str) -> List[AdventCalendarEntry]:
        r = requests.get(url)
        r.raise_for_status()

        return list(AdventCalendarEntryList.from_html(r.text))

    def new_entry(self, entry: AdventCalendarEntry) -> bool:
        if entry.empty():
            return False

        #La notification est ignorée
        if entry.date_str in self.date_cache:
            return False

        return True

Utilisation

Entrez les valeurs requises comme indiqué ci-dessous, exécutez-le et vous serez informé de Slack!

url = "https://qiita.com/advent-calendar/2019/aratana"
cache_path = "./list.json"
title = "aratana Advent Calendar Notification de nouveaux messages"

slack_notification = SlackNotification(
    [INCOMING_WEBHOOK_Entrez l'URL!]
)

notify = AdventCalendarNotify(title, cache_path, slack_notification)
notify.run(url)

Si vous changez slack_notification en PrintNotification (), vous pouvez vérifier l'opération sans notification Slack!