LINE Bot 必須能接收家屬或長輩的回饋，才能形成完整的雙向溝通迴圈。這次實作透過 LINE Webhook 機制讓服務即時監聽使用者訊息，自動將「已服藥」狀態寫回 SQLite 資料庫。完成項目：
為了追蹤長輩每日的服藥狀況，在 SQLite 中新增 intake_logs 資料表。
SQL Schema 設計
CREATE TABLE IF NOT EXISTS intake_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prescription_id TEXT,
user_id TEXT NOT NULL,
status TEXT NOT NULL, -- 'taken' 或 'skipped'
taken_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
使用 SQLite CLI 手動擴充資料庫
為避免刪除原有測試藥單資料，使用 Terminal 直接進入 SQLite CLI 完成建表：
# 進入 SQLite CLI 介面
sqlite3 prescription_vlm.db
在 sqlite> 提示字元下貼入 SQL 指令並驗證：
-- 1. 建立服藥紀錄表
CREATE TABLE IF NOT EXISTS intake_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prescription_id TEXT,
user_id TEXT NOT NULL,
status TEXT NOT NULL,
taken_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 2. 驗證是否建表成功（預期列出 intake_logs, medicines, prescriptions）
.tables
-- 3. 離開 SQLite CLI
.exit
app.py 整合 LINE Webhook 雙向互動機制開啟 app.py，引入 WebhookHandler，新增 /callback 路由與訊息監聽器：
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,
ReplyMessageRequest,
TextMessage
)
from linebot.v3.webhook import WebhookHandler
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.webhooks import MessageEvent, TextMessageContent
# 配置 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_log = logging.StreamHandler()
handler_log.setFormatter(JsonFormatter())
logger = logging.getLogger("prescription_service")
logger.setLevel(logging.INFO)
logger.addHandler(handler_log)
# 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")
line_secret = os.getenv("LINE_CHANNEL_SECRET")
if not api_key:
logger.error("找不到 GEMINI_API_KEY，請檢查 .env 設定！")
raise ValueError("❌ 錯誤：找不到 GEMINI_API_KEY，請檢查 .env 設定！")
client = genai.Client(api_key=api_key)
# 初始化 LINE SDK
line_bot_api = None
handler = None
if line_access_token and line_secret:
configuration = Configuration(access_token=line_access_token)
line_api_client = ApiClient(configuration)
line_bot_api = MessagingApi(line_api_client)
handler = WebhookHandler(line_secret)
else:
logger.warning("未完整設定 LINE Token 或 Secret，Webhook 與推播功能將受限。")
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. 資料庫初始化 (包含 intake_logs 表防呆)
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)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS intake_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
prescription_id TEXT,
user_id TEXT NOT NULL,
status TEXT NOT NULL,
taken_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
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)}")
# 4. LINE Webhook 接收與處理
@app.route("/callback", methods=['POST'])
def callback():
signature = request.headers.get('X-Line-Signature', '')
body = request.get_data(as_text=True)
if not handler:
logger.error("Webhook 處理失敗：未設定 LINE Channel Secret")
return 'LINE configuration error', 500
try:
handler.handle(body, signature)
except Exception as e:
logger.warning(f"Webhook 處理紀錄: {str(e)}")
response = Flask.response_class(
response="OK",
status=200,
mimetype='text/plain'
)
response.headers['ngrok-skip-browser-warning'] = 'true'
return response
# 只有在 handler 成功建立時註冊事件監聽
if handler:
@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
user_msg = event.message.text.strip()
user_id = event.source.user_id
if any(keyword in user_msg for keyword in ["吃藥", "服藥", "吃了"]):
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
cursor.execute(
"INSERT INTO intake_logs (user_id, status) VALUES (?, ?)",
(user_id, "taken")
)
conn.commit()
conn.close()
reply_text = "❤️ 收到！已為您記錄今天的服藥狀況。請繼續保持，祝您身體健康！"
logger.info("已記錄服藥狀態", extra={"extra_data": {"user_id": user_id, "status": "taken"}})
else:
reply_text = "您好！我是您的藥單關懷助手。如果今天已經吃過藥了，請回覆「已吃藥」讓我為您記錄喔！"
line_bot_api.reply_message(
ReplyMessageRequest(
reply_token=event.reply_token,
messages=[TextMessage(text=reply_text)]
)
)
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)
1. 取得 Channel Secret
至 LINE Developers Console 點進頻道，於 Basic settings 底下複製 Channel secret，填入 .env（不加雙引號與空白）：
LINE_CHANNEL_SECRET=你的ChannelSecret字串
2.安裝與啟動 ngrok 內網穿透
前往 ngrok 官網 下載或透過指令安裝：
# Mac 使用者 (Homebrew)
brew install ngrok/ngrok/ngrok
# Windows 使用者 (PowerShell)
winget install ngrok.ngrok
啟動穿透隧道映射至 5000 Port：
ngrok http 5000
(執行後請複製畫面上產生的 https://xxxx.ngrok-free.app 網址)
3.設定 LINE Webhook URL
至 Messaging API 頁籤，修改 Webhook URL 為：https://xxxx.ngrok-free.app/callback（請替換為實際網址），開啟 Use webhook 開關並點擊 Verify 驗證通過。
啟動 Docker 並以手機 LINE 進行實測：
# 1. 重新啟動容器
docker rm -f prescription_service
docker run -d -p 5000:5000 --env-file .env --name prescription_service prescription-vlm:v1.0
# 2. 檢視 Webhook 接收日誌
docker logs -f prescription_service
用手機 LINE 傳送訊息：「我吃藥了」。
手機即時收到關懷機器人回覆：「❤️ 收到！已為您記錄今天的服藥狀況...」。
Docker 容器日誌印出標準 JSON 紀錄：
{"timestamp": "2026-09-14 15:38:10,120", "level": "INFO", "message": "已記錄服藥狀態", "module": "app", "user_id": "U7884631...", "status": "taken"}
測試成功後，將更新後的檔案提交至 GitHub：
git add .
git commit -m "保留雙引號 改填寫自己要記錄的標記 ex.鐵人賽第十四天"
git push
今天實現了 LINE Bot 的雙向互動閉環。家屬與長輩能透過簡單對話反饋服藥狀態，自動寫入 SQLite 資料庫持久化紀錄。
明天（Day 15）進行 Web 管理儀表板實作。讓使用者能在 Web 介面上即時查看藥單歷程與服藥紀錄。
iThome鐵人賽
看影片追技術