今天進入微服務的「可觀測性（Observability）」強化階段。在生產環境中純文字的 console log 難以被分析工具處理與檢索。今天將日誌升級為 JSON 格式的結構化日誌，並追蹤每筆請求的生命週期。完成項目包含：
引入 Python logging 標準庫與 JSON Log Formatter — 替換傳統 print()，將所有系統 Event 與 API 呼叫格式化為 JSON 結構。
app.py 完整日誌紀錄點 — 包含 VLM 呼叫耗時、SQLite 讀寫狀態與 LINE 推播結果。為了讓未來的 Log 管理工具（如 ELK、Grafana Loki、Datadog）方便解析，定義統一的 JSON 日誌結構：
{
"timestamp": "2026-09-14 11:30:00",
"level": "INFO",
"event": "prescription_analyzed",
"prescription_id": "ca642acf",
"duration_ms": 1250,
"status_code": 200,
"client_ip": "127.0.0.1"
}
app.py 整合 Structured Logging 機制開啟 app.py，導入 logging 與 time 模組，並設置 JSON Formatter 與 Request 處理計時：
import os
import json
import uuid
import time
import sqlite3
import logging
from flask import Flask, request, jsonify, send_from_directory
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
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
)
# 配置 Structured JSON Logger
class JsonFormatter(logging.Formatter):
def format(self, record):
log_record = {
"timestamp": self.formatTime(record, self.datefmt),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module
}
if hasattr(record, "extra_data"):
log_record.update(record.extra_data)
return json.dumps(log_record, ensure_ascii=False)
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("prescription_service")
logger.setLevel(logging.INFO)
logger.addHandler(handler)
# 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:
logger.error("找不到 GEMINI_API_KEY，請檢查 .env 設定！")
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
limiter = Limiter(
get_remote_address,
app=app,
default_limits=["200 per day", "50 per hour"],
storage_uri="memory://"
)
@app.errorhandler(429)
def ratelimit_handler(e):
logger.warning("觸發 Rate Limit 流量限制", extra={"extra_data": {"client_ip": request.remote_addr, "status_code": 429}})
return jsonify({
"error": "rate_limit_exceeded",
"message": "請求過於頻繁，系統保護中。請稍後再試。",
"detail": str(e.description)
}), 429
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:
logger.warning("未設定 LINE Token 或 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)
logger.info("LINE 關懷推播發送成功", extra={"extra_data": {"medicines_count": medicines_count}})
except Exception as e:
logger.error(f"LINE 推播發送失敗: {str(e)}")
prescription_schema = {
"type": "OBJECT",
"properties": {
"spoken_summary": {"type": "STRING"},
"safety_warnings": {"type": "ARRAY", "items": {"type": "STRING"}},
"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'])
@limiter.limit("5 per minute")
def analyze_prescription():
start_time = time.time()
if 'image' not in request.files:
logger.warning("請求缺少圖片檔案", extra={"extra_data": {"status_code": 400}})
return jsonify({"error": "unsupported_media_type", "message": "未提供圖片檔案"}), 400
file = request.files['image']
if file.filename == '' or not allowed_file(file.filename):
logger.warning("上傳不支援的檔案格式", extra={"extra_data": {"filename": file.filename, "status_code": 400}})
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
)
vlm_start = time.time()
response = client.models.generate_content(
model='gemini-3.6-flash',
contents=[image, prompt],
config=config
)
vlm_duration = round((time.time() - vlm_start) * 1000, 2)
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
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()
send_line_notification(spoken_text, len(meds), safety_warnings)
total_duration = round((time.time() - start_time) * 1000, 2)
logger.info(
"藥單解析流程完成",
extra={
"extra_data": {
"prescription_id": prescription_id,
"vlm_duration_ms": vlm_duration,
"total_duration_ms": total_duration,
"medicines_count": len(meds),
"warnings_count": len(safety_warnings),
"status_code": 200
}
}
)
return jsonify(result_data), 200
except Exception as e:
total_duration = round((time.time() - start_time) * 1000, 2)
logger.error(
f"伺服器處理失敗: {str(e)}",
extra={"extra_data": {"total_duration_ms": total_duration, "status_code": 500}}
)
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:
logger.error(f"查詢歷史紀錄失敗: {str(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)
更新 app.py 後，重新構建 Docker 容器並使用 test_e2e.py 觸發測試，觀察 Console 輸出的 JSON Log 結構：
# 1. 重新建置與啟動 Docker 容器
docker rm -f prescription_service
docker build -t prescription-vlm:v1.0 .
docker run -d -p 5000:5000 --env-file .env --name prescription_service prescription-vlm:v1.0
# 2. 執行 E2E 測試以產生 Log
python test_e2e.py
# 3. 檢視 Docker 容器的 JSON Structured Logs
docker logs prescription_service
{"timestamp": "2026-09-14 11:30:15", "level": "WARNING", "message": "上傳不支援的檔案格式", "module": "app", "filename": "test.txt", "status_code": 400}
{"timestamp": "2026-09-14 11:30:18", "level": "INFO", "message": "藥單解析流程完成", "module": "app", "prescription_id": "8f3a1b2c", "vlm_duration_ms": 1120.5, "total_duration_ms": 1450.2, "medicines_count": 3, "warnings_count": 0, "status_code": 200}
測試成功後，將更新後的檔案提交至 GitHub：
git add .
git commit -m "保留雙引號 改填寫自己要記錄的標記 ex.鐵人賽第十三天"
git push
今天為微服務加入了生產等級的 Structured JSON Logging，實現了 API 請求時間、呼叫耗時與邊界例外的完整追蹤。
明天（Day 14）進入前端／LINE Bot 雙向互動模組。實作家屬接收 Card 訊息並回覆「已服藥」更新 SQLite 狀態的閉環流程。
iThome鐵人賽
看影片追技術