今天進入微服務的系統整合（System Integration）與端到端（E2E）測試階段。目的是確保 Docker 容器內的 API 在面對真實世界的複雜與異常輸入時能保持穩定。完成項目包含：
app.py） — 設定副檔名白名單（.jpg, .jpeg, .png），針對非圖片檔案直接攔截並回傳 HTTP 400。test_e2e.py） — 模擬用戶上傳藥單照片、觸發 VLM 辨識、生成 TTS 語音、寫入 SQLite 與觸發 LINE 推播的全流程測試。為了確保 API 的健壯度（Robustness），針對上傳非圖片檔案的情境，在 app.py 中加入了副檔名白名單檢查機制。若使用者上傳非 .jpg, .jpeg, .png 檔案，系統將直接攔截並回應 HTTP 400，避免後續不必要的資源浪費與報錯。
以下為加入格式驗證後的完整 app.py 原始碼：
import os
import json
import uuid
import sqlite3
from flask import Flask, request, jsonify, send_from_directory
from dotenv import load_dotenv
from google import genai
from google.genai import types
from PIL import Image
from gtts import gTTS
# LINE Bot SDK 引入
from linebot.v3.messaging import (
Configuration,
ApiClient,
MessagingApi,
PushMessageRequest,
TextMessage
)
# 1. 載入環境變數與初始化
load_dotenv()
api_key = os.getenv("GEMINI_API_KEY")
line_access_token = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
line_user_id = os.getenv("LINE_USER_ID")
if not api_key:
raise ValueError("❌ 錯誤：找不到 GEMINI_API_KEY，請檢查 .env 設定！")
client = genai.Client(api_key=api_key)
if line_access_token:
configuration = Configuration(access_token=line_access_token)
line_api_client = ApiClient(configuration)
line_bot_api = MessagingApi(line_api_client)
else:
line_bot_api = None
app = Flask(__name__)
app.json.ensure_ascii = False
AUDIO_DIR = os.path.join(os.getcwd(), 'static', 'audio')
DATABASE_PATH = os.path.join(os.getcwd(), 'prescription_vlm.db')
os.makedirs(AUDIO_DIR, exist_ok=True)
# 設定允許上傳的圖片格式與驗證函式
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg'}
def allowed_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
# 2. 資料庫初始化
def init_db():
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS prescriptions (
id TEXT PRIMARY KEY,
spoken_summary TEXT NOT NULL,
audio_url TEXT NOT NULL,
safety_warnings TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS medicines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prescription_id TEXT NOT NULL,
name TEXT NOT NULL,
type TEXT NOT NULL,
frequency TEXT NOT NULL,
dosage TEXT NOT NULL,
timing TEXT NOT NULL,
warning TEXT,
FOREIGN KEY (prescription_id) REFERENCES prescriptions (id)
)
''')
conn.commit()
conn.close()
init_db()
# 3. LINE 推播函式
def send_line_notification(summary, medicines_count, safety_warnings):
if not line_bot_api or not line_user_id:
print("⚠️ 尚未設定 LINE_CHANNEL_ACCESS_TOKEN 或 LINE_USER_ID，跳過推播。")
return
warning_text = ""
if safety_warnings:
warning_text = f"\n\n⚠️【用藥安全提醒】\n" + "\n".join([f"• {w}" for w in safety_warnings])
push_text = f"💊【長輩用藥通知】\n剛才已完成藥袋辨識，共有 {medicines_count} 種藥品。{warning_text}\n\n白話摘要：\n{summary}"
try:
push_message_request = PushMessageRequest(
to=line_user_id,
messages=[TextMessage(text=push_text)]
)
line_bot_api.push_message(push_message_request)
print("✅ LINE 關懷推播發送成功！")
except Exception as e:
print(f"❌ LINE 推播發送失敗: {str(e)}")
# 4. JSON Schema 定義
prescription_schema = {
"type": "OBJECT",
"properties": {
"spoken_summary": {
"type": "STRING",
"description": "適合唸給長輩聽的白話文總結，語氣溫柔親切。"
},
"safety_warnings": {
"type": "ARRAY",
"items": {"type": "STRING"},
"description": "重複藥性或風險提醒，無則填空陣列。"
},
"medicines": {
"type": "ARRAY",
"items": {
"type": "OBJECT",
"properties": {
"name": {"type": "STRING"},
"type": {"type": "STRING"},
"frequency": {"type": "STRING"},
"dosage": {"type": "STRING"},
"timing": {"type": "STRING"},
"warning": {"type": "STRING"}
},
"required": ["name", "type", "frequency", "dosage", "timing"]
}
}
},
"required": ["spoken_summary", "safety_warnings", "medicines"]
}
@app.route('/ping', methods=['GET'])
def ping():
return jsonify({"status": "online", "service": "PrescriptionVLM Engine"}), 200
@app.route('/analyze-prescription', methods=['POST'])
def analyze_prescription():
if 'image' not in request.files:
return jsonify({"error": "unsupported_media_type", "message": "未提供圖片檔案"}), 400
file = request.files['image']
# 副檔名檢查機制
if file.filename == '' or not allowed_file(file.filename):
return jsonify({
"error": "unsupported_media_type",
"message": "不支援的檔案格式，請上傳 .jpg, .jpeg 或 .png 圖片。"
}), 400
try:
image = Image.open(file.stream)
prompt = """
你是一位專業且細心的藥師助手。請分析這張藥袋照片：
1. 將藥品分類為口服或外用，精準提取名稱、頻率、劑量與吃藥時間。
2. 檢查是否有重複藥性或高風險注意事項，填入 safety_warnings。
3. 針對高齡長輩，撰寫一段溫柔白話的 spoken_summary。
"""
config = types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=prescription_schema
)
response = client.models.generate_content(
model='gemini-3.6-flash',
contents=[image, prompt],
config=config
)
result_data = json.loads(response.text)
spoken_text = result_data.get("spoken_summary", "解析完成。")
safety_warnings = result_data.get("safety_warnings", [])
if safety_warnings:
prefix = "長輩請注意，這份藥單有特別需要留意的地方：" + "；".join(safety_warnings) + "。"
spoken_text = f"{prefix} {spoken_text}"
result_data['spoken_summary'] = spoken_text
prescription_id = uuid.uuid4().hex[:8]
filename = f"speech_{prescription_id}.mp3"
filepath = os.path.join(AUDIO_DIR, filename)
tts = gTTS(text=spoken_text, lang='zh-tw')
tts.save(filepath)
audio_url = f"/static/audio/{filename}"
result_data['audio_url'] = audio_url
result_data['prescription_id'] = prescription_id
# 寫入 SQLite
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
warnings_json = json.dumps(safety_warnings, ensure_ascii=False)
cursor.execute(
"INSERT INTO prescriptions (id, spoken_summary, audio_url, safety_warnings) VALUES (?, ?, ?, ?)",
(prescription_id, spoken_text, audio_url, warnings_json)
)
meds = result_data.get("medicines", [])
for med in meds:
cursor.execute(
"""INSERT INTO medicines
(prescription_id, name, type, frequency, dosage, timing, warning)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
prescription_id,
med.get("name"),
med.get("type"),
med.get("frequency"),
med.get("dosage"),
med.get("timing"),
med.get("warning", "")
)
)
conn.commit()
conn.close()
# 觸發 LINE 推播
send_line_notification(spoken_text, len(meds), safety_warnings)
return jsonify(result_data), 200
except Exception as e:
return jsonify({"error": f"伺服器處理失敗: {str(e)}"}), 500
@app.route('/prescriptions', methods=['GET'])
def get_prescriptions():
try:
conn = sqlite3.connect(DATABASE_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM prescriptions ORDER BY created_at DESC")
prescriptions = cursor.fetchall()
history = []
for p in prescriptions:
cursor.execute("SELECT name, type, frequency, dosage, timing, warning FROM medicines WHERE prescription_id = ?", (p['id'],))
meds = [dict(m) for m in cursor.fetchall()]
history.append({
"id": p['id'],
"spoken_summary": p['spoken_summary'],
"audio_url": p['audio_url'],
"safety_warnings": json.loads(p['safety_warnings']) if p['safety_warnings'] else [],
"created_at": p['created_at'],
"medicines": meds
})
conn.close()
return jsonify({
"status": "success",
"data": history
}), 200
except Exception as e:
return jsonify({"error": f"查詢失敗: {str(e)}"}), 500
@app.route('/static/audio/<filename>', methods=['GET'])
def get_audio(filename):
return send_from_directory(AUDIO_DIR, filename)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
在專案根目錄建立 test_e2e.py：
import os
import requests
import traceback
BASE_URL = "http://127.0.0.1:5000"
def test_health_check():
"""1. 驗證服務 Health Check"""
response = requests.get(f"{BASE_URL}/ping")
assert response.status_code == 200
assert response.json().get("status") == "online"
print("✅ [Pass] Health Check Endpoint")
def test_invalid_file_format():
"""2. 驗證非圖片檔案上傳的防禦機制"""
files = {"image": ("test.txt", b"Hello World", "text/plain")}
response = requests.post(f"{BASE_URL}/analyze-prescription", files=files)
assert response.status_code == 400
assert response.json().get("error") == "unsupported_media_type"
print("✅ [Pass] Edge Case: Non-image File Defense")
def test_analyze_prescription_e2e():
"""3. 驗證完整藥單辨識與流程整合"""
test_image_path = "test_rx.jpg"
if not os.path.exists(test_image_path):
print("⚠️ [Skip] 找不到 test_rx.jpg，跳過真實照片 E2E 測試")
return
with open(test_image_path, "rb") as img:
files = {"image": img}
response = requests.post(f"{BASE_URL}/analyze-prescription", files=files)
assert response.status_code == 200
data = response.json()
# 斷言關鍵欄位是否存在
assert "prescription_id" in data
assert "medicines" in data
assert "audio_url" in data
assert "safety_warnings" in data
print(f"✅ [Pass] E2E Prescription Analysis (ID: {data['prescription_id']})")
if __name__ == "__main__":
print("🚀 開始執行 Day 11 端到端整合測試...\n")
try:
test_health_check()
test_invalid_file_format()
test_analyze_prescription_e2e()
print("\n🎉 所有 E2E 整合與邊界測試項目通過！")
except AssertionError:
print("\n❌ 測試失敗：斷言條件不符合。")
traceback.print_exc()
except Exception as e:
print(f"\n❌ 測試異常：{e}")
traceback.print_exc()
更新 app.py 後，先重新構建並啟動 Docker 容器，再執行測試：
# 1. 強制移除舊容器並重新建置 Image
docker rm -f prescription_service
docker build -t prescription-vlm:v1.0 .
# 2. 啟動容器
docker run -d -p 5000:5000 --env-file .env --name prescription_service prescription-vlm:v1.0
# 3. 執行端到端自動化測試
python test_e2e.py
🚀 開始執行 Day 11 端到端整合測試...
✅ [Pass] Health Check Endpoint
✅ [Pass] Edge Case: Non-image File Defense
✅ [Pass] E2E Prescription Analysis (ID: 13)
🎉 所有 E2E 整合與邊界測試項目通過！
測試成功後，將更新後的 app.py 與 test_e2e.py 提交至 GitHub：
git add .
git commit -m "保留雙引號 改填寫自己要記錄的標記 ex.鐵人賽第十一天"
git push
今天完成了 API 檔案格式驗證與 E2E 自動化測試。系統在 Docker 環境下面對合法與異常輸入都能保持穩定。
明天（Day 12）進行 API Rate Limiting（限流保護）與 Resilience 機制。為系統加上防惡意刷單與 API 請求過載的保護。
iThome鐵人賽
看影片追技術