aratana Advent Calendar 2019の9日目です。
会社Slackにアドベントカレンダー用チャンネルがあり、aratana アドベントカレンダーに投稿されると、Slack通知するようにしてます。（一時間間隔
1年前のアドベントのときに作ったのですが、書き直してコード貼っつけておきます。
主な動作
- 投稿された記事の「タイトル, 日付, URL」を通知
- 全期間チェックして投稿された記事を通知（後日投稿される場合があるかもなので
- 一投稿ずつ通知
コード
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:
# 未投稿の場合
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
# 通知済みは、スキップ
if entry.date_str in self.date_cache:
return False
return True
利用
下記のように必要な値をいれて、実行すれば、Slack通知されます！
url = "https://qiita.com/advent-calendar/2019/aratana"
cache_path = "./list.json"
title = "aratanaアドベントカレンダー新規投稿の通知"
slack_notification = SlackNotification(
[INCOMING_WEBHOOK_URLを入れる！]
)
notify = AdventCalendarNotify(title, cache_path, slack_notification)
notify.run(url)
slack_notificationをところを、PrintNotification()に変更すれば、Slack通知せずに動作確認できます！