2025年7月29日 星期二

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

在前一篇測試中已經勾勒出一個簡單的 HTTP API 函式執行平台雛形, 本篇主要是在此基礎上進行部分修改與功能擴充. 

本系列全部的測試紀錄參考 :


首先來修改主程式 serverless.py, 主要修改重點為 :
  • 改用 importlib.util.spec_from_file_location() 搭配 importlib.util.module_from_spec() 根據模組檔案的絕對路徑來動態載入模組, 此方式彈性較大, 可載入放在任何位置的模組檔案 (雖然我規劃固定放在 functions 子目錄下), 而且模組不需要加入 sys.path 中 (避免汙染系統路徑).
  • 新增錯誤日誌檔來紀錄錯誤與例外資訊以利查錯. 
關於用 importlib 動態載入模組方法參考 :
主程式修改如下 : 

# serverless.py
from flask import Flask, request, jsonify
import importlib.util
import os
import logging

app=Flask(__name__)
# 指定函式模組所在的資料夾
FUNCTIONS_DIR=os.path.expanduser('./functions')
# 指定錯誤日誌檔 (在目前工作目錄下)
logging.basicConfig(filename='serverless_error.log', level=logging.ERROR)
# 定義路由裝飾器 (支援 RESTful) 
# 若 URL=/function/hello : subpath 預設為 ''
# 若 URL=/function/hello/Tony :  subpath 為 'Tony'
@app.route('/function/<func_name>', defaults={'subpath': ''}, methods=['GET', 'POST'])
@app.route('/function/<func_name>/<path:subpath>', methods=['GET', 'POST'])
def handle_function(func_name, subpath):  # 傳入 subpath 支援 RESTful
    # 1. 取得檔案路徑
    func_path=os.path.join(FUNCTIONS_DIR, f'{func_name}.py')
    if not os.path.isfile(func_path):  # 模組檔案不存在 -> 回 404
        return jsonify({'error': f'Function "{func_name}" not found'}), 404
    try:
        # 2. 動態載入模組 (絕對路徑)
        spec=importlib.util.spec_from_file_location(func_name, func_path)
        module=importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        # 3. 檢查模組中有無 main() 函式 :
        if not hasattr(module, 'main'):  # 模組中無 main() 函式
            return jsonify({'error': f'Module "{func_name}" has no main()'}), 400
        # 4. 將 subpath 加入 request 中 (支援 RESTful)
        request.view_args['subpath']=subpath  
        # 5. 執行模組中的函式 :        
        result=module.main(request)  # 將 request 傳給被載入模組由其自行處理參數
        # 6. 傳回函式執行結果
        return result 
    except Exception as e:
        logging.exception(f'Error in function {func_name}\n{e}')  # 紀錄錯誤於日誌檔
        return jsonify({'error': 'Function execution failed'}), 500

此處藍色字體為修改或增加的部分, 利用 os.path.expanduser() 指定函式模組所在子目錄的方式更普適, 重點是動態載入模組的方式改為可從任何指定的絕對路徑載入且不用更動 sys.path 表. 與前一篇一樣把 subpath 加入 request 的 view_args 字典以支援 RESTful 網址路徑呼叫. 

測試用的函式模組 hello.py 與前一篇相同, 重抄如下 :

# hello.py
def main(request):
    name=request.args.get('name')
    if not name:
        # 從 RESTful 子路徑中抓變數 (如 /function/hello/Tony)
        subpath=request.view_args.get('subpath', '')
        parts=subpath.strip('/').split('/')
        if parts:
            name=parts[0]
    if not name:
        name='World'
    return f'Hello {name}!'

每次修改 Flask 主程式 serverless.py 後要重啟此程式的系統服務 (Nginx 不用) :

sudo systemctl restart serverless.service  

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl restart serverless.service   
[sudo] tony1966 的密碼: 

用瀏覽器訪問下列網址 : 





可見與前一篇的結果一樣. 

如果要查看錯誤日誌 serverless.err.log 可先用下列指令檢查檔案存不存在 :

ls -lh serverless_error.log   

tony1966@LX2438:~/flask_apps/serverless$ ls -lh serverless_error.log   
ls: 無法存取 'serverless_error.log': 沒有此一檔案或目錄

檔案存在的話再用 tail 或 cat 指令檢視日誌檔內容 :

tail serverless_error.log   (預設顯示最後 10 行)
tail -n 50 serverless_error.log  (檢視最後 50 行)
tail -f serverless_error.log  (debug 即時持續追蹤)
cat serverless_error.log  (檢視全部)
grep 'my_function' serverless_error.log  (篩選與 'my_function' 有關的)

注意, 只有等級為 ERROR 或更嚴重例如 CRITICAL 的錯誤訊息才會寫入 serverless_error.log, 一般 404 或 500 之類的錯誤不會被記錄, 除非修改函式模組 (例如 hello.py) 加入 404 與 500 的路由裝飾器, 呼叫 logging.error() 來記錄, 例如 :

@app.errorhandler(404)
def handle_404(e):
    logging.error(f'404 Not Found: {request.path}')
    return jsonify({'error': 'Not Found'}), 404

@app.errorhandler(500)
def handle_500(e):
    logging.error(f'500 Internal Server Error: {traceback.format_exc()}')
    return jsonify({'error': 'Internal Server Error'}), 500


2025-07-30 補充 :

今天在測試函式列表功能時出現錯誤, 被記錄到錯誤日誌檔 :

tony1966@LX2438:~/flask_apps/serverless$ tail serverless_error.log  
    '''.format(len(py_files))
KeyError: ' font-family'
ERROR:root:Error in function functions_list
' font-family'
Traceback (most recent call last):
  File "/home/tony1966/flask_apps/serverless/serverless.py", line 33, in handle_function
    result=module.main(request)  # 將 request 傳給被載入模組由其自行處理參數
  File "/home/tony1966/flask_apps/serverless/./functions/functions_list.py", line 43, in main
    '''.format(len(py_files))
KeyError: ' font-family'

原來是輸出網頁中的 CSS 大括號導致錯誤 :

            a { 
                text-decoration: none; 
                color: #007bff; 
                font-weight: bold; 
                }

用重複的大括號來跳脫就可以解決此錯誤了 :

            a {{ 
                text-decoration: none; 
                color: #007bff; 
                font-weight: bold; 
                }}

2025年7月27日 星期日

2025 年第 30 周記事

上兩周周末閃到腰痛了一個禮拜, 本周算是完全好了, 但仍不敢完全恢復運動, 僅週二去河堤走路為貓, 近兩周沒去河堤小咪還記得我 :




週三去載菁菁時說她在工作室養了一隻兔子, 傳了照片給我, 是一隻垂耳兔, 我問它大便是不是很臭, 聽說不會哩 (而且也不是想像中要吃胡蘿蔔) :




昨天下午忙完程式正要去給芭樂套袋, 剛好小安帶小朋友來玩, 我只好先陪他們, 到五點多他們回去才去套袋, 本周只套了 60 顆而已. 

小灰的兩隻貓崽已經稍稍長大些, 一家三口住在二樓, 我上去巡視時貓崽看我很陌生, 一直盯著我看, 看我走近就慌忙躲起來 : 




