2025年4月13日 星期日

Python 學習筆記 : 用 Telegram Bot 取代 LINE Notify (六)

本篇旨在將前一篇用 HTTP API 傳送影片的程式碼改成用 Bot 物件來實作. 

本系列之前的文章參考 :


我的 Telegram Bot API 權杖與聊天室識別碼都存放在環境變數檔 .env 裡面, 可用 dotenv 套件載入後再用 os 模組讀取出來放在 token 與 chat_id 變數裡 : 

>>> from dotenv import load_dotenv   
>>> load_dotenv()    
True 
>>> import os    
>>> token=os.environ.get('TELEGRAM_TOKEN')   
>>> chat_id=os.environ.get('TELEGRAM_ID')   

# load_secrets_py
from dotenv import load_dotenv  
import os  

load_dotenv()    
token=os.environ.get('TELEGRAM_TOKEN')   
chat_id=os.environ.get('TELEGRAM_ID') 

參考 :


Telegram 雖然支援常見的 mp4/mov/avi/mkv/webm 等影片格式, 但其中以 H.264 編碼的 mp4 支援度最佳, 傳送結果最穩定, mov 格式則要視編碼與音訊格式部分支援, 至於 avi/mkv/webm 格式則不保證能順利傳送, 所以基本上還是以傳送 mp4 影片為主 (H.264 + AAC 格式). 


9. 利用 Bot 物件傳送影片 :     

新版 (v20+) 的 python-telegram-bot 套件使用非同步架構, 所有呼叫 Bot 物件的方法的函式要用 async 修飾, 要用 await 去呼叫方法. 關於 Bot 物件用法參考本系列第二篇測試 :



(1). 傳送網路影片 :  

使用 Bot 物件傳送單一網路影片的程式碼如下 :

# telegram_send_web_video_bot.py
import asyncio
from telegram import Bot
from telegram.error import TelegramError

async def telegram_web_video_bot(token, chat_id, video_url, caption=None, parse_mode='Markdown'):
    bot=Bot(token=token)
    try:
        await bot.send_video(
            chat_id=chat_id,
            video=video_url,
            caption=caption,
            parse_mode=parse_mode
            )
        print('Message sent successfully.')
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False

if __name__ == '__main__':
    token='YOUR TOKEN'
    chat_id='YOUR CHAT ID'
    video_url='URL OF VIDEO'  # 例如:https://example.com/sample.mp4
    caption='THE CAPTION'
    asyncio.run(telegram_web_video_bot(token, chat_id, video_url, caption))

結構上比起用 HTTP API 要簡潔許多. 注意, def 前須加 async, 呼叫 bot.send_video() 前要加 await, 執行時要把函式呼叫傳給 asyncio.run() 以非同步方式執行. 

測試如下 : 

>>> import asyncio   
>>> from telegram import Bot    
>>> from telegram.error import TelegramError   
>>> async def telegram_web_video_bot(token, chat_id, video_url, caption=None, parse_mode='Markdown'):
    bot=Bot(token=token)
    try:
        await bot.send_video(
            chat_id=chat_id,
            video=video_url,
            caption=caption,
            parse_mode=parse_mode
            )
        print('Message sent successfully.')
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False

用 samplelib.com 上的範例影片測試 :

