更新於 2024/12/15閱讀時間約 22 分鐘

技術筆記-用 python 叫 discord 發訊息的幾種方式

「通知」這種功能現在已經非常稀鬆平常了,三不五時手機跳出訊息或聲音,不勝其擾!但不要因噎廢食,對於「真正重要的事件」,即時通知還是非常必需的。像是刷卡通知,應該沒有人會關掉吧,這種服務早期還需要付費的呢!所以對於一個交易系統而言,牽涉到金錢的進出,通知功能是一定要的啦。有 line notify 相伴的年代,輕鬆愉快:

import resuests
def notify(msg):
url = "https://notify-api.line.me/api/notify"
token = "xxx"
headers = {"Authorization": f"Bearer {token}"}
data = {"message": msg }
res = requests.post(url, headers = headers, data = data)
if res.status_code in (200, 204):
print(f"Request fulfilled with response: {res.text}")
else:
print(f"Request failed with response: {res.status_code}-{res.text}")

以上幾行程式碼,簡單發出 http request 就完成了。但輕鬆的時代即將結束, line notify 於 2015 Q1 就要結束服務了!心想別怕,好好找找新玩具,也許送走一個輕鬆的時代,將迎來一個「更輕鬆」的全新境界,此時 discord 吸引了我的目光。


印象中的 discord,已經在遊戲玩家,幣圈,厲害的開發者社群等等之中,已經火紅很久了。研究了一天之後,發現此平台太強大,發訊息達成通知功能實在太容易了,簡直是殺雞用牛刀!但一開始不知道,走了迂迴的道路,在 multi-thread,async await 森林中迷失,但也學習到相關的寶貴知識,以下娓娓道來。

方法一:webhook

這是最簡單的方式,在建立一個自己的伺服器之後,按個鍵「產生 webhook」,就完成了。真的很絕,真的沒有額外動作,用上述呼叫 line notify 一樣的方式,甚至更簡單,因為連認證的 token 都不需要,直接完成!

discord 非常慷慨,每個人都可以建立自己的「伺服器」,建立後會有預設的文字頻道,按齒輪圖示「編輯頻道」,依上述截圖點一點,即刻生成 webhook 網址,複製之貼入以下程式:

def notify_discord_webhook(msg):
url = 'https://discord.com/api/webhooks/131743762990xxxyyy'
headers = {"Content-Type": "application/json"}
data = {"content": msg, "username": "newmanBot"}
res = requests.post(url, headers = headers, json = data)
if res.status_code in (200, 204):
print(f"Request fulfilled with response: {res.text}")
else:
print(f"Request failed with response: {res.status_code}-{res.text}")

這段程式可以在系統任何地方呼叫,可即時送出訊息到此頻道,保持手機 discord app 允許通知,系統的重要事件不會漏接。若只需要通知功能,這樣已經足夠,下文也可以略過了。但強大的 discord 只玩這樣未免可惜,還是建個 bot (聊天機器人) 來玩玩吧,這就需要進一步了解 api。

方法二:原生 api

上述從 webhook 發出的訊息,並不具備身份識別,所以發完訊息後完全沒有後續互動的可能性,也無法發出「私訊」,但這些才是 discord 的主要功能。要體驗這些功能,必須先到 developer portal 建立一個 application,這是一切開發的起點。進到所建立的 application,首先取得第一項重要資訊:token,注意只會出現一次,複製起來以備程式使用:

有了 token,相當於這個 bot 的身分證,接下來是授權給這個 bot 做事的權限,如允許他讀取群組人員上線狀態,訊息內容等等,先把下面三個勾打開:

然後還要針對某個頻道授權,discord 的設計相當貼心,他可以幫我們產生一串 url,到另一個瀏覽頁面打開,打開時需要登入認證,然後就可選擇授權自己旗下有權限的頻道,給 bot 運作施展的空間:

從以上畫面選擇 bot permission,自己用的當然是給足權限 Administrator,捲到下面複製 url,貼到新的瀏覽器頁面:

到此終於完成了,而 token 就是鑰匙,將這所有的授權內容鎖住,程式只要持有 token 就可以解鎖這些資源了。因此一樣簡單的程式,差別只是帶著 token:

