LINE Bot 學習筆記 : LINE 訊息解析

撰寫 LINE Bot 聊天機器人須對 LINE 的訊息結構有深入了解, 本篇旨在利用我在 render.com 上架設的 serverless 函式執行平台 (類似 GCF 或 AWS Lambda) 上撰寫的測試函式來了解 LINE 訊息結構, 關於利用 render.com 免費帳戶執行 Python web app 的方法參考 :

# render.com 學習筆記索引

關於 serverless 函式執行平台參考 :

# Mapleboard MP510-50 測試 (三十五) : HTTP API 函式執行平台 (1)

# Mapleboard MP510-50 測試 (三十六) : HTTP API 函式執行平台 (2)

# Mapleboard MP510-50 測試 (三十七) : HTTP API 函式執行平台 (3)

# Mapleboard MP510-50 測試 (三十八) : HTTP API 函式執行平台 (4)

# Mapleboard MP510-50 測試 (三十九) : HTTP API 函式執行平台 (5)

# Mapleboard MP510-50 測試 (四十一) : HTTP API 函式執行平台 (6)

# Mapleboard MP510-50 測試 (四十三) : HTTP API 函式執行平台 (7)

render.com 是透過寄存於 GitHub 的 repo 來佈署 web app 的, 最新的 serverless repo 網址如下 :

# https://github.com/tony1966/serverless/tree/main  

在 Serverless 平台上新增的函式模組必須定義一個 main() 函式, 並傳入 request 物件與關鍵字參數 kwargs, 程式結構如下 : 

# func_module.py

def main(request, **kwargs):

    # 取得主程式 serverless 傳遞的環境變數 (存於 ~/.env)

    # config=kwargs.get('config', {})  # 預設傳回空 dict

    # telegram_token=config.get('TELEGRAM_TOKEN')

    # telegram_id=config.get('TELEGRAM_ID')

    result='do something'

    return result

注意, **kwargs 參數是必須的, 主要用來接收從主程式傳來的 API key/token 參數 (例如 Telegram 權杖與聊天室 ID, 或 OpenAI API 金鑰等), 即使此函式用不到也必須接收. 

主程式 serverless 已經匯入如下套件模組

from flask import Flask, request, jsonify, session, redirect

import importlib.util

import os

import logging

from dotenv import dotenv_values

import sqlite3

函式模組會被動態地載入主程式, 因此若有用到上面這些模組就直接使用, 毋須再匯入. 本篇測試旨在利用 LINE App 輸入聊天訊息, LINE 訊息伺服器收到後會傳送給放在 render.com 的 serverless 平台上的 Webhook 程式處理, 此程式會把 LINE 請求訊息傳回 LINE 訊息伺服器回覆提問者, 此 Webhook 程式碼如下 :

# line_message_1.py

import json

from linebot import LineBotApi, WebhookHandler

from linebot.models import TextSendMessage

def main(request, **kwargs):

    # 取得主程式傳入的 LINE secret & token

    config=kwargs.get('config', {})

    secret=config.get('LINE_CHANNEL_SECRET')

    token=config.get('LINE_CHANNEL_ACCESS_TOKEN')

    # 檢查必要參數

    if not secret or not token:

        return {'error': 'Missing LINE or Gemini credentials'}    

    try:

        # 初始化 LINE Bot API 與 Webhook Handler 物件

        line_bot_api=LineBotApi(token)

        handler=WebhookHandler(secret)

        # 取得請求內容

        body=request.get_data(as_text=True)

        # 取得簽章

        signature=request.headers.get('X-Line-Signature', '')

        # 驗證簽章

        try:

            handler.parser.parse(body, signature)

        except InvalidSignatureError:

            return {'error': 'Invalid signature'}

        # 將請求內容轉成 JSON 格式

        json_data=json.loads(body)

        events=json_data.get('events', [])

        # 若為 verify webhook 測試回傳 ok 即可 (沒有事件) 

        if not events:

            return {'status': 'ok'}        

        reply_token=events[0]['replyToken']

        # 設定回傳訊息為請求內容

        text_message=TextSendMessage(text=json.dumps(json_data, ensure_ascii=False))

        # 回傳訊息

        line_bot_api.reply_message(reply_token,text_message)

    except Exception as e:

        return {'error': str(e)}

    return {'status': 'ok'}

此程式需匯入 LineBotApi, WebhookHandler, 與 TextSendMessage 類別, 在 main() 函式中先用關鍵字參數字典取得主程式 serverless.py 傳來的 LINE 訊息通道的 Secret 與 Access token, 以便初始化 LINE Bot API 與 Webhook Handler 物件. 然後用 Flask.request 物件取得 LINE 訊息伺服器傳來的請求內容並利用它驗證數位簽章 (確認此請求來自 LINE), 接著將請求內容轉成字典, 利用有無包含 events 鍵 (值為串列) 來判別是否為 Webhook Verify (無 events 鍵) 還是聊天訊息 (有 events 鍵), 若為 Verify 動作就直接回 200 OK 表示 verify Success; 否則就從 events 中取出 replyToken, 呼叫 LINE Bot API 物件的 reply_message() 方法將請求內容字串傳回. 

先在 render.com 的 serverless 平台新增函式 line_message_1.py :

然後將滑鼠移到函式列表 line_message_1 的執行欄按右鍵, 點選 "複製連結網址" :

接下來要登入 LINE 開發者網站 : 

# https://developers.line.biz/

切到 Messaging API 頁籤, 按 Webhook URL 項目的 Edit 鈕, 將上面 line_message_1.py 函式的網址貼上儲存 : 

按 Verify 驗證網址顯示 Success 即完成 Webhook 設定, 可以用 LINE App 來測試是否能收到 LINE 訊息伺服器傳送給 Webhook 程式的請求內容. 

1. 傳送文字訊息 : 

 

在 LINE 聊天室中切換到 LINE Bot 的官方帳號 (此處為小狐狸事務所機器人), 傳送文字訊息例如 '你好', LINE 訊息伺服器會向上面的 Webhook 程式 line_message_1.py 發出 HTTP 請求, Webhook 程式會將此請求訊息回傳, LINE 訊息伺服器收到後就傳送到手機 LINE App 的 LINE Bot 官方帳號聊天室, 結果如下 : 

收到的 JSON 回覆內容 :

{"destination": "Ufaec4230b58e2d1449895879a-------", "events": [{"type": "message", "message": {"type": "text", "id": "585818269928128938", "quoteToken": "iNYTI1QbQh8SEfdtrEmF3w1NgfWEeDKKZuSIy_qfwvkg3OIZ_MqaDE5vWri4zCpFd0RileKFfof8YzXwkjGBWIDNh6ZFxJ2AbsVY5FbKewiwkj6XZ5qwSHShKv3WfT5bvkwSUI1ZwgmQCQDb3LS-Hw", "text": "你好"}, "webhookEventId": "01K8ZW1YQJSHCXJ9QRF6ZAXJSJ", "deliveryContext": {"isRedelivery": false}, "timestamp": 1762006202834, "source": {"type": "user", "userId": "使用者 ID"}, "replyToken": "a61c2f85aaa64bd99ba534992d10b99a", "mode": "active"}]}

可見 LINE 訊息伺服器發給 Webhook 程式的請求訊息是一個只有 destination 與 events 屬性的 JSON 格式資訊, destination 表示我 LINE 官方帳號 (LINE Bo) 的 User ID, 用來標示這是給哪個 Bot (哪個 LINE channel) 的事件, 而 events 則為訊息事件之內容, 其值為單元素串列, 內容是一個物件, 其中 type, message, timeStamp, source, replyToken, mode 等均為 LINE 訊息的通用屬性, 不管訊息是文字, 圖片, 音訊, 視訊等都有這些屬性, 裡面的 replyToken 最重要, 因為 Webhook 程式在呼叫 reply_message() 時需要用它來指定回覆對象. 

