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