>>> video_url='https://samplelib.com/lib/preview/mp4/sample-5s.mp4'   
>>> caption='範例影片'   
>>> asyncio.run(telegram_web_video_bot(token, chat_id, video_url, caption))   
Message(caption='範例影片', channel_chat_created=False, chat=Chat(first_name='<我的名字>', id=<聊天室 ID>, type=<ChatType.PRIVATE>, username='<我的 username>'), date=datetime.datetime(2025, 4, 12, 15, 19, 42, tzinfo=datetime.timezone.utc), delete_chat_photo=False, from_user=User(first_name='twstock', id=7938146214, is_bot=True, username='twstock168_bot'), group_chat_created=False, message_id=174, supergroup_chat_created=False, video=Video(api_kwargs={'thumb': {'file_id': 'AAMCBAADGQMAA5pn-c9qwM3evHoXsPcf3ovsgwFAzAACpgcAAu6tHVKvU2k_MzPKIwEAB20AAzYE', 'file_unique_id': 'AQADpgcAAu6tHVJy', 'file_size': 29129, 'width': 320, 'height': 180}}, duration=6, file_id='BAACAgQAAxkDAAOaZ_nPasDN3rx6F7D3H96L7IMBQMwAAqYHAALurR1Sr1NpPzMzyiM2BA', file_name='sample-5s.mp4', file_size=2848208, file_unique_id='AgADpgcAAu6tHVI', height=1080, mime_type='video/mp4', thumbnail=PhotoSize(file_id='AAMCBAADGQMAA5pn-c9qwM3evHoXsPcf3ovsgwFAzAACpgcAAu6tHVKvU2k_MzPKIwEAB20AAzYE', file_size=29129, file_unique_id='AQADpgcAAu6tHVJy', height=180, width=320), width=1920))
Message sent successfully.
True

App 收到所傳送的影片 : 




使用 Bot 物件傳送多個影片時, 須使用 telegram. InputMediaPhoto 類別來打包這些影片, 程式碼如下 : 

# telegram_send_web_videos_bot.py
import asyncio
from telegram import Bot, InputMediaVideo   
from telegram.error import TelegramError

async def telegram_web_videos_bot(token, chat_id, media_list, caption_list, parse_mode='Markdown'):
    bot=Bot(token=token)
    media_group=[
        InputMediaVideo(
            media=url,
            caption=caption,
            parse_mode=parse_mode
            )
        for url, caption in zip(media_list, caption_list)
        ]
    try:
        await bot.send_media_group(chat_id=chat_id, media=media_group)
        print('Videos sent successfully.')
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False

if __name__ == '__main__':
    token='YOUR TOKEN'
    chat_id='YOUR CHAT ID'
    video_urls=[
        'https://yaohuang1966.github.io/media/cat.mp4',
        'https://yaohuang1966.github.io/media/remind.mp4'
        ]
    captions=['小喵三兄弟', '明天記得上班']
    asyncio.run(telegram_web_videos_bot(token, chat_id, video_urls, captions))

此處利用 InputMediaVideo 類別與串列生成式將每一個網路 mp4 檔打包後放進串列中, 然後在呼叫 Bot 物件的 send_media_group() 方法時將此媒體串列傳給 media 參數, 測試結果如下 :

>>> import asyncio   
>>> from telegram import Bot, InputMediaVideo   
>>> from telegram.error import TelegramError    
>>> async def telegram_web_videos_bot(token, chat_id, media_list, caption_list, parse_mode='Markdown'):
    bot=Bot(token=token)
    media_group=[
        InputMediaVideo(
            media=url,
            caption=caption,
            parse_mode=parse_mode
            )
        for url, caption in zip(media_list, caption_list)
        ]
    try:
        await bot.send_media_group(chat_id=chat_id, media=media_group)
        print('Videos sent successfully.')
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False

用目前工作目錄下的兩個 mp4 檔測試 :

>>> video_urls=[
    'https://yaohuang1966.github.io/media/cat.mp4',
    'https://yaohuang1966.github.io/media/remind.mp4'
    ]