可以將此 JSON 字串貼到 JSON Viewer 網站來觀察其資料結構 : 

# https://jsonviewer.stack.hu/

2. 傳送 Sticker (表情貼圖) 訊息 : 

按 LINE 聊天輸入框右邊的笑臉, 點選任一表情貼圖傳送給 LINE Bot :

這時會收到 Webhook 程式回傳的請求訊息 : 

回覆內容如下 :

{"destination": "Ufaec4230b58e2d1449895879a-------", "events": [{"type": "message", "message": {"type": "sticker", "id": "585828359058489380", "quoteToken": "v4wMDoLMbMtNnOsw7yeZjnN6-xx8d0MQ5h7HNMj_N7c2GfTdLL2Q_YfUaYNNa5BoDhnwdGB2YFe_BzQOaN65cdXz1b3K4gZ4Ge455XakD_GWKlEPjV183ZuDgMdZUJlrl8Q7g5eydK5OBJqjzJRlRg", "stickerId": "51626501", "packageId": "11538", "stickerResourceType": "ANIMATION", "keywords": ["alright", "super", "roger!", "goodidea", "good", "great", "okie", "OK", "line", "moon", "thumbsup", "got it", "of course", "very good", "Affirmative"]}, "webhookEventId": "01K901SFJVF4J9MFH734BMGQ74", "deliveryContext": {"isRedelivery": false}, "timestamp": 1762012216419, "source": {"type": "user", "userId": "使用者ID"}, "replyToken": "e1f6811d581b488690dca89e0ef0413b", "mode": "active"}]}

可見傳送表情貼時, message 屬性值為 sticker, 後面有 stickerId 與 packageId 等與表情貼圖相關之屬性, 其餘為所有訊息共通之屬性. 

但如果要用這方法在 LINE App 上顯示傳送圖片時的請求訊息將不會獲得回應, 因為圖片的訊息量太大, 會超過 LINE 的限制而無法處理, 必須在本機用 ngrok 模擬才行, 參考下面這本書做法 :

# 一本精通-LINE BOT+Python+Google Dialogflow 完整掌握LINE BOT的開發技巧 打造全方位AI機器人

此書作者 OXXO 將原始碼放在 GitHub, 關於訊息解析的程式碼網址如下 :

# https://github.com/oxxostudio/book-code/blob/master/linebot/ch5/code01.py

Python 學習筆記 : 市圖借書與預約爬蟲程式改版 v11

在前一篇測試中已修改爬蟲程式順利抓到典藏地資訊, 改成一次只抓一個借書證帳戶也讓 Selenium 爬蟲的處理程序較為單純, 讓網頁擷取的成功率上升. 本篇旨在紀錄如何將抓取到的資訊儲存在 Mapleboard 或 render.com 的 serverless 平台的 ksml_books 資料表裡, 以便後續讓 LINE Bot 程式隨時能讀取到最新借書資訊. 作為前置作業, 這個周末我先將 serverless 平台生版為 v5, 加入資料表線上管理功能, 參考 :

# Mapleboard MP510-50 測試 (四十三) : HTTP API 函式執行平台 (7)

本系列之前的文章參考 : 

# Python 學習筆記 : 網頁爬蟲實戰 (九) 市立圖書館個人書房借書資訊 (上)

# Python 學習筆記 : 網頁爬蟲實戰 (九) 市立圖書館個人書房借書資訊 (中)

# Python 學習筆記 : 網頁爬蟲實戰 (九) 市立圖書館個人書房借書資訊 (下)

# Python 學習筆記 : 用 Telegram 傳送市圖借書與預約訊息

# Python 學習筆記 : 市圖借書與預約爬蟲程式改版 v10 (一)

1. 建立 ksml_books 資料表 : 

儲存市圖爬蟲結果的資料表取名為 ksml_books, 具有四個欄位 : 

• account : 借書證帳號 (主鍵)
• borrow_books : 借書資訊摘要
• reserve_books : 預約資訊摘要
• updated_at : 最近更新時間

在 serverless 平台的輸入下列 Schema 建立此資料表 :

account PRIMARY KEY, borrow_books TEXT, reserve_books TEXT, updated_at TEXT

2. 撰寫資料表更新程式 update_ksml_books.py : 

此程式接收來自爬蟲程式的的 POST 請求, 將爬取結果 (借書與預約資訊) 存入 ksml_books 資料表, 程式碼如下 :

# update_ksml_books.py

import sqlite3

from datetime import datetime

def main(request, **kwargs):

    """

    POST 請求範例：

    {

        "account": "tony",

        "borrow_books": "書名1 (到期日 2025-10-22); 書名2 ...",

        "reserve_books": "書名A (已到館); 書名B (順位 2) ..."

    }

    """

    DB_PATH='./serverless.db'

    try:  # 從 POST 請求 body 中解析 JSON 格式資料並轉成 Python 字典

        data=request.get_json(force=True)

    except Exception as e:

        return {"status": "error", "message": f"解析 JSON 失敗: {str(e)}"}

    # 從字典中取得參數值

    account=data.get('account')

    borrow_books=data.get('borrow_books', '')

    reserve_books=data.get('reserve_books', '')

    # 檢查主鍵 account 

    if not account:

        return {"status": "error", "message": "缺少帳號資訊"}

    try:  # 更新 ksml_books 資料表

        conn=sqlite3.connect(DB_PATH)

        cur=conn.cursor()

        # 建立 ksml_books 資料表 (若不存在)

        cur.execute("""

            CREATE TABLE IF NOT EXISTS ksml_books (

                account TEXT PRIMARY KEY,

                borrow_books TEXT,

                reserve_books TEXT,

                updated_at TEXT

            )

        """)

        # 取得現在時間

        now_str=datetime.now().strftime('%Y-%m-%d %H:%M:%S')

        # 使用 INSERT OR REPLACE 寫入紀錄，如果帳號已存在就更新

        cur.execute("""

            INSERT OR REPLACE INTO ksml_books (account, borrow_books, reserve_books, updated_at)

            VALUES (?, ?, ?, ?)

        """, (account, borrow_books, reserve_books, now_str))

        conn.commit()

        conn.close()

        return {"status": "success", "message": f"{account} 的資料已更新"}

    except Exception as e:

        return {"status": "error", "message": str(e)}

此程式首先用 request.get_json() 從 POST 本體中將 JSON 字串解析為字典, 然後字典物件的呼叫 get() 方法取得 borrow_books (借閱資訊) 與 reserve_books (預約資訊) 後以 account 為主鍵寫入 ksml_books 表裡面. 在 render.com 的 serverless 平台新增此程式 :

在執行連結上按滑鼠右鍵即可複製此 API 端點網址 ~/function/update_ksml_books.

接下來在本機撰寫一個測試程式來呼叫此 API, 看看能否成功地將資料寫入資料表裡, 只要用 requests.post() 攜帶一個 JSON payload 呼叫 ~/function/update_ksml_books 即可, 程式碼如下 :

# ksml_test.py

import requests

url="https://serverless-fdof.onrender.com/function/update_ksml_books"

payload={

    "account": "tony",

    "borrow_books": "京都時光 (到期日 2025-10-22)",   

    "reserve_books": "LINE Bot 全攻略 (順位 1)"

    }

res=requests.post(url, json=payload)

print(res.json())

執行此程式成功地將資料寫入資料表 :

>>> %Run ksml_test.py   

{'message': 'tony 的資料已更新', 'status': 'success'}

這時檢視資料表列表, 會發現 ksml_books 資料表已有一筆紀錄 :

按檢視即可看到全部記錄 :

3. 修改爬蟲程式 : 

將上面的測試程式寫進前一篇的市圖爬蟲程式 (v10) 裡改為 v11 版, 把爬取到的借書與預約資訊寫入 ksml_books 資料表 :