周末這兩天我在忙 Mapleboard 主機上的 serverless 函式呼叫機制的原型測試, 所以上週的待辦清單一項也沒完成, 所以本周照抄且新增一項 (芭樂樹施肥) :
  • 鄉下車庫屋頂透明罩翻新 (需買一個 Silicon 槍)
  • 洗衣機三通管接管
  • 舊豬舍外牆電源開關防雨盒製作
  • 清理冰箱
  • 清理車庫雜物
  • 清理庫房雜物
  • 芭樂樹施肥
不過早上去市場經過小漢有進去買一個洗衣機三通管接管要用的短接頭, 還完成一項上周清單忘記列的項目 : 把蔥苗種在盆子裡, 這算是完成一項啦. 

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

這一個月來忙完 Mapleboard 主機上的 Nginx+Gunicorn+Flask 網站架構配置與安全防護後, 終於可以進行應用程式開發與佈署了. 

本系列全部的測試紀錄參考 :


今天在下面這本書的最後一章看到作者使用 Google Cloud Functions (GCF) 的 serverless 函式服務作為 LINE Bot 的 webhook, 我突然想到何不在 Mapleboard 主機上打造一個簡化版的 GCF 服務呢? 現在 Vibe coding 正夯, 透過與 AI 協作應該可以快速實現.


Google Cloud Functions 是 Google 雲端服務平台上的一個無伺服器 (serverless) 執行環境, 允許開發者撰寫獨立且輕量的函式 (Function), 可連結 Google Cloud 各種事件並在雲端自動回應, 也可透過 HTTP 介面呼叫函式, 使用者無需管理伺服器或基礎設施, Google 會自動負責資源調度與彈性擴展, 並依實際執行時間計費, 是成本效益非常高的一種雲端服務, 非常適合 Webhook, 輕量級 API, 行動裝置後端服務, 物聯網設備資料處理等合事件驅動與有即時運算需求的微服務場景. 

但我沒必要複製一個完整的 GCF 功能, 我只要能達成下列目標即可 :
  • 每個 endpoint 都像一個獨立的 function
  • 只需寫 function 就能對外提供 HTTP API
  • 模組只要放入 ~/functions/ 即可透過 HTTP 呼叫執行
  • 可以透過簡單的部署方式新增/更新函式
以下就從最簡單的雛型結構開始一步一步搭建系統, 漸進優化與增加功能. 


1. 系統架構雛形 : 

在前面的測試中, 我已在 Mapleboard 的 Ubuntu Mate 上安裝好 Nginx+Gunicorn+Flask 執行環境, 而且註冊了 https://tony196cc.cc 網域與 https://flask.tony1966.cc 子網域, 並於使用者根目錄下建立了 flask_apps 子目錄來存放所有 Flask 應用程式, 我打算將這個仿 GCF 函式執行平台放在 serverless 子目錄下, 要執行的函式都放在下一層的 functions 子目錄中. 

此系統的初步結構如下圖所示 : 

/home/tony1966/flask_apps/serverless/
  ├── serverless.py      
  └── functions/
        ├── __init__.py
        └── hello.py

注意, 由於要從 functions 資料夾動態載入模組, 所以它底下必須有一個空白的 __init__.py, 這樣 functions 才會被視為一個套件, 否則載入模組時會出現錯誤. 

# __init__.py
# 此為讓 functions 目錄被視為一個套件的空模組

其次, 每一個要執行的函式都要有一個接收 request 物件的 main() 函式, 其內容就是此模組要執行的邏輯, 例如 :

# hello.py
def main(request):
    name=request.args.get('name', 'World')  # 從 request 物件取得 name 參數 (預設值 'World')
    return f'Hello {name}!'

所有要呼叫的函式模組都放在 ~/flask_apps/serverless/functions/ 下 :




此系統的主程式 serverless.py 則是利用 Python 內建的 importlib 模組依據 HTTP 的 URL 路徑來動態載入 functions 子目錄下要執行的模組, 程式碼如下 : 

# serverless.py
from flask import Flask, request, jsonify
import importlib
import os

app=Flask(__name__)
FUNCTION_DIR='functions'  # 函式模組目錄
# 定義路由裝飾器
@app.route('/function/<func_name>', methods=['GET', 'POST'])
def run_function(func_name):  # 傳入函式模組名稱
    try:
        module_path=f'{FUNCTION_DIR}.{func_name}' # 模組路徑 functions.函數名稱
        mod=importlib.import_module(module_path)  # 動態載入要求之函式模組
        if hasattr(mod, 'main'):  # 檢查模組中是否有 main() 函式
            return mod.main(request)  # 執行 main() 函式並傳回結果
        else:  # 沒有 main() 函式傳回 400 錯誤
            return 'Function must define a `main(request)` method', 400
    except Exception as e:  # 出現異常傳回 500 錯誤
        return f'Error: {e}', 500

此系統目前只有一個路由裝飾器, 用來將 https://flask.tony1966.cc/function/<func_name> 的網址綁定到 run_function() 函式上. 在 run_function() 中則式呼叫 importlib.import_module() 來載入路徑為 functions.<func_name> 的模組 (等同於 import functions.<func_name> 之效果), 然後檢查函式有無 main() 函式, 有就呼叫它並傳回結果, 否則傳回  500 錯誤訊息. 




2. 啟動 Gunicorn 應用伺服器 :

上面已將 HTTP API 函式執行平台的原型架構準備好, 就可以開始來運行它了, 先切換到主程式 serverless.py 所在的 serverless 子目錄下 : 

tony1966@LX2438:~$ cd flask_apps   
tony1966@LX2438:~/flask_apps$ cd serverless  
tony1966@LX2438:~/flask_apps/serverless$ ls -ls  
總用量 8
4 drwx------ 2 tony1966 tony1966 4096 Jul 25 14:08 functions
4 -rw-rw-r-- 1 tony1966 tony1966  851 Jul 24 23:38 serverless.py

用下列指令啟動 Gunicorn 伺服器程序來執行主程式中的 app 物件 :

gunicorn -w 4 -b 127.0.0.1:8000 serverless:app

tony1966@LX2438:~/flask_apps/serverless$ gunicorn -w 4 -b 127.0.0.1:8000 serverless:app   
[2025-07-25 15:21:47 +0800] [686727] [INFO] Starting gunicorn 23.0.0
[2025-07-25 15:21:47 +0800] [686727] [INFO] Listening at: http://127.0.0.1:8000 (686727)
[2025-07-25 15:21:47 +0800] [686727] [INFO] Using worker: sync
[2025-07-25 15:21:47 +0800] [686728] [INFO] Booting worker with pid: 686728
[2025-07-25 15:21:47 +0800] [686729] [INFO] Booting worker with pid: 686729
[2025-07-25 15:21:47 +0800] [686730] [INFO] Booting worker with pid: 686730
[2025-07-25 15:21:47 +0800] [686731] [INFO] Booting worker with pid: 686731

按 Ctrl + C 跳出被占用的終端機, 這會自動停止手動啟動的 Gunicorn 程序. 如果按 Ctrl +C 無反應, 可以另開一個終端機, 用下列指令檢視有哪些 Gunicorn 程序在運行 : 

ps aux | grep gunicorn   