>>> captions=['小喵三兄弟', '明天記得上班']   
>>> asyncio.run(telegram_web_videos_bot(token, chat_id, video_urls, captions))  
(Message(caption='小喵三兄弟', channel_chat_created=False, chat=Chat(first_name='<我的名字>', id=<聊天室 ID>, type=<ChatType.PRIVATE>, username='<我的 username>'), date=datetime.datetime(2025, 4, 12, 22, 48, 47, tzinfo=datetime.timezone.utc), delete_chat_photo=False, from_user=User(first_name='twstock', id=7938146214, is_bot=True, username='twstock168_bot'), group_chat_created=False, media_group_id='13955985018210413', message_id=182, supergroup_chat_created=False, video=Video(api_kwargs={'thumb': {'file_id': 'AAMCBAADFQdn-h--aWyJYqCw7AsR1boJCNTdoAACmQcAAhk71FOnbd3lJ4PIGgEAB20AAzYE', 'file_unique_id': 'AQADmQcAAhk71FNy', 'file_size': 9156, 'width': 320, 'height': 149}}, duration=17, file_id='BAACAgQAAxUHZ_ofvmlsiWKgsOwLEdW6CQjU3aAAApkHAAIZO9RTp23d5SeDyBo2BA', file_name='cat.mp4', file_size=3947077, file_unique_id='AgADmQcAAhk71FM', height=596, mime_type='video/mp4', thumbnail=PhotoSize(file_id='AAMCBAADFQdn-h--aWyJYqCw7AsR1boJCNTdoAACmQcAAhk71FOnbd3lJ4PIGgEAB20AAzYE', file_size=9156, file_unique_id='AQADmQcAAhk71FNy', height=149, width=320), width=1280)), Message(caption='明天記得上班', channel_chat_created=False, chat=Chat(first_name='<我的名字>', id=<聊天室 ID>, type=<ChatType.PRIVATE>, username='<我的 username>'), date=datetime.datetime(2025, 4, 12, 22, 48, 47, tzinfo=datetime.timezone.utc), delete_chat_photo=False, from_user=User(first_name='twstock', id=7938146214, is_bot=True, username='twstock168_bot'), group_chat_created=False, media_group_id='13955985018210413', message_id=183, supergroup_chat_created=False, video=Video(api_kwargs={'thumb': {'file_id': 'AAMCBAADGQMAA59n-guWrp2yFseDaNM-6HdKwv0N3QACggcAAkA31FORB6cGxgUTDQEAB20AAzYE', 'file_unique_id': 'AQADggcAAkA31FNy', 'file_size': 18223, 'width': 320, 'height': 320}}, duration=13, file_id='BAACAgQAAxkDAAOfZ_oLlq6dshbHg2jTPuh3SsL9Dd0AAoIHAAJAN9RTkQenBsYFEw02BA', file_name='remind.mp4', file_size=516935, file_unique_id='AgADggcAAkA31FM', height=360, mime_type='video/mp4', thumbnail=PhotoSize(file_id='AAMCBAADGQMAA59n-guWrp2yFseDaNM-6HdKwv0N3QACggcAAkA31FORB6cGxgUTDQEAB20AAzYE', file_size=18223, file_unique_id='AQADggcAAkA31FNy', height=320, width=320), width=360)))
Videos sent successfully.
True

開啟 Telegram App 檢視, 果然收到這兩個影片 : 





(2). 傳送本機影片 :  

我們直接來改寫前一篇測試中用 HTTP API 傳送一個本機影片的 telegram_send_local_video() 函式, 其 Bot 物件版的寫法如下 :

# telegram_send_local_video_bot.py
import asyncio
from telegram import Bot
from telegram.error import TelegramError

async def telegram_local_video_bot(token, chat_id, video_path, caption=None, parse_mode='Markdown'):
    bot=Bot(token=token)
    try:
        with open(video_path, 'rb') as video_file:
            await bot.send_video(
                chat_id=chat_id,
                video=video_file,
                caption=caption,
                parse_mode=parse_mode
                )
        print('Message sent successfully.')
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False
    except OSError as e:
        print(f'File error: {e}')
        return False

if __name__ == '__main__':
    token='YOUR TOKEN'
    chat_id='YOUR CHAT ID'
    video_path='VIDEO FILE PATH'  # 例如 './video/cat.mp4'
    caption='THE CAPTION'
    asyncio.run(telegram_local_video_bot(token, chat_id, video_path, caption))

此處使用 open() 開啟影片檔後直接將其傳給 Bot 物件的 send_video() 方法的 video 參數, 測試結果如下 : 

>>> import asyncio   
>>> from telegram import Bot, InputMediaVideo   
>>> from telegram.error import TelegramError    
>>> async def telegram_local_video_bot(token, chat_id, video_path, caption=None, parse_mode='Markdown'):
    bot=Bot(token=token)
    try:
        with open(video_path, 'rb') as video_file:
            await bot.send_video(
                chat_id=chat_id,
                video=video_file,
                caption=caption,
                parse_mode=parse_mode
                )
        print('Message sent successfully.')
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False
    except OSError as e:
        print(f'File error: {e}')
        return False

