StarBugs Weekly ChatBot 開發筆記

smalltown

Published in

Starbugs Weekly 星巴哥技術專欄

11 min readFeb 16, 2022

Background

跟社群朋友一起負責維護的 StarBugs Weekly 現在有兩個網站:

https://weekly.starbugs.dev/: 負責刊載每週專欄文章 & 匯集一整週各領域推薦文章
https://recommend.starbugs.dev/: 負責發布每週各領域推薦文章

而以往我們幾個人每週輪流使用 MailChimp 寄出週刊文章，倒也還過得去，畢竟一週才只有一次，但後來將推薦文章拆分成獨立網站之後，這塊的訊息發布並沒有去做到，假如還是用 Email 的話，感覺一週會多好幾封信，大家現在使用 IM 服務的頻率都比 Email 多太多了，因此我就想說來寫個 Telegram ChatBot 負責整個 StarBug Weekly 的文章發布通知，想要做這一個 Side Project 很久了，但一直找不到時間去執行，剛好趁著這次過年長假完成了一個 MVP (Minimum Viable Product)

在這個發布 StarBugs Weekly 文章的 ChatBot 中，目前有三個比較重要的元件，將其列在底下，而接下來的文章會就這些部分再做更近一步的介紹，將自己的開發過程筆記下來

Telegram ChatBot: 負責發送文章給有訂閱的使用者
GitHub API Integration: 負責監看是否有新文章發布
Celery: 負責在背景處理這些工作任務

Telegram ChatBot

自己在工作中有一點點 ChatBot 的開發經驗，不過是在另外一個叫做 Slack 的平台，這是第一次自己嘗試開發 Telegram ChatBot，老實說 Slack 對於 ChatBot 這一塊上面來說，整個管理與開發生態系比 Telegram 完善一些，Telegram 在 ChatBot 這塊自己覺得算是麻雀雖小五臟俱全，在開始的第一個步驟很酷，要先去找 @BotFather 幫忙產生 ChatBot 出來，透過指令 /help 就可以得知他可以幫忙做哪一些事情

接著透過 /newbot 指令開始申請 ChatBot，會需要給這個機器人一個顯示名稱，例如這邊使用 StarBugs Weekly，還有一個正式的辨識 ID, 而且這個 ID 需要有 bot 結尾，申請成功後會獲得 API Token，下面便會透過這個 API Token 來跟 Telegram 的 API 進行互動

根據良好的工程師 DRY (Don’t repeat yourself) 習慣，我先上網找了一下 Telegram 有什麼好用的 Python 開發套件，果然立馬就找到 python-telegram-bot，然後根據他的 “Your first Bot” 文件很快的就寫出可以接受指令與發送訊息的簡易程式碼，而 Telegram Chat 有分成 Pull (Updater) 和 Push (Webhook) 兩種模式來接收到使用者的訊息，這邊我使用 Pull 模式，所以就先不需要準備對外端點曝露出來接收 Telegram 送過來的請求，然後就把自己想要實作的指令一個個用 Function 定義好，例如 help, subscribe, unsubscribe，然後透過 dispatcher 將這些指令加到 ChatBot 中，這樣一來就完成 ChatBot 的外部功能了！

# 詳細程式碼請參考 https://github.com/StarBugsWeekly/newsBot/blob/master/src/functions/telegram/telegram.pydef __init__(self):
    self.tg_bot        = Bot(token=telegram_api_token)
    self.tg_updater    = Updater(token=telegram_api_token, use_context=True)
    self.tg_dispatcher = self.tg_updater.dispatcherdef go_live(self):
    self.command_handler('help', self.help)
    self.command_handler('subscribe', self.subscribe)
    self.command_handler('unsubscribe', self.unsubscribe)    self.tg_updater.start_polling()def command_handler(self, command: str, handle_function: methods)
    self.tg_dispatcher.add_handler(handler)# Define Telegram command help
def help(self, update: Update, context: CallbackContext):
...# Define Telegram command subscribe
def subscribe(self, update: Update, context: CallbackContext):
...# Define Telegram command unsubscribe
def unsubscribe(self, update: Update, context: CallbackContext):
...

GitHub API