# ksml_personal_11.py

from selenium import webdriver

from selenium.webdriver.common.by import By

from selenium.webdriver.firefox.options import Options

import re

from datetime import datetime

import time

import requests

import asyncio

from telegram import Bot

import sys

async def telegram_send_text(text):

    bot=Bot(token=token)

    try:

        await bot.send_message(

            chat_id=chat_id,

            text=text

            )

        return True

    except Exception as e:

        print(f'Error sending text: {e}')

        return False

def get_books(account, password):

    try:

        # 登入我的書房

        options=Options()

        options.add_argument("--headless")

        driverpath='geckodriver.exe'

        browser=webdriver.Firefox(options=options)

        browser.implicitly_wait(60)

        browser.get('https://webpacx.ksml.edu.tw/personal/')

        loginid=browser.find_element(By.ID, 'logxinid')

        loginid.send_keys(account)

        pincode=browser.find_element(By.ID, 'pincode')

        pincode.send_keys(password)

        div_btn_grp=browser.find_element(By.CLASS_NAME, 'btn_grp')

        login_btn=div_btn_grp.find_element(By.TAG_NAME, 'input')

        login_btn.click()

        # 擷取借閱紀錄

        div_redblock=browser.find_element(By.CLASS_NAME, 'redblock')

        div_redblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        borrow_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

            reg=r'\d{4}-\d{2}-\d{2}'

            due_date=book.find_element(By.XPATH, './ul[4]/li[2]').text

            item['due_date']=re.findall(reg, due_date)[0] 

            due_times=book.find_element(By.XPATH, './ul[5]/li[1]').text

            item['due_times']=re.findall(r'\d{1}', due_times)[0]

            try: 

                state=book.find_element(By.XPATH, './ul[6]/li[1]').text

            except:

                state=''

            finally:

                if '有人預約' in state:

                    item['state']=', 有人預約'

                else:

                    item['state']=''

            borrow_books.append(item)

        browser.back() # 回上一頁

        # 擷取預約紀錄

        div_blueblock=browser.find_element(By.CLASS_NAME, 'blueblock')

        div_blueblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        reserve_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            sequence=book.find_element(By.XPATH, './ul[7]/li[1]').text

            if '預約待取' in sequence:  # 已到館

                item['ready_for_pickup']=True

                reg=r'\d{4}-\d{2}-\d{2}'

                item['expiration']=re.findall(reg, sequence)[0]

                item['sequence']='0'

            else: # 預約中

                item['ready_for_pickup']=False

                item['expiration']=''

                item['sequence']=re.findall(r'\d+', sequence)[0]

            reserve_books.append(item)

        browser.close()

        return (borrow_books, reserve_books)        

    except Exception as e:

        print(e)

        return None, None

    