tony1966@LX2438:~/flask_apps/serverless$ ps aux | grep gunicorn   
tony1966  223187  0.0  0.1  34156  5268 ?        Ss   Jul06   9:51 gunicorn: master [hello:app]
tony1966  223192  0.0  0.3  42480 12952 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  223193  0.0  0.3  42476 13364 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  223194  0.0  0.3  42476 15452 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  223195  0.0  0.4  42476 15796 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  686727  0.0  0.5  34088 20576 pts/1    S+   15:21   0:01 gunicorn: master [serverless:app]
tony1966  686728  0.0  0.6  41152 26360 pts/1    S+   15:21   0:00 gunicorn: worker [serverless:app]
tony1966  686729  0.0  0.6  41156 26356 pts/1    S+   15:21   0:00 gunicorn: worker [serverless:app]
tony1966  686730  0.0  0.6  41156 26352 pts/1    S+   15:21   0:00 gunicorn: worker [serverless:app]
tony1966  686731  0.0  0.6  41156 26376 pts/1    S+   15:21   0:00 gunicorn: worker [serverless:app]
tony1966  686904  0.0  0.0   9136  2024 pts/0    S+   15:59   0:00 grep --color=auto gunicorn

此處 PID 後面的 pts/0 或 pts/1 (關聯當前使用者的 shell) 以及 S+ (在前景執行的子行程) 表示這些程序都是手動運行的; 而 ? (沒有關聯終端機) 與 S 或 Ss (daemon 狀態) 表示為 Systemd 系統服務所啟動. 可以指定應用程式名稱手動刪除程序, 例如 :

pkill -f "gunicorn.*serverless:app"  

tony1966@LX2438:~/flask_apps/serverless$ pkill -f "gunicorn.*serverless:app"   
tony1966@LX2438:~/flask_apps/serverless$ ps aux | grep gunicorn    
tony1966  223187  0.0  0.1  34156  5268 ?        Ss   Jul06   9:52 gunicorn: master [hello:app]
tony1966  223192  0.0  0.3  42480 12952 ?        S    Jul06   0:58 gunicorn: worker [hello:app]
tony1966  223193  0.0  0.3  42476 13364 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  223194  0.0  0.3  42476 15452 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  223195  0.0  0.4  42476 15796 ?        S    Jul06   0:57 gunicorn: worker [hello:app]
tony1966  688107  0.0  0.0   9136  2028 pts/0    S+   16:16   0:00 grep --color=auto gunicorn

這樣就全部刪除手動的 Gunicorn 程序了. 


3. 建立 Systemd 服務檔案 : 

接下來要將啟動 Gunicorn 去執行 app 的任務做成一個系統服務 (Systemd), 這樣萬一系統當機重開機時, 就會自動啟動 Gunicorn 程序來執行應用程式, 程式若當掉也會自動重啟. 但首先要用 which gunicorn 指令來確定 Gunicorn 的安裝位置 : 

tony1966@LX2438:~/flask_apps/serverless$ which gunicorn   
/home/tony1966/.local/bin/gunicorn   

然後用 Nano 建立此系統服務的設定檔 serverless.service : 

tony1966@LX2438:~/flask_apps/serverless$ sudo nano /etc/systemd/system/serverless.service
[sudo] tony1966 的密碼: 

輸入下列資訊 : 

[Unit]
Description=Serverless Flask App
After=network.target

[Service]
User=tony1966
Group=tony1966
WorkingDirectory=/home/tony1966/flask_apps/serverless
ExecStart=/home/tony1966/.local/bin/gunicorn -w 4 -b 127.0.0.1:8000 serverless:app
Restart=always

[Install]
WantedBy=multi-user.target

注意, ExecStart 參數必須指定 which gunicorn 所傳回的 Gunicorn 執行程式位置, 否則無法成啟動此服務. 按 Ctrl+O 儲存後按 Ctrl+X 跳出 nano. 

然後輸入下列指令重新讀取 systemd 配置 : 

sudo systemctl daemon-reexec   
sudo systemctl daemon-reload   

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl daemon-reexec  
tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl daemon-reload   

接下來用下列指令讓 serverless 服務在系統重開機時自動啟動 : 

sudo systemctl enable serverless  

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl enable serverless    
Created symlink /etc/systemd/system/multi-user.target.wants/serverless.service → /etc/systemd/system/serverless.service. 

此指令不會馬上啟動服務, 而是會在 /etc/systemd/system/ 下建立符號連結, 連結到上面建立的  serverless.service 檔案. 

最後用下列指令啟動此 serverless 服務 : 

sudo systemctl start serverless   
  
tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl start serverless    

這時可用下列指令檢查服務狀態 :

sudo systemctl status serverless   

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl status serverless    
● serverless.service - Serverless Flask App
     Loaded: loaded (/etc/systemd/system/serverless.service; enabled; vendor preset: enabled)
     Active: active (running) since Fri 2025-07-25 16:59:53 CST; 9s ago
   Main PID: 688560 (gunicorn: maste)
      Tasks: 5 (limit: 4213)
     Memory: 65.4M
        CPU: 3.025s
     CGroup: /system.slice/serverless.service
             ├─688560 "gunicorn: master [serverless:app]" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" ">
             ├─688561 "gunicorn: worker [serverless:app]" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" ">
             ├─688562 "gunicorn: worker [serverless:app]" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" ">
             ├─688563 "gunicorn: worker [serverless:app]" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" ">
             └─688564 "gunicorn: worker [serverless:app]" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" ">

Jul 25 16:59:53 LX2438 systemd[1]: Started Serverless Flask App.
Jul 25 16:59:54 LX2438 gunicorn[688560]: [2025-07-25 16:59:54 +0800] [688560] [INFO] Starting gunicorn 23.0.0
Jul 25 16:59:54 LX2438 gunicorn[688560]: [2025-07-25 16:59:54 +0800] [688560] [INFO] Listening at: http://127.0.0.1:8000 (688560)
Jul 25 16:59:54 LX2438 gunicorn[688560]: [2025-07-25 16:59:54 +0800] [688560] [INFO] Using worker: sync
Jul 25 16:59:54 LX2438 gunicorn[688561]: [2025-07-25 16:59:54 +0800] [688561] [INFO] Booting worker with pid: 688561
Jul 25 16:59:54 LX2438 gunicorn[688562]: [2025-07-25 16:59:54 +0800] [688562] [INFO] Booting worker with pid: 688562
Jul 25 16:59:54 LX2438 gunicorn[688563]: [2025-07-25 16:59:54 +0800] [688563] [INFO] Booting worker with pid: 688563
Jul 25 16:59:54 LX2438 gunicorn[688564]: [2025-07-25 16:59:54 +0800] [688564] [INFO] Booting worker with pid: 688564

看到 active (running) 表示這個 serverless 服務已經正常運行中.  

以後若有修改服務設定檔 serverless.service, 必須用下列指令重新載入設定檔與重新啟動 : 

sudo systemctl daemon-reexec  # 或 daemon-reload  
sudo systemctl restart serverless  


4. 更新 Nginx 的站台設定檔 flask.tony1966.cc : 

我在之前的測試中已經在 Namecheap 網域平台上建立了一個子網域 https://flask.tony1966.cc 專門用來執行 web app, 參考 : 