使用目前工作目錄下的 cat.mp4 測試 : 

>>> video_path='cat.mp4'   
>>> caption='小喵三兄弟'   
>>> asyncio.run(telegram_local_video_bot(token, chat_id, video_path, caption))     
Message(caption='小喵三兄弟', channel_chat_created=False, chat=Chat(first_name='<我的名字>', id=<聊天室 ID>, type=<ChatType.PRIVATE>, username='<我的 username>'), date=datetime.datetime(2025, 4, 12, 23, 23, 12, tzinfo=datetime.timezone.utc), delete_chat_photo=False, from_user=User(first_name='twstock', id=7938146214, is_bot=True, username='twstock168_bot'), group_chat_created=False, message_id=185, supergroup_chat_created=False, video=Video(api_kwargs={'thumb': {'file_id': 'AAMCBQADGQMAA7ln-vXgEh0dMrmAyQKmK3XZ1KBo_wAChxUAAhT62FdWAedgDpAVbwEAB20AAzYE', 'file_unique_id': 'AQADhxUAAhT62Fdy', 'file_size': 9156, 'width': 320, 'height': 149}}, duration=17, file_id='BAACAgUAAxkDAAO5Z_r14BIdHTK5gMkCpit12dSgaP8AAocVAAIU-thXVgHnYA6QFW82BA', file_name='cat.mp4', file_size=3947077, file_unique_id='AgADhxUAAhT62Fc', height=596, mime_type='video/mp4', thumbnail=PhotoSize(file_id='AAMCBQADGQMAA7ln-vXgEh0dMrmAyQKmK3XZ1KBo_wAChxUAAhT62FdWAedgDpAVbwEAB20AAzYE', file_size=9156, file_unique_id='AQADhxUAAhT62Fdy', height=149, width=320), width=1280))
Message sent successfully.
True

查看 Telegram App 果然有收到此影片 : 




如果要傳送本機多個影片, 與上面傳送多個網路影片一樣要用到 InputMediaVideo 類別. 其實不論是網路影片還是本機影片, 如果要傳送多個影片都要用 InputMediaVideo 類別來將各影片打包, 差別只是本機影片需要開啟檔案后再打包, 網路影片則是直接將影片網址打包而已. 

下面是將前一篇中的 telegram_local_videos() 改寫為 Bot 物件版的寫法 :

# telegram_send_local_videos_bot.py
import asyncio
from telegram import Bot, InputMediaVideo
from telegram.error import TelegramError

async def telegram_local_videos_bot(token, chat_id, video_paths, captions):
    bot=Bot(token=token)
    media=[]  # 儲存用 InputMediaVideo 打包後的影片
    files=[]  # 儲存開啟的所有影片檔案
    try:
        for i, path in enumerate(video_paths):
            f=open(path, 'rb')
            files.append(f)
            caption=captions[i] if i < len(captions) else None
            media.append(InputMediaVideo(media=f, caption=caption))
        await bot.send_media_group(chat_id=chat_id, media=media)
        print("Message sent successfully.")
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False
    except OSError as e:
        print(f'File error: {e}')
        return False
    finally:
        for f in files:
            f.close()

if __name__ == '__main__':
    token='YOUR TOKEN'
    chat_id='YOUR CHAT ID'
    video_paths=['cat.mp4', 'remind.mp4']
    captions=['小喵三兄弟', '記得明天要上班']
    asyncio.run(telegram_local_videos_bot(token, chat_id, video_paths, captions))

此例與上面傳送多個網路影片不同之處是需要一個 files 串列來儲存開啟過的影片檔案參考, 以便能在傳送完成後記得要關閉那些檔案. 測試結果如下 :