StarBugs Weekly 的文章都是透過 GitHub Repository 來做管理，所以我先開始研究我們的 Repository 在發布新文章時會有什麼樣的 Git 行為，發布的文章網址又會長什麼樣子，確定好他們的行為模式後，我又去看了看 GitHub API 文件，發現透過比較不同 Git Commit 之間的檔案差異就可以判斷是否有新的文章會產生應該是個可行方案，所以就決定朝這個方向去進行看看

在一樣秉持著 DRY 的原則之下，找到了 PyGithub 這個 Python Library 來做使用，這邊的程式碼其實主要就是在做文件跟字串處理，我試著去比較兩個 Commit 間的差異是否有新增檔案的行為發生，假如有的話，透過正規表示式去判斷新增的文章是否為想要發布給讀者閱讀的文章，假如是的話，就將其之後會發布在網站上的網址字串拼湊出來，然後 Telegram ChatBot 就可以把這些網址發送給訂閱的使用者

# 詳細程式碼請參閱 https://github.com/StarBugsWeekly/newsBot/blob/master/src/functions/github/github.pydef check_publish_status(self, repo_name: str, redis_key: str, publish_type: str, url_pattern: str, file_regex: re.Pattern, date_regex: re.Pattern):    target_repo   = self.github_client.get_repo(repo_name)
    cached_info   = self.get_cached_info(redis_key)
    latest_commit = self.get_latest_commit(target_repo, publish_type)
    publich_urls  = []    if not cached_info:
        self.upsert_current_commit(redis_key, latest_commit, publich_urls)    else:
        current_commit = cached_info['current_commit']
        publich_urls = self.get_publich_urls(target_repo, current_commit, latest_commit, url_pattern, file_regex, date_regex)
    
        if current_commit != latest_commit and publich_urls:
            self.upsert_current_commit(redis_key, latest_commit, publich_urls)    return publich_urls

Celery

理論上，前兩個功能實作完畢之後，ChatBot 就已經可以正常運作了，但是自己覺得好像還不太完整，假如在訂閱的人數持續增加，而且想要做的事情更多的情況之下，一直使用同步的流程可能會不太好，所以在這裡將 Python 領域內很有名的 Celery 給加了進來，讓某些工作可以非同步來完成

Celery 是用來協助實作 Task Queue 的函式庫，他可以讓工作任務在不同的 Thread 或是機器中來分散合作完成，可以簡單拆分成三個角色來看：

Producer: 產生工作任務的角色
Queue: 工作任務等待被處理時待的地方
Consumer: 處理工作任務的角色

以這次實作的 ChatBot 來說，從 Redis 中查詢訂閱客戶時，可以把想要送給某個 Telegram 使用者訊息的工作丟到 Queue 中，然後就直接繼續查詢下一個訂閱客戶，不需要等到送出一個訂閱客戶的訊息後，才能繼續從 Redis 中查詢下一個訂閱客戶，讓需要比較久時間的任務先送到 Queue 去等待 Worker 去進行非同步執行

# Producer 的完整程式碼: https://github.com/StarBugsWeekly/newsBot/blob/master/src/producer.pyfor recommend_publich_url in recommend_publich_urls:           
    for subscriber in redis_client.scan_iter():               
        consumer.send_message.delay(subscriber.decode(), recommend_publich_url)# Consumer 的完成程式碼: https://github.com/StarBugsWeekly/newsBot/blob/master/src/celery_app/consumer.py@newsBot.task
def send_message(chat_id, text):
    bot.send_message(chat_id, text)

Conclusion

透過這次的 Side Project，學習到最多的便是如何去開發 Telegram ChatBot，也順便複習到 GitHhub API 和 Celery 這兩個工具，而這次並沒有偷懶將發布文章的消息傳到到某個 Telegram Group 中讓有興趣的人加入就好的主要原因在於…我有一個想要做到個人訂閱客製化的願景，所以這次設計 ChatBot 時讓使用者分開各自對 ChatBot 進行訂閱！

而這次 StarBugs Weekly 將每週文章推薦的 Website 跟 GitHub Repository 獨立切開，最主要是希望有更多的新血加入，讓讀者可以閱讀到更多不同領域的科技文章，歡迎任何想要督促自己閱讀科技資訊吸收新知，增加英文閱讀能力，練習表達能力的人一起來貢獻 💪