上面創建的 HTTP API 函式執行平台 serverless 也是要掛在此子網域下運行, 因此必須在其站台設定檔 flask.tony1966.cc 中添加 location 區塊, 將所有對 https://flask.tony1966.cc/function/ 的請求轉發給 Gunicorn 所監聽的本機端口 http://127.0.0.1:8000/. 用 nano 編輯站台設定檔 :

sudo nano /etc/nginx/sites-available/flask.tony1966.cc 

tony1966@LX2438:~/flask_apps/serverless$ sudo nano /etc/nginx/sites-available/flask.tony1966.cc 
[sudo] tony1966 的密碼: 

在裡面添加如下轉發設定 : 

    location /function/ {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

按 Ctrl+O 儲存後按 Ctrl+X 跳出 nano. 

轉發指令 proxy_pass 會將 /function/ 路徑轉發給本機網址 http://127.0.0.1:8000, 注意, 8000 後面沒有右斜線, 這很重要, 這表示會保留前綴 (此處為 /function/) 並附加至後端 URL, 如果末端有右斜線則會去除前綴, 用法摘要如下表 :


proxy_pass 寫法 轉發方式
http://127.0.0.1:8000/ 去除前綴 (此處為 /function/) 並將剩餘路徑轉給後端 :
例如請求 /function/hello 會轉成 http://127.0.0.1:8000/hello
http://127.0.0.1:8000 保留前綴 (此處為 /function/) 並附加至後端 URL : 
例如請求 /function/hello 會轉成 http://127.0.0.1:8000/function/hello


此處採用保留前綴的第二個方案, 好處是 :
  • 可提供類似 REST API 的函式呼叫端點
  • 方便未來加上 Blueprint 或 API 文件 (如 Swagger/OpenAPI) 
  •  URL 結構直接映射到功能模組 (使用者看到 /function/hello, Flask 也正好就是處理 /function/hello)
然後用 sudo nginx -t 指令檢查全部設定檔 : 

tony1966@LX2438:~/flask_apps/serverless$ sudo nginx -t   
[sudo] tony1966 的密碼: 
nginx: the configuration file /etc/nginx/nginx.conf syntax is ok
nginx: configuration file /etc/nginx/nginx.conf test is successful

都正常的話重啟 Nginx : 

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl restart nginx   


5. 測試 serverless 函式 : 

訪問下列網址 : 





可見都能正常顯示網頁內容. 


6. 支援 RESTful 請求 : 

上面的 Flask app 配置方式只支援用 URL 傳遞或表單傳遞參數, 不支援 RESTful 呼叫, 例如訪問下列網址將出現 404 回應 : 





如果要支援 RESTful 的呼叫網址, 必須同時修改 Flask 程式 serverless.py 與函式模組 hello.py 才行, 主程式 serverless.py 修改如下 :

# serverless.py
from flask import Flask, request, jsonify
import importlib
import os

app=Flask(__name__)
FUNCTION_DIR='functions'  # 函式模組目錄
# 定義路由裝飾器 (支援 RESTful) 
# 若 URL=/function/hello : subpath 預設為 ''
# 若 URL=/function/hello/Tony :  subpath 為 'Tony'
@app.route('/function/<func_name>', defaults={'subpath': ''}, methods=['GET', 'POST'])
@app.route('/function/<func_name>/<path:subpath>', methods=['GET', 'POST'])
def run_function(func_name, subpath):  # 傳入函式模組名稱與 subpath 參數
    try:
        module_path=f'{FUNCTION_DIR}.{func_name}' # 模組路徑 functions.函數名稱
        mod=importlib.import_module(module_path)  # 動態載入要求之函式模組
        if hasattr(mod, 'main'):  # 檢查模組中是否有 main() 函式
            # 將 subpath 附加到 request 物件中 (用 request.view_args)
            request.view_args['subpath']=subpath     # 將 subpath 加入 request 中
            return mod.main(request)  # 執行 main() 函式並傳回結果
        else:  # 沒有 main() 函式傳回 400 錯誤
            return 'Function must define a `main(request)` method', 400
    except Exception as e:  # 出現異常傳回 500 錯誤
        return f'Error: {e}', 500

注意此處在 request.view_args 字典中手動新增了 'subpath' 鍵, 用來記錄 URL 路徑中模組名稱以下的子路徑字串, 例如 https://flask.tony1966.cc/function/hello/Tony, 則 subpath='Tony', 它會被記錄在 request 中傳給模組的 main() 函式處理, main() 可以從 request.view_args 字典中取出 subpath 鍵之值來得到 RESTful 呼叫中傳遞的參數 (例如 'Tony'). 

總之, request.view_args 字典用來記錄 URL /user/<username> 格式中的動態路徑參數 (例如 username), 但也可以手動自訂新的鍵, 它與 request.args 不同, request.view_args 是可變的字典, 而 request.args 則是用來記錄 URL 網旨中以 ?key=value 格式攜帶的參數, 它是不可變的字典. 

函式模組 hello.py 修改如下 : 

def main(request):
    # 先從請求字串擷取 name 參數 (?name=)
    name=request.args.get('name')
    if not name:   # 請求字串沒有 name 參數
        # 嘗試從 RESTful 子路徑 (例如 /function/hello/Tony) 中擷取 name 參數 
        subpath=request.view_args.get('subpath', '')   # 取得 Flask 傳入的 subpath 參數
        parts=subpath.strip('/').split('/')   # # 拆解成串列例如 ['Tony']
        if parts:
            name=parts[0]   # 取第一段作為 name (例如 'Tony')
    if not name:
        name='World'
    return f'Hello {name}!'

然後系統服務 serverless.service (Nginx 不需要重啟) :

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl restart serverless   
[sudo] tony1966 的密碼: 

這樣此系統就支援 RESTful API 呼叫了, 訪問下列網址 : 




可見無論是 query string 或 RESTful 方式呼叫均能正常顯示網頁.  

2025年7月24日 星期四

Python 學習筆記 : 網頁爬蟲實戰 (十八) 擷取隨機的格言 (中)

前一篇測試中利用格言爬蟲從網路上擷取隨機格言 (英文), 然後透過 Gemini API 將其翻譯為繁體中文組成中英雙語格言字串, 最後再利用 Google STMP 郵件伺服器將每日格言寄到指定的 Email 信箱. 本篇則是要將這個 dailly_quote 程式改寫為傳送 Telegram 訊息. 

本系列全部測試文章索引參考 :


關於使用 Telegram Bot API 傳訊息方法參考 :


每日格言程式修改如下 :

# dailly_quote.py
import requests
import google.generativeai as genai

def get_quote():
    url='https://zenquotes.io/api/random'
    try:
        res=requests.get(url, timeout=10)
        data=res.json()
        return data[0]['q'], data[0]['a']  # 成功傳回 (quote, author)
    except Exception as e:
        return None, f'Failed to fetch quote: {e}'  # 失敗傳回 (None, '失敗訊息')
    
def translate_quote(quote, api_key):
    genai.configure(api_key=api_key)
    model=genai.GenerativeModel('gemini-2.0-flash')
    prompt=f'請將以下英文翻譯為繁體中文, 只要翻譯一次即可:{quote}'
    try:
        reply=model.generate_content(prompt)
        return reply.text.strip().strip('\"\n ')  # 清除結尾開頭之雙引號跳行與空格
    except Exception as e:
        return f'格言翻譯失敗:{e}'
 
def bilingual_quote(quote, quote_zh_tw, author):
    return f'早安! 今日格言:\n「{quote_zh_tw}」\n{quote}\n—— {author}'

def telegram_msg(token, chat_id, text):
    url=f'https://api.telegram.org/bot{token}/sendMessage'
    data={'chat_id': chat_id, 'text': text}
    try:
        with requests.post(url, data=data, timeout=10) as r:
            r.raise_for_status()  # 若 status_code != 2xx 會拋出例外
            result=r.json()
            if result.get('ok'):
                print('Message sent successfully.')
                return True
            else:
                print(f'Telegram API Error: {result}')
                return False
    except requests.exceptions.RequestException as e:
        print(f'Request error: {e}')
        return False

# 取得每日格言與其中文翻譯
quote, author=get_quote()
gemini_api_key='我的 Gemini API Key'
quote_zh_tw=translate_quote(quote, gemini_api_key)
msg=bilingual_quote(quote, quote_zh_tw, author)
# 用 Telegram Bot API 傳訊息
token='我的 Telegram 金鑰'
chat_id='我的 Telegram chat id'
telegram_msg(token, chat_id, msg) 

此處因為爬蟲本身就要用到 requests 模組, 所以 Telegram Bot API 就使用 RESTful API 就好 (當然也可以用 Bot 物件來做), 結果如下 :




嗯, 效果還不錯. 

2025年7月23日 星期三

Python 學習筆記 : 用 importlib 動態載入模組

最近為了在 Mapleboard 上架設一個仿 Google Cloud Function (GCF) 的輕量級 Serverless 執行平台, 我詢問 AI 解決方案時發現可以用 Python 內建模組 importlib 來實現, 我之前都沒用過, 所以就先來測試一下此模組用法. 

importlib 為 Python 標準函式庫中的內建模組, 用來動態載入 (dynamic import) 其他模組, 或是在程式執行時重新載入已經載入過的模組, 它提供了比一般 import 更靈活的功能, 特別適合在執行時根據字串名稱動態導入模組或呼叫其函式 (需要導入的模組名稱事先未知), 常見於 Plugin 系統, Serverless 執行平台, 與模組熱更新等應用場景. 

首先匯入 importlib 模組 : 

>>> import importlib   

用 dir() 檢視 importlib 模組內容 : 

>>> dir(importlib)   
['_RELOADING', '__all__', '__builtins__', '__cached__', '__doc__', '__file__', '__import__', '__loader__', '__name__', '__package__', '__path__', '__spec__', '_abc', '_bootstrap', '_bootstrap_external', '_imp', '_pack_uint32', '_unpack_uint32', 'find_loader', 'import_module', 'invalidate_caches', 'machinery', 'reload', 'sys', 'util', 'warnings']

可以用自訂模組 members 的 list_members() 顯示類別, 函式, 與子模組等成員, 參考 : 


>>> from members import list_members   
>>> list_members(importlib)      
find_loader <class 'function'>
import_module <class 'function'>   
invalidate_caches <class 'function'>
machinery <class 'module'>
reload <class 'function'>
sys <class 'module'>
util <class 'module'>
warnings <class 'module'>

可見 importlib 有四個函式與子模組 (find_loader 已被棄用), 常用函式如下表 : 


 importlib 常用函式 功能說明
 importlib.import_module(name) 動態匯入模組, 效果與 import 相同, 可指定字串路徑.
 importlib.reload(module) 重新載入已載入的模組, 常用於模組更新後的即時刷新.
 importlib.util.find_spec(name) 傳回指定模組的 ModuleSpec, 可查詢路徑與 loader 等資訊.
 importlib.util.module_from_spec(spec) 根據 spec 產生尚未初始化的模組物件.
 spec.loader.exec_module(module) 將指定模組物件執行, 真正「啟動」模組內容.
 importlib.invalidate_caches() 清除 importlib 的快取, 強迫重新尋找模組.



1. 載入目前目錄下模組或系統模組 : 

importlib.import_module() 函式用來依據傳入之模組名稱 (字串) 動態載入其他模組, 這些模組的位置必須在目前的工作目錄下或系統模組 (可以在 sys.path 找到的模組), 意即它們都在標準查找路徑內. 例如載入標準函式庫中的模組, 只要傳入內建函式名稱即可 :  

>>> math_module=importlib.import_module('math')    
>>> math_module.sqrt(2)   
1.4142135623730951

此與直接匯入 math 做計算效果是一樣的 :

>>> import math   
>>> math.sqrt(2)   
1.4142135623730951

但差別在於傳入 importlib.import_module() 的是字串變數, 因此可以動態地載入任何模組, 而不須於程式開頭靜態載入 (事實上如果模組名稱未知, 而是在執行時才確定, 那也無法使用 import 靜態載入). 

也可以呼叫內建函式 getattr() 從 importlib.import_module() 載入的模組中取出指定之物件或函式, 例如數學函式庫 math_module :

>>> sqrt_func=getattr(math_module, 'sqrt')   
>>> sqrt_func(2)     
1.4142135623730951   

另外也可以載入目前工作目錄下的模組, 只要傳入模組名稱即可, 如果位於子目錄下則加上以點號隔開的路徑即可, 例如 '子目錄.孫目錄.模組名稱'. 

例如在目前工作目錄下建立了一個子目錄 functions :

D:\python\test>mkdir functions  

然後在底下分別編輯三個模組 add.py, sub.py, 與 mul.py :

# add.py
def calc(x, y):
    return x + y

# sub.py
def calc(x, y):
    return x - y

# mul.py
def calc(x, y):
    return x * y

另外為了讓 functions 子目錄符合 Python 套件要求, 在 functions 子目錄下須新增一個空檔案 __init__.py, 檔案目錄結構如下圖 :




然後回上一層 (工作目錄), 新增一個 Python 檔案, 例如 dynamic_loading_1.py :

# dynamic_loading_1.py
import importlib

modules=['add', 'sub', 'mul']  # 模組清單
x, y=10, 3  # 傳入參數
# 執行每個模組的 calc() 函式
for module_name in modules:  
    module_path=f'functions.{module_name}'    # 例如 'functions.add'
    module=importlib.import_module(module_path)
    result=module.calc(x, y)
    print(f'{module_name}.calc({x}, {y})={result}')

此程式中的 modules 串列儲存了 functions 子目錄下的三個模組名稱, 然後用迴圈走訪此串列, 於迴圈中用 importlib.import_module() 依序載入此三模組實例, 呼叫其 calc() 函式進行運算後輸出, 結果如下 :

>>> %Run dynamic_loading_1.py   
add.calc(10, 3)=13
sub.calc(10, 3)=7
mul.calc(10, 3)=30
 
此簡單範例說明了動態載入模組的原理, 在上面的應用程式 dynamic_loading_1.py 中的模組名稱為已知之串列內容, 但是在實際應用上 (例如 serverless 平台) functions 底下的模組名稱可能是動態變化的, 且真正要執行的模組是由客戶端傳來的變數, 無法用 import 靜態載入, 這時就必須用到此例所展示的動態載入技巧了. 


2. 載入其他位置下的模組 : 

如果要載入的模組既非系統模組 (即不在 sys.path 內) 也不在目前工作目錄或其子目錄下, 而是在其他任意位置的話 (僅知其檔案路徑), 就無法使用 importlib.import_module() 載入, 必須使用 importlib.util.spec_from_file_location() 搭配 importlib.util.module_from_spec() 才能載入. 

在上面的範例中, 要載入的模組位於 D:\python\test\functions 資料夾下面 : 

D:\python>tree test\functions /f   
列出磁碟區 新增磁碟區 的資料夾 PATH
磁碟區序號為 1258-16B8
D:\PYTHON\TEST\FUNCTIONS
    add.py
    mul.py
    sub.py
    __init__.py

我在 D:\python\ 下建立一個與 test 平行的 test2 資料夾, 並且新建一個 dynamic_loading_2.py 應用程式來測試從任意位置載入模組的方法 :

D:\python>tree test2 /f  
列出磁碟區 新增磁碟區 的資料夾 PATH
磁碟區序號為 1258-16B8
D:\PYTHON\TEST2
    dynamic_loading_2.py

dynamic_loading_2.py 程式碼如下 :

# dynamic_loading_2.py
import importlib
import os

modules=['add', 'sub', 'mul']  # 模組清單
x, y=10, 3  # 傳入參數
# 定義 functions 資料夾的實際路徑(根據你的說明)
functions_dir=r'D:\python\test\functions'
# 動態載入並執行每個模組的 calc() 函式
for module_name in modules:
    # 串接路徑與檔名
    file_path=os.path.join(functions_dir, f'{module_name}.py')  
    # 從指定檔案路徑建立模組載入規格 (傳回 ModuleSpec 物件 )
    spec=importlib.util.spec_from_file_location(module_name, file_path)
    # 根據模組規格 spec 建立一個新的模組物件 (傳回 ModuleType 物件)
    module=importlib.util.module_from_spec(spec)
    # 執行模組程式碼 (將模組內容載入到記憶體中的 module 物件)
    spec.loader.exec_module(module)
    # 執行 module 物件中的 calc() 函式
    result=module.calc(x, y)  
    print(f'{module_name}.calc({x}, {y})={result}')

此處先定義函式模組的資料夾變數 functions_dir, 然後在遍歷 modules 的迴圈中, 用 importlib.util.spec_from_file_location() 建立載入該模組的規格 (ModuleSpec) 物件, 然後呼叫 importlib.util.module_from_spec() 並傳入規格物件來建立一個模組物件, 最後呼叫 spec.loader.exec_module() 並傳入模組物件將模組載入記憶體, 這樣便能呼叫模組物件中的函式了, 結果如下 : 

>>> %cd 'D:\python'    
>>> %cd 'D:\python\test'  
>>> %cd 'D:\python'  
>>> %cd 'D:\python\test2'  
>>> %Run dynamic_loading_2.py  
add.calc(10, 3)=13
sub.calc(10, 3)=7
mul.calc(10, 3)=30

可見與上面範例一樣都能動態載入模組執行裡面的函式. 但此方法比上面的老, 因為可載入放在任何位置的 .py 模組檔案, 而且該模組不需要在 sys.path 中 (不用汙染 sys.path), 也不需位於合規套件內 (即含有 __init__.py). 

PS :

加入 sys.path 的方法 : 

import sys   
sys.path.append('/home/你的使用者')     

2025年7月21日 星期一

Python 學習筆記 : 用 Google SMTP 伺服器傳送 Email

這兩天在 Mapleboard 上安裝 postfix 與 mailutils 工具包成功地從命令列透過 Google SMTP 郵件伺服器傳送 Email, 其實也可以用 Python 的 smtplib 套件來做, 我之前在鄉下那台樹莓派 Pi 3 主機也是用這方法將光世代網路的外網 IP 寄到我的 Hinet 信箱以便能遠端連線, 參考 :


今天我將其中 Python 程式 reportip3.py 的寄信功能寫成一個 send_mail() 函式方便應用程式呼叫.


1. 傳送 ASCII 信件 : 

如果信件內容是英數字等 ASCII 編碼的字元, 可用下列函式 send_mail() 來傳送 : 

# smtp_mail_test_1.py
import smtplib

def send_mail(account, password, subject, from_addr, to_addr, cc_addr=None, body=''):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    '''content=(
        f"Subject:{subject}\n"
        f"From:{from_addr}\n"
        f"To:{', '.join(to_addr)}\n"
        f"Cc:{', '.join(cc_addr)}\n"
        f"{body}"
        )'''
    content=(
        "Subject:{subject}\n"
        "From:{from_addr}\n"
        "To:{to_addr}\n"
        "Cc:{cc_addr}\n"
        "{body}"
        ).format(
            subject=subject,
            from_addr=from_addr,
            to_addr=', '.join(to_addr),
            cc_addr=', '.join(cc_addr),
            body=body
            )    
    all_recipients=to_addr + cc_addr  # 合併收件人和副本收件人
    status=smtp.sendmail(from_addr, all_recipients, content)
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject=''                   # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']      # 副本收件人(無設為 None)
body='您好, 這是測試信'                # 信件內容
send_mail(account, password, subject, from_addr, to_addr, cc_addr, body)

此處因為我的 Pi 3 安裝的 Python 是 3.5 版不支援 f 字串, 因此改用 format() 來塞變數到字串中. 如果是 Python 3.6+ 版就可以使用 f 字串了. 此程式是從我的 Gmail 信箱傳送 Email 到我的 Hinet 與公司信箱 (副本), 結果兩個信箱都有成功地收到郵件 : 




注意, 程式中的 password 並不是 Google 密碼, 而是啟用 Google 二階段驗證之後, 到下列網址產生的 Google 應用程式密碼 : 


在 "應用程式名稱" 框內輸入可辨識的名稱, 按右下角的 "建立" 鈕即可 : 





將產生的密碼複製下來, 去除中間的分隔用的空白字元後即可使用, 但最好先儲存到文字檔中備查, 因為它只顯示一次, 若沒記下來, 下次要用時須重新建立. 


2. 傳送非 ASCII 信件 : 

不過上面的函式只能用來傳送 ASCII 字元, 如果主旨與內容含有非 ASCII 字元例如中文, 執行時會出現 "UnicodeEncodeError: 'ascii' codec can't encode characters" 的錯誤訊息. 

解決辦法是用 Python 內建的 email 模組中的 email.mime.text.MIMEText 類別來處理 Unicode, 此外, 為了能寄出帶有 PDF與圖檔等附件的信件, 還需要 email.mime.multipart.MIMEMultipart 來打包郵件內容, 完整程式碼如下 :

# smtp_mail_test_2.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_mail(account, password, subject, from_addr, to_addr, cc_addr=None, body=''):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    # 登入 Google SMTP 伺服器
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    # 建立郵件內容
    content=MIMEMultipart()
    content['Subject']=subject
    content['From']=from_addr
    content['To']=', '.join(to_addr)
    content['Cc']=', '.join(cc_addr)
    # 添加郵件正文
    content.attach(MIMEText(body, 'plain', 'utf-8'))
    # 合併收件人和副本收件人
    all_recipients=to_addr + cc_addr
    status=smtp.sendmail(from_addr, all_recipients, content.as_string())
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject='郵件測試'                   # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']      # 副本收件人(無設為 None)
body='您好, 這是測試信'                # 信件內容
send_mail(account, password, subject, from_addr, to_addr, cc_addr, body)

結果如下 :




如果只是傳送純文字的中文 Email, 那麼這個程式就堪用了. 但是如果傳送 HTML 格式的內容, 例如將 body 改為如下粗體 :

body='<b>您好, 這是測試信</b>' 

則它會以 HTML 碼純文字傳送, 結果如下 :




3. 傳送網頁格式信件 :    

如果郵件內容是 HTML 網頁碼, 則在呼叫建構式 MIMEText() 時要傳入 mode 參數為 'html' :

MIMEText(body, 'html', 'utf-8')

修改 send_mail() 函式添加一個 mode 參數來控制內容模式為 'plain' 或 'html' :

# smtp_mail_test_3.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_mail(account, password, subject, from_addr, to_addr, cc_addr=None, body='', mode='html'):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    # 登入 Google SMTP 伺服器
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    # 建立郵件內容
    content=MIMEMultipart()
    content['Subject']=subject
    content['From']=from_addr
    content['To']=', '.join(to_addr)
    content['Cc']=', '.join(cc_addr)
    # 添加郵件正文
    if mode == 'html':
        content.attach(MIMEText(body, 'html', 'utf-8'))
    else:
        content.attach(MIMEText(body, 'plain', 'utf-8'))
    # 合併收件人和副本收件人
    all_recipients=to_addr + cc_addr
    status=smtp.sendmail(from_addr, all_recipients, content.as_string())
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

# 使用範例
account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject='郵件測試'                         # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']          # 副本收件人(無設為 None)
body='<b><i>您好, 這是測試信<i><b>'   # 信件內容
mode='html'                        # 內容模式 'plain'/'html'
send_mail(account, password, subject, from_addr, to_addr, cc_addr, body, mode)

此處信件內容為粗體斜體的中文字, 結果如下 :




如果將 mode 改成純文字 mode='html' 就沒有網頁效果了 : 




4. 傳送 Text+HTML雙模格式信件 :   

上面的 send_mail() 函式呼叫時要用 mode 參數決定內容格式為純文字 (plain) 或網頁 (html) 格式, 純文字格式可以在所有郵件客戶端顯示; 網頁格式較美觀, 但有些郵件軟體不支援, 這時可能會直接顯示原始 HTML 碼, 空白或純文字亂碼等. 

解決辦法是採用 multipart/alternative 內容格式, 也就是同時傳送純文字內容 text_body 與網頁格式內容 html_body, 當收信端收到這種 multipart/alternative 格式信件時, 若支援 HTML 就會顯示網頁格式之內容 html_body; 反之則顯示純文字格式內容 text_body. 

修改 send_mail() 函式加入 text_body 與 html_body 參數並取消 mode 參數 :

# smtp_mail_test_4.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_mail(account, password, subject, from_addr, to_addr, cc_addr=None, text_body='', html_body=''):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    # 登入 Google SMTP 伺服器
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    # 建立郵件內容
    content=MIMEMultipart()
    content['Subject']=subject
    content['From']=from_addr
    content['To']=', '.join(to_addr)
    content['Cc']=', '.join(cc_addr)
    # 添加郵件正文 (雙模)
    content.attach(MIMEText(text_body, 'plain', 'utf-8'))  # 純文字
    if html_body:
        content.attach(MIMEText(html_body, 'html', 'utf-8'))  # 網頁格式
    # 合併收件人和副本收件人
    all_recipients=to_addr + cc_addr
    status=smtp.sendmail(from_addr, all_recipients, content.as_string())
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

# 使用範例
account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject='郵件測試'                         # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']          # 副本收件人(無設為 None)
text_body='您好, 這是測試信'         # 純文字信件內容
html_body='<b><i>您好, 這是測試信<i><b>'   # 網頁格式信件內容
send_mail(account, password, subject, from_addr, to_addr, cc_addr, text_body, html_body)

結果如下 :




5. 傳送有附檔之信件 :    

傳送附檔需要 open() 函式以二進位模式讀取附檔, 用 email.mime.application.MIMEApplication 類別傳給 MIMEApplication() 後加入信件本體中, 修改 send_mail() 函式添加 attachments 參數如下 : 

# smtp_mail_test_5.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication   
import os   

def send_mail(account, password, subject, from_addr, to_addr, cc_addr=None, text_body='', html_body=None, attachments=None):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    if attachments is None:  # 可變資料不宜做參數
        attachments=[]       # 應付迴圈需求
    # 登入 Google SMTP 伺服器
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    # 建立郵件內容
    content=MIMEMultipart()
    content['Subject']=subject
    content['From']=from_addr
    content['To']=', '.join(to_addr)
    content['Cc']=', '.join(cc_addr)
    # 添加郵件正文 (雙模)
    if not text_body and not html_body:  # text_body 與 html_body 均未傳
        text_body='(No message content)'   # 預設內容
    content.attach(MIMEText(text_body, 'plain', 'utf-8'))  # 純文字
    if html_body:
        content.attach(MIMEText(html_body, 'html', 'utf-8'))  # 網頁格式
    # 加入附件
    for filepath in attachments:
        if os.path.isfile(filepath):
            filename=os.path.basename(filepath)
            with open(filepath, 'rb') as f:
                part=MIMEApplication(f.read(), Name=filename)  
                part.add_header('Content-Disposition', 'attachment', filename=filename)
                content.attach(part)
        else:
            print(f'找不到附件:{filepath}')                
    # 合併收件人和副本收件人
    all_recipients=to_addr + cc_addr
    status=smtp.sendmail(from_addr, all_recipients, content.as_string())
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

# 使用範例
account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject='郵件測試'                         # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']          # 副本收件人(無設為 None)
text_body='您好, 這是測試信'         # 純文字信件內容
html_body='<b><i>您好, 這是測試信<i><b>'   # 網頁格式信件內容
attachments=['cat1.jpg', 'cat2.jpg']  # 附檔
send_mail(account, password, subject, from_addr, to_addr, cc_addr, text_body, html_body, attachments)

藍色字的是為了傳送附檔所增加的部分, 結果如下 :




6. 處理主旨的 Unicode 與 Text+HTML 雙模問題 :    

在上面的測試中主旨含有中文 Unicode 都能順利傳送且在收信端正常顯示中文, 其實這些信件並不符合 RFC 標準格式 (因為 RFC 標頭只允許 ASCII 字元), 但因為目前多數的郵件客戶端會試圖去猜測編碼方式, 通常會自動用 UTF-8 來解釋非 ASCII 字元, 所以大都能正常顯示信件內容. 其實比較保險, 符合 RFC 規範的做法是用 email.header.Header 類別來處理主旨的 Unicode 問題, 只要將主旨內容傳給 Header() 建構式即可. 

其次, 我將上面的函式提交給 AI 檢查, 它建議雙模應該採取兩層結構, 把 text/plain 和 text/html 包在 MIMEMultipart('alternative') 中, 再讓它作為內容的一部分附加到 MIMEMultipart('mixed') 外層, 讓彼此在 multipart/alternative 中互為備案 (fallback 機制), 這樣才算是符合 RFC 標準, 否則部分收信端可能只會讀第一個文字段落而忽略 HTML 部分. 

# smtp_mail_test_6.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.header import Header  
import os

def send_mail(account, password, subject, from_addr, to_addr, cc_addr=None, text_body='', html_body=None, attachments=None):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    if attachments is None:  # 可變資料不宜做參數
        attachments=[]       # 應付迴圈需求
    # 登入 Google SMTP 伺服器
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    # 建立郵件內容
    content=MIMEMultipart('mixed')  # 外層 mixed
    content['Subject']=Header(subject, 'utf-8')   # 處理主旨 Unicode
    content['From']=from_addr
    content['To']=', '.join(to_addr)
    content['Cc']=', '.join(cc_addr)
    # 添加郵件正文 (雙模)    
    alt_part=MIMEMultipart('alternative')  # 內層:alternative(純文字 + HTML)
    alt_part.attach(MIMEText(text_body or '(No message content)', 'plain', 'utf-8'))
    if html_body:
        alt_part.attach(MIMEText(html_body, 'html', 'utf-8'))
    content.attach(alt_part)  # 將內層加入外層
    # 加入附件
    for filepath in attachments:
        if os.path.isfile(filepath):
            filename=os.path.basename(filepath)
            with open(filepath, 'rb') as f:
                part=MIMEApplication(f.read(), Name=filename)
                part.add_header('Content-Disposition', 'attachment', filename=filename)
                content.attach(part)
        else:
            print(f'找不到附件:{filepath}')                
    # 合併收件人和副本收件人
    all_recipients=to_addr + cc_addr
    status=smtp.sendmail(from_addr, all_recipients, content.as_string())
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

# 使用範例
account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject='郵件測試'                         # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']          # 副本收件人(無設為 None)
text_body='您好, 這是測試信👋'       # 純文字信件內容
html_body='<b><i>您好, 這是測試信<i><b>🌈'   # 網頁格式信件內容
attachments=['cat1.jpg']  # 附檔
send_mail(account, password, subject, from_addr, to_addr, cc_addr, text_body, html_body, attachments)

黃底色為新增或修改的部分, 因為已經處理了 Unicode, 所以不論中文或 emoji 圖示均可傳送, 結果如下 :




7. 在 HTML 內容中內嵌圖片 :    

上面範例是透過附件檔案方式傳送圖片, 使用夾帶附件的方式可用來傳送任何檔案, 但如果想看到附件內容須點擊開啟附件檔. 如果想在信件內容中直接顯示圖片, 則必須以 HTML 格式用 img 標籤內嵌圖片來傳送信件, 且須使用 email.mime.image,MIMEImage 類別來包裝 open() 讀取的圖片 byte 資料後加入信件標頭中, 將 send_mail() 函式改寫如下 :

# smtp_mail_test_7.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.header import Header
from email.mime.image import MIMEImage
import os

def send_mail(account,
              password,
              subject,
              from_addr,
              to_addr,
              cc_addr=None,
              text_body='',
              html_body=None,
              attachments=None,
              inline_images=None):
    if cc_addr is None:
        cc_addr=[]  # 預設空串列
    if attachments is None:  # 可變資料不宜做參數
        attachments=[]       # 應付迴圈需求
    if inline_images is None:  # 無內嵌圖片
        inline_images={}        
    # 登入 Google SMTP 伺服器
    smtp=smtplib.SMTP('smtp.gmail.com', 587)
    smtp.ehlo()
    smtp.starttls()  # 啟動 TLS 安全傳輸
    smtp.login(account, password)  # 登入 Google SMTP 伺服器
    # 建立郵件內容
    content=MIMEMultipart('mixed')  # 外層 mixed
    content['Subject']=Header(subject, 'utf-8')  # 處理主旨 Unicode
    content['From']=from_addr
    content['To']=', '.join(to_addr)
    content['Cc']=', '.join(cc_addr)
    # 添加郵件正文 (雙模)    
    alt_part=MIMEMultipart('alternative')  # 內層:alternative(純文字 + HTML)
    alt_part.attach(MIMEText(text_body or '(No message content)', 'plain', 'utf-8'))
    # 若含有 inline 圖片使用 multipart/related 包住 HTML
    if html_body:
        related_part=MIMEMultipart('related')
        related_part.attach(MIMEText(html_body, 'html', 'utf-8'))
        for cid, img_path in inline_images.items():
            if os.path.isfile(img_path):
                with open(img_path, 'rb') as f:
                    img=MIMEImage(f.read())
                    img.add_header('Content-ID', f'<{cid}>')
                    img.add_header('Content-Disposition', 'inline', filename=os.path.basename(img_path))
                    related_part.attach(img)
            else:
                print(f'找不到內嵌圖片:{img_path}')
        alt_part.attach(related_part)  # 將 related 加入 alt
    content.attach(alt_part)  # 將 alt 加入信件內容
    # 加入附件
    for filepath in attachments:
        if os.path.isfile(filepath):
            filename=os.path.basename(filepath)
            with open(filepath, 'rb') as f:
                part=MIMEApplication(f.read(), Name=filename)
                part.add_header('Content-Disposition', 'attachment', filename=filename)
                content.attach(part)
        else:
            print(f'找不到附件:{filepath}')                
    # 合併收件人和副本收件人
    all_recipients=to_addr + cc_addr
    status=smtp.sendmail(from_addr, all_recipients, content.as_string())
    if status == {}:
        print('郵件傳送成功!')
    else:
        print('郵件傳送失敗!')
    smtp.quit()

# 使用範例
account='mygmail@gmail.com'     # Gmail 帳號
password='azfkqbjbftodjucd'         # Google 應用程式密碼 (這是樣本)
subject='郵件測試'                         # 主旨
from_addr='mygmail@gmail.com'   # 寄件人
to_addr=['myhinet@ms5.hinet.net']  # 收件人
cc_addr=['tony@xxx.com.tw']          # 副本收件人(無設為 None)
text_body='您好, 這是測試信👋'       # 純文字信件內容
html_body='''
<html>
  <body>
    <h3>您好!</h3>
    <p>這是內嵌圖片的測試:</p>
    <img src="cid:cat1" width="300">
  </body>
</html>
'''
attachments=None  # 無附檔
inline_images={"cat1": "cat1.jpg"}  # 內嵌圖片 (鍵與 img 之 src 對應)
send_mail(
    account,
    password,
    subject,
    from_addr,
    to_addr, 
    cc_addr,
    text_body,
    html_body,
    attachments,
    inline_images
    )

此處藍色與黃色為改寫或增加的部分, 結果如下 :




可見圖片是直接內嵌在信件內容中.

最後這一版 send_mail() 就是集大成的版本了 (Unicode+純文字網頁雙模+可傳附件檔+可內嵌圖片), 我們可以同時夾帶附件檔與內嵌圖片. 但如果信件內容只是英數字而已的話, 其實用最上面那個範例中的簡易版 send_mail() 即可.