def notify_discord_channel(msg, token, channel_id):
url_base = 'https://discord.com/api/v10'
url = f"{url_base}/channels/{channel_id}/messages"
headers = {
"Authorization": f"Bot {token}",
"Content-Type": "application/json"
}
data = {
"content": msg,
}
res = requests.post(url, headers = headers, json = data)
if res.status_code in (200, 204):
print(f"Request fulfilled with response: {res.text}")
else:
print(f"Request failed with response: {res.status_code}-{res.text}")

以上 channel id,只要在 discord 的「使用者設定」的「進階」中,打開「開發者模式」,之後在頻道中按右鍵就會出現「複製頻道 ID」,可以取得。

實測之後,有沒有發現訊息的圖示不一樣了,是一個「有身份的人」發出的訊息:

以上已經完成 bot 的初體驗。好像沒有比 webhook 強到哪裡去?的確是的。都走到這一步了,就再多走一點吧,bot 能不能發「私訊」?可以的。不由得再度讚賞 discord 的寬容大度:

# 發訊息給「個人」,先用另一個 api 取得對該 user「動態產生的 channel id」
# 再呼叫傳送到 channel 的 function 即可​
def notify_discord_user(msg, token, user_id):
url_base = 'https://discord.com/api/v10'
url = f"{url_base}/users/@me/channels"
headers = {
"Authorization": f"Bot {token}",
"Content-Type": "application/json"
}
data = {
"content": msg,
"recipient_id": user_id
}
res = requests.post(url, headers = headers, json = data)
if res.status_code == 200:
dm_channel = res.json()
channel_id = dm_channel["id"]
CrawlService.notify_discord_channel(msg, token, channel_id)
else:
print(f"Request failed with response: {res.status_code}-{res.text}")

像阿甘跑步一樣,都跑這麼遠了,再多跑一點吧!一個 bot 應該要可以回訊息吧?當然要的,也必須可以由訊息驅動任何程式動作。這些功能若用原生 api 覺得有點吃力,但有網路大神,已經把 api 包起來了,成為 python 套件了,太強大,太感謝。

方法三:python.py

此套件用標準 pip 安裝即可: pip install python.py

坑#1

先預告一個坑,若你遇到「No module named 'audioop'.」,代表你的 python 版本太新了,這在 python@3.13 是一個 known issue,暫時未解,只好降版到 python@3.12。

坑#2

然後再預告一個坑,初始化連線就會遇到的:

Cannot connect to host discord.com:443 ssl:True \
[SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] \
certificate verify failed: unable to get local issuer certificate \
(_ssl.c:1000)')]

這個錯誤非常隱晦,竟然有點似曾相識,咦!就是在測試富果 api 建立 socket 連線時遇到的,本來以為是憑證檔的問題,後來在社群大哥的提點下,終於搞清楚,是 ssl 底層基本設定出問題了,解法如下:

# 先執行以下程式,得到 openssl_capath
import ssl
result = ssl.get_default_verify_paths()
print(result)
-------------
DefaultVerifyPaths(cafile=None, capath=None, openssl_cafile_env='SSL_CERT_FILE', \
openssl_cafile='/Library/Frameworks/Python.framework/Versions/3.12/etc/openssl/cert.pem', \
openssl_capath_env='SSL_CERT_DIR', \
openssl_capath='/Library/Frameworks/Python.framework/Versions/3.12/etc/openssl/certs')

在 mac 中用 finder 到上述 openssl_capath 看,發現空空的,應該要有東西,我們需要複製一個憑證檔進去,檔案來源也非常隱晦,請到自己的 python 環境中,尋找 certifi 套件位置,其中有一個憑證檔:

把憑證檔 copy 進去上述的 openssl path 中,記得還要改檔名,如我的 case 甚至是 certs 沒有副檔名!真是詭異。無論如何,這樣搞一搞就可以了,這樣只到 import discord 成功而已,才剛開始要寫程式呢!哈哈,還有路要走。

正式開始 coding

先寫一個最小版本的 bot,能讀訊息和回訊息,一個獨立的 py 檔如下:

import discord
from datetime import datetime

intents = discord.Intents.default()
intents.message_content = True
intents.messages = True
intents.guilds = True

bot = discord.Client(intents=intents)

DISCORD_TOKEN = 'MTMxNjg5NTE4MjE0NjM3xxx'