if __name__ == '__main__':

    start=time.time()

    token='Telegram 權杖'

    chat_id='Telegram 聊天室 ID'

    if len(sys.argv) != 3:

        print('用法: python3 ksml_personal_10.py 帳號 密碼')

        sys.exit(1)

    # 取得傳入的帳密參數

    account=sys.argv[1]

    password=sys.argv[2]

    # 呼叫 get_books() 取得借書與預約書        

    borrow_books, reserve_books=get_books(account, password)

    b_msg=''  # 借書資訊字串初始值

    r_msg=''  # 預約資訊字串初始值

    # 處理借書 

    if borrow_books: 

        borrow=[]

        for book in borrow_books:

            book_name=book['book_name']

            book_site=book['book_site'] 

            due_times=book['due_times']

            due_date=book['due_date']

            state=book['state']

            due_date=datetime.strptime(due_date, '%Y-%m-%d') # 到期日   

            today_str=datetime.today().strftime('%Y-%m-%d')   

            today=datetime.strptime(today_str, "%Y-%m-%d")   

            delta=(due_date-today).days  # 計算離到期日還有幾天

            if delta < 0:  # 負數=已逾期

                msg=f'🅧 {book_name} (逾期 {abs(delta)} 天{state}, {book_site})'

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg=f'⓿ {book_name} (今日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg=f'❶ {book_name} (明日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg=f'❷ {book_name} (後天到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg=f'✦ {book_name} ({book["due_date"]} 到期, '\

                    f'續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

        # 傳送借閱書到期摘要至 Telegram 

        if len(borrow) != 0:

            borrow.insert(0, f'\n❖ {account} 的借閱 :')

            b_msg='\n'.join(borrow)  # 更新借書資訊字串

            if asyncio.run(telegram_send_text(b_msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    # 處理預約書

    if reserve_books:

        reserve=[]

        i=0

        j=['①', '②', '③', '④', '⑤']

        k=['❶', '❷', '❸', '❹', '❺']

        # 預約狀態

        for book in reserve_books:

            book_name=book['book_name']

            sequence=book['sequence']

            ready_for_pickup=book['ready_for_pickup'] # 已到館

            expiration=book['expiration']  # 取書截止日

            if ready_for_pickup:

                msg=f'{k[i]} {book_name} (已到館, 保留期限 {expiration})'

            else:

                msg=f'{j[i]} {book_name} (順位 {sequence})'

            reserve.append(msg)

            i += 1

        # 傳送預約書狀態摘要至 Telegram    

        if len(reserve) != 0:

            reserve.insert(0, f'\n❖ {account} 的預約 :')

            r_msg='\n'.join(reserve)  # 更新資訊字串

            if asyncio.run(telegram_send_text(r_msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    if b_msg or r_msg:  # 任一不為空字串就更新資料表

        url="https://serverless-fdof.onrender.com/function/update_ksml_books"

        payload={

            "account": account,

            "borrow_books": b_msg,   

            "reserve_books": r_msg

            }

        res=requests.post(url, json=payload)

        print(res.json())        

    end=time.time()

    print(f'執行時間:{end-start}')

主要的變更在於把用來儲存借書與預約資訊的 msg 變數分成 b_msg (借書) 與 r_msg (預約), 初始值設為空字串以便在最後面用來判別是否需要更新 ksml_books 資料表 (兩者均為空字串不用更新), 使用兩張借書證測試結果如下 : 

資料表列表顯示 ksml_books 有兩筆紀錄, 按檢視會顯示全部紀錄 :

為了測試方便我仍然保留將借書與預約資訊傳送到 Telegram 的功能, 但這部分程式碼在正式佈署到樹莓派 Pi 3 時會被刪除, 因為市圖爬蟲可能每一或二小時會爬一次來更新 ksml_books 資料表, 如果每次爬完都傳一次訊息感覺聊天室同樣訊息會太多, 取而代之是另外用一個 crontab 專責讀取資料表傳訊息, 傳訊機器人一天只需要中午與晚上各傳一次 (因應預約書到館時間), 會分別傳送到 LINE 與 Telegram. 

佈署樹莓派 Pi 3 的程式碼如下 (使用無頭模式的 Chronium) :

# ksml_personal_11_deploy.py

from selenium import webdriver

from selenium.webdriver.common.by import By

from selenium.webdriver.chrome.options import Options

from selenium.webdriver.chrome.service import Service

import re

from datetime import datetime

import time

import requests

import sys

async def telegram_send_text(text):

    bot=Bot(token=token)

    try:

        await bot.send_message(

            chat_id=chat_id,

            text=text

            )

        return True

    except Exception as e:

        print(f'Error sending text: {e}')

        return False

def get_books(account, password):

    try:

        # 登入我的書房

        options=Options()

        options.add_argument("--headless")

        driverpath='/usr/lib/chromium-browser/chromedriver'

        service=Service(driverpath)

        browser=webdriver.Chrome(options=options, service=service)

        browser.implicitly_wait(60)

        browser.get('https://webpacx.ksml.edu.tw/personal/')

        loginid=browser.find_element(By.ID, 'logxinid')

        loginid.send_keys(account)

        pincode=browser.find_element(By.ID, 'pincode')

        pincode.send_keys(password)

        div_btn_grp=browser.find_element(By.CLASS_NAME, 'btn_grp')

        login_btn=div_btn_grp.find_element(By.TAG_NAME, 'input')

        login_btn.click()

        # 擷取借閱紀錄

        div_redblock=browser.find_element(By.CLASS_NAME, 'redblock')

        div_redblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        borrow_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

            reg=r'\d{4}-\d{2}-\d{2}'

            due_date=book.find_element(By.XPATH, './ul[4]/li[2]').text

            item['due_date']=re.findall(reg, due_date)[0] 

            due_times=book.find_element(By.XPATH, './ul[5]/li[1]').text

            item['due_times']=re.findall(r'\d{1}', due_times)[0]

            try: 

                state=book.find_element(By.XPATH, './ul[6]/li[1]').text

            except:

                state=''

            finally:

                if '有人預約' in state:

                    item['state']=', 有人預約'

                else:

                    item['state']=''

            borrow_books.append(item)

        browser.back() # 回上一頁

        # 擷取預約紀錄

        div_blueblock=browser.find_element(By.CLASS_NAME, 'blueblock')

        div_blueblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        reserve_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            sequence=book.find_element(By.XPATH, './ul[7]/li[1]').text

            if '預約待取' in sequence:  # 已到館

                item['ready_for_pickup']=True

                reg=r'\d{4}-\d{2}-\d{2}'

                item['expiration']=re.findall(reg, sequence)[0]

                item['sequence']='0'

            else: # 預約中

                item['ready_for_pickup']=False

                item['expiration']=''

                item['sequence']=re.findall(r'\d+', sequence)[0]

            reserve_books.append(item)

        browser.close()

        return (borrow_books, reserve_books)        

    except Exception as e:

        print(e)

        return None, None

    

if __name__ == '__main__':

    start=time.time()

    token='Telegram 權杖'

    chat_id='Telegram 聊天室 ID'

    if len(sys.argv) != 3:

        print('用法: python3 ksml_personal_10.py 帳號 密碼')

        sys.exit(1)

    # 取得傳入的帳密參數

    account=sys.argv[1]

    password=sys.argv[2]

    # 呼叫 get_books() 取得借書與預約書        

    borrow_books, reserve_books=get_books(account, password)

    b_msg=''  # 借書資訊字串初始值

    r_msg=''  # 預約資訊字串初始值

    # 處理借書 

    if borrow_books: 

        borrow=[]

        for book in borrow_books:

            book_name=book['book_name']

            book_site=book['book_site'] 

            due_times=book['due_times']

            due_date=book['due_date']

            state=book['state']

            due_date=datetime.strptime(due_date, '%Y-%m-%d') # 到期日   

            today_str=datetime.today().strftime('%Y-%m-%d')   

            today=datetime.strptime(today_str, "%Y-%m-%d")   

            delta=(due_date-today).days  # 計算離到期日還有幾天

            if delta < 0:  # 負數=已逾期

                msg=f'🅧 {book_name} (逾期 {abs(delta)} 天{state}, {book_site})'

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg=f'⓿ {book_name} (今日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg=f'❶ {book_name} (明日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg=f'❷ {book_name} (後天到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg=f'✦ {book_name} ({book["due_date"]} 到期, '\

                    f'續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

        # 製作借書到期摘要字串 

        if len(borrow) != 0:

            borrow.insert(0, f'\n❖ {account} 的借閱 :')

            b_msg='\n'.join(borrow)  # 更新借書資訊字串

    # 處理預約書

    if reserve_books:

        reserve=[]

        i=0

        j=['①', '②', '③', '④', '⑤']

        k=['❶', '❷', '❸', '❹', '❺']

        # 預約狀態

        for book in reserve_books:

            book_name=book['book_name']

            sequence=book['sequence']

            ready_for_pickup=book['ready_for_pickup'] # 已到館

            expiration=book['expiration']  # 取書截止日

            if ready_for_pickup:

                msg=f'{k[i]} {book_name} (已到館, 保留期限 {expiration})'

            else:

                msg=f'{j[i]} {book_name} (順位 {sequence})'

            reserve.append(msg)

            i += 1

        # 製作預約書摘要字串    

        if len(reserve) != 0:

            reserve.insert(0, f'\n❖ {account} 的預約 :')

            r_msg='\n'.join(reserve)  # 更新資訊字串

    if b_msg or r_msg:  # 任一不為空字串就更新資料表

        url="https://serverless-fdof.onrender.com/function/update_ksml_books"

        payload={

            "account": account,

            "borrow_books": b_msg,   

            "reserve_books": r_msg

            }

        res=requests.post(url, json=payload)

        print(res.json())        

    end=time.time()

    print(f'執行時間:{end-start}')        

此處已拿掉 Telegram 傳送訊息的程式碼, 只將借書與預約資訊更新資料表. 

用 chmod 指令將此佈署版爬蟲改為可執行 :

pi@raspberrypi:~ $ sudo chmod +x ksml_personal_11_deploy.py  

pi@raspberrypi:~ $ ls -ls ksml_personal_11_deploy.py  

8 -rwxr-xr-x 1 pi pi 7218 10月 20 23:29 ksml_personal_11_deploy.py  

4. 撰寫 serverless 平台的傳訊程式 : 

此程式不是樹莓派本地端程式, 而是要佈署在 serverless 平台上的函式模組, 當它被呼叫時會從 serverless.db 資料庫的 ksml_books 資料表裡讀取各借書證之借書與預約欄位, 然後傳送到 Telegram, 程式碼如下 (後續會加上傳送至 LINE 聊天室之功能) :

# send_ksml_books_messages.py

import asyncio

import sqlite3

from telegram import Bot

async def telegram_send_text(token, chat_id, text):

    """非同步傳送 Telegram 訊息"""

    try:

        bot=Bot(token=token)

        await bot.send_message(chat_id=chat_id, text=text)

        return True

    except Exception as e:

        print(f"傳送失敗: {e}")

        return False

def main(request, **kwargs):

    DB_PATH='./serverless.db'

    config=kwargs.get('config', {})

    telegram_token=config.get('TELEGRAM_TOKEN')

    telegram_id=config.get('TELEGRAM_ID')

    if not telegram_token or not telegram_id:

        return '未設定 TELEGRAM_TOKEN 或 TELEGRAM_ID'

    try:  # 連線資料庫

        conn=sqlite3.connect(DB_PATH)

        cur=conn.cursor()

        cur.execute("SELECT borrow_books, reserve_books FROM ksml_books;")

        rows=cur.fetchall()

        conn.close()

    except Exception as e:

        return f'資料庫讀取失敗: {e}'

    if not rows:

        return '沒有任何資料可傳送'

    # 傳送訊息

    success_count=0

    fail_count=0

    for borrow_books, reserve_books in rows:

        for msg in [borrow_books, reserve_books]:

            if msg and msg.strip():

                ok=asyncio.run(telegram_send_text(telegram_token, telegram_id, msg))

                if ok:

                    success_count += 1

                else:

                    fail_count += 1

    return f'傳送完成：成功 {success_count} 筆，失敗 {fail_count} 筆'

此程式會先從主程式 serverless.py 傳來的 kwargs 參數取出 config 字典, 從中取得 Telegram 的權杖與聊天室 id, 然後從 sml_books 資料表取得全部借書證之借書與預約資訊, 用迴圈拜訪每個紀錄逐一傳送到 Telegram 聊天室. 

在 serverless 平台 (例如 render.com) 新增函式 send_books_messages.py 後, 在它的 "執行" 連結上按滑鼠右鍵選取 "複製連結網址" 取得 API 網址 : 

只要在樹莓派寫個 get_books_messages.py 程式, 用 requests 送出 HTTP GET 請求到此 serverless API 網址, 則 send_books_messages.py 就會將最新的借書與預約資訊送到 Telegram 聊天室, 這個本機程式如下 : 

# get_books_messages.py

import requests

url="https://serverless-fdof.onrender.com/function/send_books_messages"

res=requests.get(url)

print(res)

用 chmod 指令更改此程式權限為可執行 :

pi@raspberrypi:~ $ sudo chmod +x get_books_messages.py   

pi@raspberrypi:~ $ ls -ls get_books_messages.py   

4 -rwxr-xr-x 1 pi pi 167 10月 22 23:37 get_books_messages.py

接下來就可以將它放在 crontab 中定期執行. 

3. 修改樹莓派 Pi 3 的 crontab : 

接下來要修改樹莓派的 crontab, 之前的爬蟲程式 (v9) 將爬取與傳訊一手包, 每爬完一張借書證就將結果傳到 Telegram 聊天室, crontab 設定為一天爬兩次; 上面新版 v11 程式只單純地將爬到的結果儲存到 serverless 平台的資料表裡, 所以可以提高爬取頻率為每小時爬一次 (每張借書證間從 0 分起間隔 4 分鐘), 設定修改如下 : 

pi@raspberrypi:~ $ crontab -l 

0 16 * * 1-5 /usr/bin/python3 /home/pi/twstock_dashboard_update.py

*/31 9-13 * * 1-5 /usr/bin/python3 /home/pi/yahoo_twstock_monitor_table_2.py

0 8,18 * * * /usr/bin/python3 /home/pi/btc_eth_prices_line_notify.py

1 12,17 * * * /usr/bin/python3 /home/pi/technews_3.py

0 9 * * * /usr/bin/python3 /home/pi/books.com.tw_66_telegram.py

0 * * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py f-----7 5-----1

4 * * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py 0-----8 9-----7

8 * * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py 0-----9 8-----6

12 * * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py 0------0 8-----4

16 * * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py 9------2 6-----3

20 * * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py S------7 3-----0

30 12, 21 * * * /usr/bin/python3 /home/pi/ksml_personal_11_deploy.py

0 6,16 * * * /usr/bin/python3 /home/pi/nkust_lib_7.py

*/5 * * * * ~/duckdns/duck.sh >/dev/null 2>&1

此處設定每天 12:30 與 21:30 會各傳送一次借書與預約資訊到 Telegram. 

Python 學習筆記 : 市圖借書與預約爬蟲程式改版 v10

今年四月因應 LINE Notify 終止服務, 圖書館爬蟲程式改版為 v9, 改用 Telegram 取代 LINE Notify 來傳送訊息, 運作均正常 (但是不太習慣). 之前的相關筆記參考 : 

# Python 學習筆記 : 網頁爬蟲實戰 (九) 市立圖書館個人書房借書資訊 (上)

# Python 學習筆記 : 網頁爬蟲實戰 (九) 市立圖書館個人書房借書資訊 (中)

# Python 學習筆記 : 網頁爬蟲實戰 (九) 市立圖書館個人書房借書資訊 (下)

# Python 學習筆記 : 用 Telegram 傳送市圖借書與預約訊息

前三篇是 LINE Notify 版; 第四篇是 Telegram 版. 

最近完成 render.com 雲端平台測試後打算將此爬蟲程式改版, 目標如下 :

• 近一周將到期的書, 增加抓 "典藏地" 資訊
• 放棄用迴圈查詢全部借書證做法, 改為執行時帶入帳密參數, 一次只查一張借書證
• 借書資訊存入 Mapleboard 的 serverless 平台資料庫 ksml_books 供 LINE Bot 查詢
• 樹莓派定時爬取圖書館網頁更新 serverless 資料庫

本篇先來處理如何抓典藏地資訊, 登入市圖網站的個人書房, 按 F12 開啟開發者頁面, 按 Element 左上角的箭頭可輕易找到書籍典藏地所對應的 HTML 位置 : 

按右鍵選擇 Copy/Copy Xpath 即可印證典藏地的位置是 ul[3]li[1] :

//*[@id="center"]/div/div/div[2]/div/div[6]/form/div/div[2]/div[2]/div[1]/ul[3]/li[1]/text()[2]

因此只要在借閱書籍 books 迴圈中增加 book_site 即可 :

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

此處使用正規式 (\S+) 來取出 '典藏地：' 後面的字串, 然後存入 item 字典的 book_site 鍵中. 然後在主程式的 borrow_books 迴圈裡取出 book_site 資訊, 加入借閱書籍資訊中 :

            book_site=book['book_site']

            ... (略) ...

            if delta < 0:  # 負數=已逾期

                msg=f'🅧 {book_name} (逾期 {abs(delta)} 天{state}, {book_site})'

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg=f'⓿ {book_name} (今日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg=f'❶ {book_name} (明日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg=f'❷ {book_name} (後天到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg=f'✦ {book_name} ({book["due_date"]} 到期, '\

                    f'續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

修改後的完整程式碼如下 :

# ksml_personal_10.py

from selenium import webdriver

from selenium.webdriver.common.by import By

from selenium.webdriver.firefox.options import Options

import re

from datetime import datetime

import time

import requests

import asyncio

from telegram import Bot

import sys  

async def telegram_send_text(text):

    bot=Bot(token=token)

    try:

        await bot.send_message(

            chat_id=chat_id,

            text=text

            )

        return True

    except Exception as e:

        print(f'Error sending text: {e}')

        return False

def get_books(account, password):

    try:

        # 登入我的書房

        options=Options()

        options.add_argument("--headless")

        driverpath='geckodriver.exe'

        browser=webdriver.Firefox(options=options)

        browser.implicitly_wait(60)

        browser.get('https://webpacx.ksml.edu.tw/personal/')

        loginid=browser.find_element(By.ID, 'logxinid')

        loginid.send_keys(account)

        pincode=browser.find_element(By.ID, 'pincode')

        pincode.send_keys(password)

        div_btn_grp=browser.find_element(By.CLASS_NAME, 'btn_grp')

        login_btn=div_btn_grp.find_element(By.TAG_NAME, 'input')

        login_btn.click()

        # 擷取借閱紀錄

        div_redblock=browser.find_element(By.CLASS_NAME, 'redblock')

        div_redblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        borrow_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

            reg=r'\d{4}-\d{2}-\d{2}'

            due_date=book.find_element(By.XPATH, './ul[4]/li[2]').text

            item['due_date']=re.findall(reg, due_date)[0] 

            due_times=book.find_element(By.XPATH, './ul[5]/li[1]').text

            item['due_times']=re.findall(r'\d{1}', due_times)[0]

            try: 

                state=book.find_element(By.XPATH, './ul[6]/li[1]').text

            except:

                state=''

            finally:

                if '有人預約' in state:

                    item['state']=', 有人預約'

                else:

                    item['state']=''

            borrow_books.append(item)

        browser.back() # 回上一頁

        # 擷取預約紀錄

        div_blueblock=browser.find_element(By.CLASS_NAME, 'blueblock')

        div_blueblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        reserve_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            sequence=book.find_element(By.XPATH, './ul[7]/li[1]').text

            if '預約待取' in sequence:  # 已到館

                item['ready_for_pickup']=True

                reg=r'\d{4}-\d{2}-\d{2}'

                item['expiration']=re.findall(reg, sequence)[0]

                item['sequence']='0'

            else: # 預約中

                item['ready_for_pickup']=False

                item['expiration']=''

                item['sequence']=re.findall(r'\d+', sequence)[0]

            reserve_books.append(item)

        browser.close()

        return (borrow_books, reserve_books)        

    except Exception as e:

        print(e)

        return None, None

    

if __name__ == '__main__':

    start=time.time()

    token='Telegram 權杖'

    chat_id='Telegram 聊天室 ID'

    if len(sys.argv) != 3:

        print('用法: python3 ksml_personal_10.py 帳號 密碼')

        sys.exit(1)

    # 取得傳入的帳密參數

    account=sys.argv[1]

    password=sys.argv[2]

    # 呼叫 get_books() 取得借書與預約書        

    borrow_books, reserve_books=get_books(account, password)

    # 處理借書 

    if borrow_books: 

        borrow=[]

        for book in borrow_books:

            book_name=book['book_name']

            book_site=book['book_site'] 

            due_times=book['due_times']

            due_date=book['due_date']

            state=book['state']

            due_date=datetime.strptime(due_date, '%Y-%m-%d') # 到期日   

            today_str=datetime.today().strftime('%Y-%m-%d')   

            today=datetime.strptime(today_str, "%Y-%m-%d")   

            delta=(due_date-today).days  # 計算離到期日還有幾天

            if delta < 0:  # 負數=已逾期

                msg=f'🅧 {book_name} (逾期 {abs(delta)} 天{state}, {book_site})'

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg=f'⓿ {book_name} (今日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg=f'❶ {book_name} (明日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg=f'❷ {book_name} (後天到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg=f'✦ {book_name} ({book["due_date"]} 到期, '\

                    f'續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

        # 傳送借閱書到期摘要至 Telegram 

        if len(borrow) != 0:

            borrow.insert(0, f'\n❖ {account} 的借閱 :')

            msg='\n'.join(borrow)

            if asyncio.run(telegram_send_text(msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    # 處理預約書

    if reserve_books:

        reserve=[]

        i=0

        j=['①', '②', '③', '④', '⑤']

        k=['❶', '❷', '❸', '❹', '❺']

        # 預約狀態

        for book in reserve_books:

            book_name=book['book_name']

            sequence=book['sequence']

            ready_for_pickup=book['ready_for_pickup'] # 已到館

            expiration=book['expiration']  # 取書截止日

            if ready_for_pickup:

                msg=f'{k[i]} {book_name} (已到館, 保留期限 {expiration})'

            else:

                msg=f'{j[i]} {book_name} (順位 {sequence})'

            reserve.append(msg)

            i += 1

        # 傳送預約書狀態摘要至 Telegram    

        if len(reserve) != 0:

            reserve.insert(0, f'\n❖ {account} 的預約 :')

            msg='\n'.join(reserve)

            if asyncio.run(telegram_send_text(msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    else:

        print('無資料')                    

    end=time.time()

    print(f'執行時間:{end-start}')

此程式使用 Selenium 透過 Firefox 瀏覽器以無頭模式爬取圖書館借閱資訊, Firefox 要

更新到最新版, 同時到下列網址下載最新的 Firefox webdriver (x86 64 位元) geckodriver :

# https://github.com/mozilla/geckodriver/releases   

執行時要傳入帳密參數 (程式比舊版多了 import sys 來取得參數), 結果如下 : 

D:\python\test>python ksml_personal_10.py myhome 123456     

訊息傳送成功!

訊息傳送成功!

執行時間:82.54115891456604

可見新版程式已成功地爬取到典藏地資訊了. 

2025-10-24 補充 :

雖然市圖爬蟲最新改版目標 (參考 v11) 是改為每日多次爬取後儲存最新資訊於 serverless 平台的資料表 ksml_books, 然後由樹莓派觸發 serverless 每日傳送兩次從資料表取得知最新訊息到 Telegram, 此模式需本地樹莓派與雲端 serverless 平台搭配才能運作, 好處是可以用在 LINE Bot 操作. 但我還是保留全部由樹莓派一手包辦爬取+發送 Telegram 的模式, 將上面於 Windows 上用 Firefox 測試的 v10 程式改為用 Chromnium 

# ksml_personal_10_deploy.py

from selenium import webdriver

from selenium.webdriver.common.by import By

from selenium.webdriver.chrome.options import Options

from selenium.webdriver.chrome.service import Service

import re

from datetime import datetime

import time

import requests

import asyncio

from telegram import Bot

import sys

async def telegram_send_text(text):

    bot=Bot(token=token)

    try:

        await bot.send_message(

            chat_id=chat_id,

            text=text

            )

        return True

    except Exception as e:

        print(f'Error sending text: {e}')

        return False

def get_books(account, password):

    try:

        # 登入我的書房

        options=Options()

        options.add_argument("--headless")

        driverpath='/usr/lib/chromium-browser/chromedriver'

        service=Service(driverpath)

        browser=webdriver.Chrome(options=options, service=service)

        browser.implicitly_wait(60)

        browser.get('https://webpacx.ksml.edu.tw/personal/')

        loginid=browser.find_element(By.ID, 'logxinid')

        loginid.send_keys(account)

        pincode=browser.find_element(By.ID, 'pincode')

        pincode.send_keys(password)

        div_btn_grp=browser.find_element(By.CLASS_NAME, 'btn_grp')

        login_btn=div_btn_grp.find_element(By.TAG_NAME, 'input')

        login_btn.click()

        # 擷取借閱紀錄

        div_redblock=browser.find_element(By.CLASS_NAME, 'redblock')

        div_redblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        borrow_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

            reg=r'\d{4}-\d{2}-\d{2}'

            due_date=book.find_element(By.XPATH, './ul[4]/li[2]').text

            item['due_date']=re.findall(reg, due_date)[0] 

            due_times=book.find_element(By.XPATH, './ul[5]/li[1]').text

            item['due_times']=re.findall(r'\d{1}', due_times)[0]

            try: 

                state=book.find_element(By.XPATH, './ul[6]/li[1]').text

            except:

                state=''

            finally:

                if '有人預約' in state:

                    item['state']=', 有人預約'

                else:

                    item['state']=''

            borrow_books.append(item)

        browser.back() # 回上一頁

        # 擷取預約紀錄

        div_blueblock=browser.find_element(By.CLASS_NAME, 'blueblock')

        div_blueblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        reserve_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            sequence=book.find_element(By.XPATH, './ul[7]/li[1]').text

            if '預約待取' in sequence:  # 已到館

                item['ready_for_pickup']=True

                reg=r'\d{4}-\d{2}-\d{2}'

                item['expiration']=re.findall(reg, sequence)[0]

                item['sequence']='0'

            else: # 預約中

                item['ready_for_pickup']=False

                item['expiration']=''

                item['sequence']=re.findall(r'\d+', sequence)[0]

            reserve_books.append(item)

        browser.close()

        return (borrow_books, reserve_books)        

    except Exception as e:

        print(e)

        return None, None

    

if __name__ == '__main__':

    start=time.time()

    token='我的 Telegram 權杖'

    chat_id='Telegram 聊天室 ID'

    if len(sys.argv) != 3:

        print('用法: python3 ksml_personal_10_deploy.py 帳號 密碼')

        sys.exit(1)

    # 取得傳入的帳密參數

    account=sys.argv[1]

    password=sys.argv[2]

    # 呼叫 get_books() 取得借書與預約書        

    borrow_books, reserve_books=get_books(account, password)

    # 處理借書 

    if borrow_books: 

        borrow=[]

        for book in borrow_books:

            book_name=book['book_name']

            book_site=book['book_site'] 

            due_times=book['due_times']

            due_date=book['due_date']

            state=book['state']

            due_date=datetime.strptime(due_date, '%Y-%m-%d') # 到期日   

            today_str=datetime.today().strftime('%Y-%m-%d')   

            today=datetime.strptime(today_str, "%Y-%m-%d")   

            delta=(due_date-today).days  # 計算離到期日還有幾天

            if delta < 0:  # 負數=已逾期

                msg=f'🅧 {book_name} (逾期 {abs(delta)} 天{state}, {book_site})'

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg=f'⓿ {book_name} (今日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg=f'❶ {book_name} (明日到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg=f'❷ {book_name} (後天到期, 續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg=f'✦ {book_name} ({book["due_date"]} 到期, '\

                    f'續借次數 {due_times}{state}, {book_site})'

                borrow.append(msg)

        # 傳送借閱書到期摘要至 Telegram 

        if len(borrow) != 0:

            borrow.insert(0, f'\n❖ {account} 的借閱 :')

            msg='\n'.join(borrow)

            if asyncio.run(telegram_send_text(msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    # 處理預約書

    if reserve_books:

        reserve=[]

        i=0

        j=['①', '②', '③', '④', '⑤']

        k=['❶', '❷', '❸', '❹', '❺']

        # 預約狀態

        for book in reserve_books:

            book_name=book['book_name']

            sequence=book['sequence']

            ready_for_pickup=book['ready_for_pickup'] # 已到館

            expiration=book['expiration']  # 取書截止日

            if ready_for_pickup:

                msg=f'{k[i]} {book_name} (已到館, 保留期限 {expiration})'

            else:

                msg=f'{j[i]} {book_name} (順位 {sequence})'

            reserve.append(msg)

            i += 1

        # 傳送預約書狀態摘要至 Telegram    

        if len(reserve) != 0:

            reserve.insert(0, f'\n❖ {account} 的預約 :')

            msg='\n'.join(reserve)

            if asyncio.run(telegram_send_text(msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    else:

        print('無資料')                    

    end=time.time()

    print(f'執行時間:{end-start}')

然後 crontab 修改為如下格式 :

0 12, 21 * * * /usr/bin/python3 /home/pi/ksml_personal_10_deploy.py 帳號 密碼

每個帳號間隔約 4 分鐘即可. 

我將上面程式佈署至鄉下的 Pi 3 發現執行錯誤, 因為 OS 是 Raspbian Strecth 的 Python 是 3.5 版, 不支援 f 字串, 所以我改成用 format() :

# ksml_personal_10_deploy.py

from selenium import webdriver

from selenium.webdriver.common.by import By

from selenium.webdriver.chrome.options import Options

from selenium.webdriver.chrome.service import Service

import re

from datetime import datetime

import time

import requests

import asyncio

from telegram import Bot

import sys

async def telegram_send_text(text):

    bot=Bot(token=token)

    try:

        await bot.send_message(

            chat_id=chat_id,

            text=text

            )

        return True

    except Exception as e:

        print('Error sending text: '.format(e))

        return False

def get_books(account, password):

    try:

        # 登入我的書房

        options=Options()

        options.add_argument("--headless")

        driverpath='/usr/lib/chromium-browser/chromedriver'

        service=Service(driverpath)

        browser=webdriver.Chrome(options=options, service=service)

        browser.implicitly_wait(60)

        browser.get('https://webpacx.ksml.edu.tw/personal/')

        loginid=browser.find_element(By.ID, 'logxinid')

        loginid.send_keys(account)

        pincode=browser.find_element(By.ID, 'pincode')

        pincode.send_keys(password)

        div_btn_grp=browser.find_element(By.CLASS_NAME, 'btn_grp')

        login_btn=div_btn_grp.find_element(By.TAG_NAME, 'input')

        login_btn.click()

        # 擷取借閱紀錄

        div_redblock=browser.find_element(By.CLASS_NAME, 'redblock')

        div_redblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        borrow_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

            reg=r'\d{4}-\d{2}-\d{2}'

            due_date=book.find_element(By.XPATH, './ul[4]/li[2]').text

            item['due_date']=re.findall(reg, due_date)[0] 

            due_times=book.find_element(By.XPATH, './ul[5]/li[1]').text

            item['due_times']=re.findall(r'\d{1}', due_times)[0]

            try: 

                state=book.find_element(By.XPATH, './ul[6]/li[1]').text

            except:

                state=''

            finally:

                if '有人預約' in state:

                    item['state']=', 有人預約'

                else:

                    item['state']=''

            borrow_books.append(item)

        browser.back() # 回上一頁

        # 擷取預約紀錄

        div_blueblock=browser.find_element(By.CLASS_NAME, 'blueblock')

        div_blueblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        reserve_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            sequence=book.find_element(By.XPATH, './ul[7]/li[1]').text

            if '預約待取' in sequence:  # 已到館

                item['ready_for_pickup']=True

                reg=r'\d{4}-\d{2}-\d{2}'

                item['expiration']=re.findall(reg, sequence)[0]

                item['sequence']='0'

            else: # 預約中

                item['ready_for_pickup']=False

                item['expiration']=''

                item['sequence']=re.findall(r'\d+', sequence)[0]

            reserve_books.append(item)

        browser.close()

        return (borrow_books, reserve_books)        

    except Exception as e:

        print(e)

        return None, None

    

if __name__ == '__main__':

    start=time.time()

    token='7938146214:AAGoV71AVEygN3e-j-ZfOWUcWSHG1qBp7KE'

    chat_id='7742981072'

    if len(sys.argv) != 3:

        print('用法: python3 ksml_personal_10.py 帳號 密碼')

        sys.exit(1)

    # 取得傳入的帳密參數

    account=sys.argv[1]

    password=sys.argv[2]

    # 呼叫 get_books() 取得借書與預約書        

    borrow_books, reserve_books=get_books(account, password)

    # 處理借書 

    if borrow_books: 

        borrow=[]

        for book in borrow_books:

            book_name=book['book_name']

            book_site=book['book_site'] 

            due_times=book['due_times']

            due_date=book['due_date']

            state=book['state']

            due_date=datetime.strptime(due_date, '%Y-%m-%d') # 到期日   

            today_str=datetime.today().strftime('%Y-%m-%d')   

            today=datetime.strptime(today_str, "%Y-%m-%d")   

            delta=(due_date-today).days  # 計算離到期日還有幾天

            if delta < 0:  # 負數=已逾期

                msg='🅧 {} (逾期 {} 天{}, {})'.format(book_name, abs(delta), state, book_site)

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg='⓿ {} (今日到期, 續借次數 {}{}, {})'.format(book_name, due_times, state, book_site)

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg='❶ {} (明日到期, 續借次數 {}{}, {})'.format(book_name, due_times, state, book_site)

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg='❷ {} (後天到期, 續借次數 {}{}, {})'.format(book_name, due_times, state, book_site)

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg='✦ {} ({} 到期, '.format(book_name, book["due_date"]) +\ 

                    '續借次數 {}{}, {})'.format(due_times, state, book_site)

                borrow.append(msg)

        # 傳送借閱書到期摘要至 Telegram 

        if len(borrow) != 0:

            borrow.insert(0, '\n❖ {} 的借閱 :'.format(account))

            msg='\n'.join(borrow)

            if asyncio.run(telegram_send_text(msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    # 處理預約書

    if reserve_books:

        reserve=[]

        i=0

        j=['①', '②', '③', '④', '⑤']

        k=['❶', '❷', '❸', '❹', '❺']

        # 預約狀態

        for book in reserve_books:

            book_name=book['book_name']

            sequence=book['sequence']

            ready_for_pickup=book['ready_for_pickup'] # 已到館

            expiration=book['expiration']  # 取書截止日

            if ready_for_pickup:

                msg='{} {} (已到館, 保留期限 {})'.format(k[i], book_name, expiration)

            else:

                msg='{} {} (順位 {})'.format(j[i], book_name, sequence)

            reserve.append(msg)

            i += 1

        # 傳送預約書狀態摘要至 Telegram    

        if len(reserve) != 0:

            reserve.insert(0, '\n❖ {} 的預約 :'.format(account))

            msg='\n'.join(reserve)

            if asyncio.run(telegram_send_text(msg)):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    else:

        print('無資料')                    

    end=time.time()

    print('執行時間:{}'.format(end-start))

執行出現另一個錯誤 : 

Traceback (most recent call last):

  File "ksml_personal_10_deploy.py", line 11, in <module>

    from telegram import Bot

  File "/home/pi/.local/lib/python3.5/site-packages/telegram/__init__.py", line 21, in <module>

    from .base import TelegramObject

  File "/home/pi/.local/lib/python3.5/site-packages/telegram/base.py", line 42

    _id_attrs: Tuple[Any, ...] = ()

             ^

SyntaxError: invalid syntax

詢問 ChatGPT 才知道是因為 Python 3.5 太舊了, python-telegram-bot 雖然能安裝, 但如果使用 Bot 物件傳訊息執行時會出錯, 須改用 HTTP 呼叫; 而且 Cromium driver 也必須用舊版寫法 (不能用 Service), 我重新安裝了 chromium-chromedriver 套件 : 

sudo apt-get install -y chromium-chromedriver

然後將程式改為舊版 driver 語法之後就可以在這台 Pi 3 的 Raspbian Stretch 上順利用 Selenium 爬取資料並將結果傳到 Telegram, 程式碼如下 :

# ksml_personal_10_deploy.py

from selenium import webdriver

from selenium.webdriver.common.by import By

from selenium.webdriver.chrome.options import Options

import re

from datetime import datetime

import time

import requests

import sys

def telegram_send_text(text):

    try:

        data={'chat_id': chat_id, 'text': text} 

        url='https://api.telegram.org/bot{}/sendMessage'.format(token)

        r=requests.post(url, data=data)

        return True

    except Exception as e:

        print('Error sending text: '.format(e))

        return False

def get_books(account, password):

    try:

        # 登入我的書房

        chrome_options=Options()

        chrome_options.add_argument("--headless")

        chrome_options.add_argument("--no-sandbox")

        chrome_options.add_argument("--disable-dev-shm-usage")

        # 舊版 selenium 用 executable_path

        browser=webdriver.Chrome(executable_path="/usr/bin/chromedriver", options=chrome_options)

        browser.implicitly_wait(60)

        browser.get('https://webpacx.ksml.edu.tw/personal/')

        loginid=browser.find_element(By.ID, 'logxinid')

        loginid.send_keys(account)

        pincode=browser.find_element(By.ID, 'pincode')

        pincode.send_keys(password)

        div_btn_grp=browser.find_element(By.CLASS_NAME, 'btn_grp')

        login_btn=div_btn_grp.find_element(By.TAG_NAME, 'input')

        login_btn.click()

        # 擷取借閱紀錄

        div_redblock=browser.find_element(By.CLASS_NAME, 'redblock')

        div_redblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        borrow_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            book_site=book.find_element(By.XPATH, './ul[3]/li[1]').text

            reg=r'典藏地：(\S+)'

            item['book_site']=re.findall(reg, book_site)[0]

            reg=r'\d{4}-\d{2}-\d{2}'

            due_date=book.find_element(By.XPATH, './ul[4]/li[2]').text

            item['due_date']=re.findall(reg, due_date)[0] 

            due_times=book.find_element(By.XPATH, './ul[5]/li[1]').text

            item['due_times']=re.findall(r'\d{1}', due_times)[0]

            try: 

                state=book.find_element(By.XPATH, './ul[6]/li[1]').text

            except:

                state=''

            finally:

                if '有人預約' in state:

                    item['state']=', 有人預約'

                else:

                    item['state']=''

            borrow_books.append(item)

        browser.back() # 回上一頁

        # 擷取預約紀錄

        div_blueblock=browser.find_element(By.CLASS_NAME, 'blueblock')

        div_blueblock.click()

        books=browser.find_elements(By.CLASS_NAME, 'bookdata')

        reserve_books=[]

        for book in books:

            item=dict()

            book_name=book.find_element(By.XPATH, './h2/a').text    

            item['book_name']=book_name.replace('/', '').strip()

            sequence=book.find_element(By.XPATH, './ul[7]/li[1]').text

            if '預約待取' in sequence:  # 已到館

                item['ready_for_pickup']=True

                reg=r'\d{4}-\d{2}-\d{2}'

                item['expiration']=re.findall(reg, sequence)[0]

                item['sequence']='0'

            else: # 預約中

                item['ready_for_pickup']=False

                item['expiration']=''

                item['sequence']=re.findall(r'\d+', sequence)[0]

            reserve_books.append(item)

        browser.close()

        return (borrow_books, reserve_books)        

    except Exception as e:

        print(e)

        return None, None

    

if __name__ == '__main__':

    start=time.time()

    token='7938146214:AAGoV71AVEygN3e-j-ZfOWUcWSHG1qBp7KE'

    chat_id='7742981072'

    if len(sys.argv) != 3:

        print('用法: python3 ksml_personal_10.py 帳號 密碼')

        sys.exit(1)

    # 取得傳入的帳密參數

    account=sys.argv[1]

    password=sys.argv[2]

    # 呼叫 get_books() 取得借書與預約書        

    borrow_books, reserve_books=get_books(account, password)

    # 處理借書 

    if borrow_books: 

        borrow=[]

        for book in borrow_books:

            book_name=book['book_name']

            book_site=book['book_site'] 

            due_times=book['due_times']

            due_date=book['due_date']

            state=book['state']

            due_date=datetime.strptime(due_date, '%Y-%m-%d') # 到期日   

            today_str=datetime.today().strftime('%Y-%m-%d')   

            today=datetime.strptime(today_str, "%Y-%m-%d")   

            delta=(due_date-today).days  # 計算離到期日還有幾天

            if delta < 0:  # 負數=已逾期

                msg='🅧 {} (逾期 {} 天{}, {})'.format(book_name, abs(delta), state, book_site)

                borrow.append(msg)

            elif delta == 0:  # 0=今天到期

                msg='⓿ {} (今日到期, 續借次數 {}{}, {})'.format(book_name, due_times, state, book_site)

                borrow.append(msg)

            elif delta == 1:  # 1=明天到期 

                msg='❶ {} (明日到期, 續借次數 {}{}, {})'.format(book_name, due_times, state, book_site)

                borrow.append(msg)

            elif delta == 2:  # 2=後天到期 

                msg='❷ {} (後天到期, 續借次數 {}{}, {})'.format(book_name, due_times, state, book_site)

                borrow.append(msg)

            elif 2 < delta < 8:  # 3 天以上一周內到期

                msg=('✦ {} ({} 到期, '.format(book_name, book["due_date"]) +

                     '續借次數 {}{}, {})'.format(due_times, state, book_site))

                borrow.append(msg)

        # 傳送借閱書到期摘要至 Telegram 

        if len(borrow) != 0:

            borrow.insert(0, '\n❖ {} 的借閱 :'.format(account))

            msg='\n'.join(borrow)

            if telegram_send_text(msg):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    # 處理預約書

    if reserve_books:

        reserve=[]

        i=0

        j=['①', '②', '③', '④', '⑤']

        k=['❶', '❷', '❸', '❹', '❺']

        # 預約狀態

        for book in reserve_books:

            book_name=book['book_name']

            sequence=book['sequence']

            ready_for_pickup=book['ready_for_pickup'] # 已到館

            expiration=book['expiration']  # 取書截止日

            if ready_for_pickup:

                msg='{} {} (已到館, 保留期限 {})'.format(k[i], book_name, expiration)

            else:

                msg='{} {} (順位 {})'.format(j[i], book_name, sequence)

            reserve.append(msg)

            i += 1

        # 傳送預約書狀態摘要至 Telegram    

        if len(reserve) != 0:

            reserve.insert(0, '\n❖ {} 的預約 :'.format(account))

            msg='\n'.join(reserve)

            if telegram_send_text(msg):

                print('訊息傳送成功!')

            else:

                print('訊息傳送失敗!')

    else:

        print('無資料')                    

    end=time.time()

    print('執行時間:{}'.format(end-start))

其中黃底色部分即 Chromium driver 舊版寫法.