私、ずっと弊社にいたのに、弊社のこと全然分かってあげられてなかった…。
「他の部署が面白いことやってるのに、社長が話してて初めて知った！」ってことが多かったので、 弊社のコーポレートサイトが更新したらChatworkに通知するbotを作りました。
ChatworkAPIに関してはこちらの記事を参考にしました。
本当はclassごとにファイルを分けるべきなんでしょうが、 jupyter notebookを使って作ったので1ファイルで作っちゃいました。
Python2.7とBeautifulSoup4を利用しました。 これをCronで一日おきにチェックすればいいかな。
#!/usr/bin/env python # -*- coding: utf-8 -*- import urllib import urllib2 import pycurl from bs4 import BeautifulSoup class heishaNewsScraper: def set_url(self, url): self.url = url def set_stop_path(self, path): self.stop_path = path def get_news_list(self): soup = self._url_to_soup(self.url) news_list = self._soup_to_list(soup) return news_list def read_stopfile(self): try: with open(self.stop_path, 'r') as f: saved_dates = f.read() return int(saved_dates.rstrip('\n')) except: return 0 # ストップファイルがない場合0を返すので、初回実行時には全部タレ流してしまうことに注意 # ストップファイル（現時点の最新記事の日付が20160101の形式で保存されているファイル）を保存する def save_stopfile(self, news_list): newest_date = self._get_newest_date(news_list) with open(self.stop_path, 'w') as f: f.write(str(newest_date)) def _url_to_soup(self, url): html = urllib2.urlopen(url) return BeautifulSoup(html) def _soup_to_list(self, soup): news_soups = self._get_news_soups(soup) return self._get_news_infos(news_soups) def _get_news_soups(self, soup): return soup.find(class_='***').find(class_='*******').findAll(class_='*****') def _get_news_infos(self, news_soups): return [self._get_news_info(n) for n in news_soups] # いろいろやった後に(日付, カテゴリー, タイトル, URL)を返す関数 def _get_news_info(self, li_soup): ret_list = [self._get_soup_string(li_soup.find(class_=c)) for c in ('date', 'cat')] ret_list[0] = self._rm_non_number(ret_list[0]) title_soup = li_soup.find(class_='title').find('a') title = self._get_soup_string(title_soup) url = self._make_url(title_soup.get('href')) ret_list.extend([title, url]) # 破壊的操作 return tuple(ret_list) # たまに他のディレクトリを参照しているURLがあったので対策 def _make_url(self, url): if 'http' in url: return url else: return '/'.join(self.url.split('/')[:3]) + url def _get_newest_date(self, news_tuple): dates = map(lambda x: x[0], news_tuple) sorted(dates, reverse=True) return dates[0] # .find(***)で対象のタグが見つからず、Noneタイプが返ってきている場合の対策 def _get_soup_string(self, soup): try: return soup.string except: return None # 比較しやすくするため、日付から数字以外の部分を取り除く def _rm_non_number(self, string): return int(''.join(c for c in string if c.isdigit())) class chatWorkBot: def set_roomid(self, roomid): self.roomid = roomid def set_token(self, token): self.token = token def make_message(self, message_tuple): ascii_message = u'弊社のニュースが更新されました\n' +\ str(message_tuple[0]) + ': ' + message_tuple[2] + '\n' +\ message_tuple[3] return unicode(ascii_message).encode('utf_8') # Python2系だと文字コードの処理が面倒だよね… def send_message(self, message): option = {u'body': message} c = pycurl.Curl() c.setopt(pycurl.URL, 'https://api.chatwork.com/v1/rooms/%s/messages'%str(self.roomid)) c.setopt(pycurl.HTTPHEADER, ['X-ChatWorkToken: %s'%self.token]) c.setopt(pycurl.POST, 1) # http://stackoverflow.com/questions/8332643/pycurl-and-ssl-cert # c.setopt(pycurl.SSL_VERIFYPEER, 0) # c.setopt(pycurl.SSL_VERIFYHOST, 0) c.setopt(pycurl.POSTFIELDS, urllib.urlencode(option)) c.perform() def main(): scraper = heishaNewsScraper() scraper.set_url('http://*************.***/****/') scraper.set_stop_path('./STOPFILE') news_list = scraper.get_news_list() saved_date = scraper.read_stopfile() ratest_news = [x for x in news_list if x[0] > saved_date] # ストップファイルに記録されている日付以降の情報をbotで投稿する bot = chatWorkBot() bot.set_roomid(*****) # chatworkの部屋のid bot.set_token('*****************') # Chatworkのtoken for n in ratest_news: message = bot.make_message(n) bot.send_message(message) # ストップファイルを更新する scraper.save_stopfile(news_list) if __name__ == '__main__': main()