@bot.event
async def on_ready():
print(f'Logged in as {bot.user.name}')
guild = bot.get_guild(GUILD_ID)
if not guild:
print('Guild not found')
return
channel = guild.get_channel(CHANNEL_ID)
if not channel:
print('Channel not found')
return
await channel.send('Hello from server')

@bot.event
async def on_message(message):
print(f'{datetime.now()} {message.author}: {message.content}')
text = message.content
if text.startswith('$'):
channel = message.channel
await channel.send(f'Thanks for sending {text[1:]}')

bot.run(DISCORD_TOKEN)

此 bot 能讀訊息,若發現前贅字是 $,則回應 $ 之後的字元。

這段程式完成了非常多任務,包括與伺服器建立 socket 連線,持續監聽頻道,on_message 事件處理器,在接到訊息之後可以串連驅動任何後端的程式模組,進行任何高深的計算,且回應任何訊息!其實已經非常強大了,大可以停在這裡,好好去開發應用了。又是因為手癢的原因,想要把他整合到 ui,這又讓我遇到了 multi-thread 的問題!技術細節再論下去實在太繁瑣,就紀錄下最終的結果吧。


以下的 class 可以由 ui 的主程式呼叫,建立 bot 的監聽訊息服務,也可隨時呼叫傳送訊息。只是經過前面的探討,傳送訊息只要額外呼叫 http request 就可完成了,
大可不必像這裏硬要呼叫套件的 send 功能,導致需要再度加開 thread 去處理,實在是疊床架屋的做法。但既然做了,就記下來欣賞一下,當作對 thread async await 的學習。

class DiscordService():
def __init__(self):
self.TOKEN = 'MTMxNjg5NTE4MjE0NjM3MTxxx'
self.GUILD_ID = 00000
self.CHANNEL_ID = 00000
self.ADMIN_USER_ID = 00000 # newman.church
self.loop = asyncio.new_event_loop()

# creating bot
self.intents = discord.Intents.default()
self.intents.message_content = True
self.intents.messages = True
self.intents.guilds = True
self.client = discord.Client(intents=self.intents)

# event handling
self.client.event(self.on_ready)
self.client.event(self.on_message)

# bot will run in seperate thread
self.bot_thread = threading.Thread(target=self.do_trigger_run_bot
, daemon=True)
self.bot_thread.start()

def send_message_channel(self, msg):
# 簡單的 http request
CrawlService.notify_discord_channel(msg, self.TOKEN, self.CHANNEL_ID)

def send_message_user(self, msg):
# 簡單的 http request
CrawlService.notify_discord_user(msg, self.TOKEN, self.ADMIN_USER_ID)

async def on_ready(self):
print(f'{datetime.now()} Logged on as {self.client.user}!')

async def on_message(self, message):
print(f'{datetime.now()} {message.author}: {message.content}')

async def trigger_send_message(self, message):
guild = self.client.get_guild(self.GUILD_ID)
if not guild:
print(f"Guild with ID {self.GUILD_ID} not found.")
return
channel = guild.get_channel(self.CHANNEL_ID)
if not channel:
print(f"Channel with ID {self.CHANNEL_ID} not found.")
return
# 這是套件的傳訊息功能
await channel.send(message)

​# 呼叫套件的傳訊息功能,必須再開立新的 thread 處理
# 因為是 async function,呼叫方式必須借用 asyncio 幫忙用安全方式呼叫
def send_message_foolish(self, message):
asyncio.run_coroutine_threadsafe(self.trigger_send_message(message), self.loop)

# 這已經轉太多彎了,暈了
def do_trigger_run_bot(self):
asyncio.set_event_loop(self.loop)
self.loop.create_task(self.trigger_run_bot())
self.loop.run_forever()

async def trigger_run_bot(self):
await self.client.start(self.TOKEN) # start is the shorthand coroutine for login() + connect().

# def run_bot(self):
# self.client.run(self.TOKEN) # run is a blocking call

以上終於完整紀錄一天以來發生的事了。達成目的只需第一段,只是既然走過就留下痕跡吧,學到經驗很值得。

Newman 2024/12/16

導覽頁:紐曼的技術筆記-索引








分享至
成為作者繼續創作的動力吧!
© 2024 vocus All rights reserved.