>>> import asyncio   
>>> from telegram import Bot, InputMediaVideo   
>>> from telegram.error import TelegramError    
>>> async def telegram_local_videos_bot(token, chat_id, video_paths, captions):
    bot=Bot(token=token)
    media=[]  # 儲存用 InputMediaVideo 打包後的影片
    files=[]  # 儲存開啟的所有影片檔案
    try:
        for i, path in enumerate(video_paths):
            f=open(path, 'rb')
            files.append(f)
            caption=captions[i] if i < len(captions) else None
            media.append(InputMediaVideo(media=f, caption=caption))
        await bot.send_media_group(chat_id=chat_id, media=media)
        print("Message sent successfully.")
        return True
    except TelegramError as e:
        print(f'Telegram API Error: {e}')
        return False
    except OSError as e:
        print(f'File error: {e}')
        return False
    finally:
        for f in files:
            f.close()

使用目前工作目錄下的 cat.mp4 與 remind.mp4 測試 : 

>>> video_paths=['cat.mp4', 'remind.mp4']   
>>> captions=['小喵三兄弟', '記得明天要上班']   
>>> asyncio.run(telegram_local_videos_bot(token, chat_id, video_paths, captions))   
(Message(caption='小喵三兄弟', channel_chat_created=False, chat=Chat(first_name='<我的名字>', id=<聊天室 ID>, type=<ChatType.PRIVATE>, username='<我的 username>'), date=datetime.datetime(2025, 4, 12, 23, 44, tzinfo=datetime.timezone.utc), delete_chat_photo=False, from_user=User(first_name='twstock', id=7938146214, is_bot=True, username='twstock168_bot'), group_chat_created=False, media_group_id='13956011520727829', message_id=186, supergroup_chat_created=False, video=Video(api_kwargs={'thumb': {'file_id': 'AAMCBQADFQdn-vq_N0rzmlKlGv1uWTvrueGBOQACihUAAhT62Fc0NX6pzVwmHwEAB20AAzYE', 'file_unique_id': 'AQADihUAAhT62Fdy', 'file_size': 9156, 'width': 320, 'height': 149}}, duration=17, file_id='BAACAgUAAxUHZ_r6vzdK85pSpRr9blk767nhgTkAAooVAAIU-thXNDV-qc1cJh82BA', file_name='cat.mp4', file_size=3947077, file_unique_id='AgADihUAAhT62Fc', height=596, mime_type='video/mp4', thumbnail=PhotoSize(file_id='AAMCBQADFQdn-vq_N0rzmlKlGv1uWTvrueGBOQACihUAAhT62Fc0NX6pzVwmHwEAB20AAzYE', file_size=9156, file_unique_id='AQADihUAAhT62Fdy', height=149, width=320), width=1280)), Message(caption='記得明天要上班', channel_chat_created=False, chat=Chat(first_name='<我的名字>', id=<聊天室 ID>, type=<ChatType.PRIVATE>, username='<我的 username>'), date=datetime.datetime(2025, 4, 12, 23, 44, tzinfo=datetime.timezone.utc), delete_chat_photo=False, from_user=User(first_name='twstock', id=7938146214, is_bot=True, username='twstock168_bot'), group_chat_created=False, media_group_id='13956011520727829', message_id=187, supergroup_chat_created=False, video=Video(api_kwargs={'thumb': {'file_id': 'AAMCBQADFQdn-vq_aNXB5SHwb2CU0UwyBjeTBAACiRUAAhT62FeAbDy1aJktFAEAB20AAzYE', 'file_unique_id': 'AQADiRUAAhT62Fdy', 'file_size': 18223, 'width': 320, 'height': 320}}, duration=13, file_id='BAACAgUAAxUHZ_r6v2jVweUh8G9glNFMMgY3kwQAAokVAAIU-thXgGw8tWiZLRQ2BA', file_name='remind.mp4', file_size=516935, file_unique_id='AgADiRUAAhT62Fc', height=360, mime_type='video/mp4', thumbnail=PhotoSize(file_id='AAMCBQADFQdn-vq_aNXB5SHwb2CU0UwyBjeTBAACiRUAAhT62FeAbDy1aJktFAEAB20AAzYE', file_size=18223, file_unique_id='AQADiRUAAhT62Fdy', height=320, width=320), width=360)))
Message sent successfully.
True

查看 Telegram App 果然收到了這兩個影片 : 



沒有留言 :