[PYTHON]

aratana

スクリーンショット 2019-12-09 12.59.19.png

Es ist der 9. Tag des aratana Adventskalenders 2019. Die Firma Slack hat einen Kanal für den Adventskalender, und wenn er im Aratana-Adventskalender veröffentlicht wird, versuche ich, Slack zu benachrichtigen. (Ich habe es zum Zeitpunkt des Advents vor einem Jahr im Abstand von einer Stunde gemacht, aber ich werde es neu schreiben und den Code einfügen. Hauptvorgang 1. Benachrichtigen Sie den "Titel, das Datum, die URL" des veröffentlichten Artikels 2. Überprüfen Sie den gesamten Zeitraum und benachrichtigen Sie den veröffentlichten Artikel (da er möglicherweise zu einem späteren Zeitpunkt veröffentlicht wird). 3. Benachrichtigen Sie jeden Beitrag

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:
            #Wenn nicht gebucht
            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

        #Benachrichtigt wird übersprungen
        if entry.date_str in self.date_cache:
            return False

        return True

Verwenden

Geben Sie die erforderlichen Werte wie unten gezeigt ein, führen Sie sie aus, und Sie werden über Slack informiert!

url = "https://qiita.com/advent-calendar/2019/aratana"
cache_path = "./list.json"
title = "aratana Adventskalender Benachrichtigung über neue Beiträge"

slack_notification = SlackNotification(
    [INCOMING_WEBHOOK_Geben Sie die URL ein!]
)

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

Wenn Sie slack_notification in PrintNotification () ändern, können Sie den Vorgang ohne Slack-Benachrichtigung überprüfen!

Recommended Posts