Drug Interaction MCP Server

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 中西藥交互作用資料庫操作模組 負責SQLite資料庫的連接、查詢和操作功能 """ import sqlite3 import logging from typing import List, Dict, Optional, Any, Tuple from pathlib import Path from datetime import datetime from .models import ( ChineseMedicine, WesternMedicine, DrugInteraction, SearchResult, InteractionSummary, BatchCheckResult ) class DrugInteractionDB: """中西藥交互作用資料庫操作類""" def __init__(self, db_path: str = "data/drug_interactions.db"): self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self.logger = logging.getLogger(__name__) def get_connection(self) -> sqlite3.Connection: """獲取資料庫連接（簡化版）""" conn = sqlite3.connect( self.db_path, check_same_thread=False, timeout=30.0 ) conn.row_factory = sqlite3.Row # 讓結果可以通過列名訪問 conn.execute("PRAGMA foreign_keys = ON") # 啟用外鍵約束 return conn def close_connections(self): """關閉連接（MCP服務器中通常不需要）""" # MCP服務器通常長期運行，不需要關閉連接 pass def initialize_database(self) -> bool: """初始化資料庫結構""" try: conn = self.get_connection() cursor = conn.cursor() # 創建中藥表 cursor.execute(""" CREATE TABLE IF NOT EXISTS chinese_medicines ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, name_pinyin TEXT, category TEXT, properties TEXT, functions TEXT, side_effects TEXT, contraindications TEXT, dosage TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # 創建西藥表 cursor.execute(""" CREATE TABLE IF NOT EXISTS western_medicines ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, generic_name TEXT, brand_name TEXT, category TEXT, therapeutic_class TEXT, mechanism TEXT, side_effects TEXT, contraindications TEXT, dosage TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # 創建交互作用表 cursor.execute(""" CREATE TABLE IF NOT EXISTS interactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, chinese_medicine_id INTEGER, western_medicine_id INTEGER, interaction_type TEXT, severity TEXT NOT NULL, mechanism TEXT, clinical_effects TEXT, recommendation TEXT, evidence_level TEXT, ref_sources TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (chinese_medicine_id) REFERENCES chinese_medicines(id), FOREIGN KEY (western_medicine_id) REFERENCES western_medicines(id) ) """) # 創建索引以提升查詢效能 indexes = [ "CREATE INDEX IF NOT EXISTS idx_chinese_name ON chinese_medicines(name)", "CREATE INDEX IF NOT EXISTS idx_chinese_pinyin ON chinese_medicines(name_pinyin)", "CREATE INDEX IF NOT EXISTS idx_chinese_category ON chinese_medicines(category)", "CREATE INDEX IF NOT EXISTS idx_western_name ON western_medicines(name)", "CREATE INDEX IF NOT EXISTS idx_western_category ON western_medicines(category)", "CREATE INDEX IF NOT EXISTS idx_interactions_chinese ON interactions(chinese_medicine_id)", "CREATE INDEX IF NOT EXISTS idx_interactions_western ON interactions(western_medicine_id)", "CREATE INDEX IF NOT EXISTS idx_interactions_severity ON interactions(severity)", "CREATE INDEX IF NOT EXISTS idx_interactions_type ON interactions(interaction_type)" ] for index_sql in indexes: cursor.execute(index_sql) conn.commit() self.logger.info("資料庫結構初始化成功") return True except Exception as e: self.logger.error(f"資料庫初始化失敗: {e}") return False finally: if 'conn' in locals(): conn.close() def load_test_data(self, sql_file: str = "data/init_data.sql") -> bool: """載入測試資料""" try: sql_path = Path(sql_file) if not sql_path.exists(): self.logger.error(f"SQL檔案不存在: {sql_file}") return False with open(sql_path, 'r', encoding='utf-8') as f: sql_content = f.read() conn = self.get_connection() cursor = conn.cursor() # 執行SQL腳本（分割成單獨的語句） # 首先移除行註釋和空行 lines = [] for line in sql_content.split('\n'): line = line.strip() # 跳過空行和純註釋行 if line and not line.startswith('--'): lines.append(line) # 將所有行重新組合並按分號分割 processed_content = '\n'.join(lines) sql_statements = [stmt.strip() for stmt in processed_content.split(';') if stmt.strip()] for stmt in sql_statements: if stmt: try: cursor.execute(stmt) except sqlite3.Error as e: self.logger.warning(f"執行SQL語句時出現警告: {e}") self.logger.debug(f"語句: {stmt[:100]}...") conn.commit() self.logger.info("測試資料載入成功") return True except Exception as e: self.logger.error(f"載入測試資料失敗: {e}") return False finally: if 'conn' in locals(): conn.close() def search_chinese_medicine(self, keyword: str, limit: int = 20) -> List[ChineseMedicine]: """搜尋中藥""" try: conn = self.get_connection() cursor = conn.cursor() # 使用LIKE進行模糊搜尋，支援中文和拼音 cursor.execute(""" SELECT id, name, name_pinyin, category, properties, functions, side_effects, contraindications, dosage FROM chinese_medicines WHERE name LIKE ? OR name_pinyin LIKE ? OR category LIKE ? ORDER BY CASE WHEN name LIKE ? THEN 1 ELSE 2 END, name LIMIT ? """, ( f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'{keyword}%', limit )) results = [] for row in cursor.fetchall(): medicine = ChineseMedicine( id=row['id'], name=row['name'], name_pinyin=row['name_pinyin'], category=row['category'], properties=row['properties'], functions=row['functions'], side_effects=row['side_effects'], contraindications=row['contraindications'], dosage=row['dosage'] ) results.append(medicine) return results except Exception as e: self.logger.error(f"搜尋中藥時發生錯誤: {e}") return [] finally: if 'conn' in locals(): conn.close() def search_western_medicine(self, keyword: str, limit: int = 20) -> List[WesternMedicine]: """搜尋西藥""" try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" SELECT id, name, generic_name, brand_name, category, therapeutic_class, mechanism, side_effects, contraindications, dosage FROM western_medicines WHERE name LIKE ? OR generic_name LIKE ? OR brand_name LIKE ? OR category LIKE ? ORDER BY CASE WHEN name LIKE ? THEN 1 ELSE 2 END, name LIMIT ? """, ( f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', f'{keyword}%', limit )) results = [] for row in cursor.fetchall(): medicine = WesternMedicine( id=row['id'], name=row['name'], generic_name=row['generic_name'], brand_name=row['brand_name'], category=row['category'], therapeutic_class=row['therapeutic_class'], mechanism=row['mechanism'], side_effects=row['side_effects'], contraindications=row['contraindications'], dosage=row['dosage'] ) results.append(medicine) return results except Exception as e: self.logger.error(f"搜尋西藥時發生錯誤: {e}") return [] finally: if 'conn' in locals(): conn.close() def check_interaction(self, chinese_medicine_id: int, western_medicine_id: int) -> Optional[DrugInteraction]: """檢查特定中藥和西藥的交互作用""" try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" SELECT i.id, i.chinese_medicine_id, i.western_medicine_id, cm.name as chinese_name, wm.name as western_name, i.interaction_type, i.severity, i.mechanism, i.clinical_effects, i.recommendation, i.evidence_level, i.ref_sources FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE i.chinese_medicine_id = ? AND i.western_medicine_id = ? """, (chinese_medicine_id, western_medicine_id)) row = cursor.fetchone() if row: return DrugInteraction( id=row['id'], chinese_medicine_id=row['chinese_medicine_id'], western_medicine_id=row['western_medicine_id'], chinese_name=row['chinese_name'], western_name=row['western_name'], interaction_type=row['interaction_type'], severity=row['severity'], mechanism=row['mechanism'], clinical_effects=row['clinical_effects'], recommendation=row['recommendation'], evidence_level=row['evidence_level'], ref_sources=row['ref_sources'] ) return None except Exception as e: self.logger.error(f"檢查交互作用時發生錯誤: {e}") return None finally: if 'conn' in locals(): conn.close() def get_interactions_by_chinese_medicine(self, chinese_medicine_id: int) -> List[DrugInteraction]: """獲取指定中藥的所有交互作用""" try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" SELECT i.id, i.interaction_type, i.severity, i.mechanism, i.clinical_effects, i.recommendation, i.evidence_level, i.ref_sources, cm.id as chinese_medicine_id, cm.name as chinese_name, wm.id as western_medicine_id, wm.name as western_name FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE i.chinese_medicine_id = ? ORDER BY i.severity DESC """, (chinese_medicine_id,)) results = [] for row in cursor.fetchall(): interaction = DrugInteraction( id=row['id'], chinese_medicine_id=row['chinese_medicine_id'], western_medicine_id=row['western_medicine_id'], chinese_name=row['chinese_name'], western_name=row['western_name'], interaction_type=row['interaction_type'], severity=row['severity'], mechanism=row['mechanism'], clinical_effects=row['clinical_effects'], recommendation=row['recommendation'], evidence_level=row['evidence_level'], ref_sources=row['ref_sources'] ) results.append(interaction) return results except Exception as e: self.logger.error(f"獲取中藥交互作用時發生錯誤: {e}") return [] finally: if 'conn' in locals(): conn.close() def get_interactions_by_western_medicine(self, western_medicine_id: int) -> List[DrugInteraction]: """獲取指定西藥的所有交互作用""" try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" SELECT i.id, i.chinese_medicine_id, i.western_medicine_id, cm.name as chinese_name, wm.name as western_name, i.interaction_type, i.severity, i.mechanism, i.clinical_effects, i.recommendation, i.evidence_level, i.ref_sources FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE i.western_medicine_id = ? ORDER BY CASE i.severity WHEN '致命' THEN 1 WHEN '重度' THEN 2 WHEN '中度' THEN 3 WHEN '輕度' THEN 4 ELSE 5 END, cm.name """, (western_medicine_id,)) results = [] for row in cursor.fetchall(): interaction = DrugInteraction( id=row['id'], chinese_medicine_id=row['chinese_medicine_id'], western_medicine_id=row['western_medicine_id'], chinese_name=row['chinese_name'], western_name=row['western_name'], interaction_type=row['interaction_type'], severity=row['severity'], mechanism=row['mechanism'], clinical_effects=row['clinical_effects'], recommendation=row['recommendation'], evidence_level=row['evidence_level'], ref_sources=row['ref_sources'] ) results.append(interaction) return results except Exception as e: self.logger.error(f"獲取西藥交互作用時發生錯誤: {e}") return [] finally: if 'conn' in locals(): conn.close() def batch_check_interactions(self, chinese_medicines: List[str], western_medicines: List[str]) -> BatchCheckResult: """批次檢查多個中藥和西藥的交互作用""" try: conn = self.get_connection() cursor = conn.cursor() # 獲取中藥和西藥的ID placeholders = ','.join(['?' for _ in chinese_medicines]) cursor.execute(f""" SELECT id, name FROM chinese_medicines WHERE name IN ({placeholders}) """, chinese_medicines) chinese_ids = {row['name']: row['id'] for row in cursor.fetchall()} placeholders = ','.join(['?' for _ in western_medicines]) cursor.execute(f""" SELECT id, name FROM western_medicines WHERE name IN ({placeholders}) """, western_medicines) western_ids = {row['name']: row['id'] for row in cursor.fetchall()} # 檢查所有組合 interactions_found = [] high_risk_pairs = 0 for cm_name, cm_id in chinese_ids.items(): for wm_name, wm_id in western_ids.items(): interaction = self.check_interaction(cm_id, wm_id) if interaction: interactions_found.append(interaction) if interaction.severity in ['重度', '致命']: high_risk_pairs += 1 total_checked_pairs = len(chinese_ids) * len(western_ids) # 生成警告訊息 warnings = [] if high_risk_pairs > 0: warnings.append(f"發現 {high_risk_pairs} 個高風險交互作用組合，請特別注意用藥安全") return BatchCheckResult( chinese_medicines=chinese_medicines, western_medicines=western_medicines, interactions_found=interactions_found, total_checked_pairs=total_checked_pairs, high_risk_pairs=high_risk_pairs, warnings=warnings ) except Exception as e: self.logger.error(f"批次檢查交互作用時發生錯誤: {e}") return BatchCheckResult( chinese_medicines=chinese_medicines, western_medicines=western_medicines, interactions_found=[], total_checked_pairs=0, high_risk_pairs=0, warnings=[f"查詢過程中發生錯誤: {str(e)}"] ) finally: if 'conn' in locals(): conn.close() def get_interaction_summary(self, medicine_name: str, medicine_type: str) -> Optional[InteractionSummary]: """獲取藥物交互作用摘要""" try: conn = self.get_connection() cursor = conn.cursor() if medicine_type == 'chinese': cursor.execute(""" SELECT COUNT(*) as total FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id WHERE cm.name = ? """, (medicine_name,)) total_interactions = cursor.fetchone()['total'] cursor.execute(""" SELECT i.severity, COUNT(*) as count FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id WHERE cm.name = ? GROUP BY i.severity """, (medicine_name,)) severity_dist = {row['severity']: row['count'] for row in cursor.fetchall()} cursor.execute(""" SELECT i.id, i.chinese_medicine_id, i.western_medicine_id, cm.name as chinese_name, wm.name as western_name, i.interaction_type, i.severity, i.mechanism, i.clinical_effects, i.recommendation, i.evidence_level, i.ref_sources FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE cm.name = ? AND i.severity IN ('重度', '致命') ORDER BY CASE i.severity WHEN '致命' THEN 1 WHEN '重度' THEN 2 ELSE 3 END LIMIT 5 """, (medicine_name,)) else: # western medicine cursor.execute(""" SELECT COUNT(*) as total FROM interactions i JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE wm.name = ? """, (medicine_name,)) total_interactions = cursor.fetchone()['total'] cursor.execute(""" SELECT i.severity, COUNT(*) as count FROM interactions i JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE wm.name = ? GROUP BY i.severity """, (medicine_name,)) severity_dist = {row['severity']: row['count'] for row in cursor.fetchall()} cursor.execute(""" SELECT i.id, i.chinese_medicine_id, i.western_medicine_id, cm.name as chinese_name, wm.name as western_name, i.interaction_type, i.severity, i.mechanism, i.clinical_effects, i.recommendation, i.evidence_level, i.ref_sources FROM interactions i JOIN chinese_medicines cm ON i.chinese_medicine_id = cm.id JOIN western_medicines wm ON i.western_medicine_id = wm.id WHERE wm.name = ? AND i.severity IN ('重度', '致命') ORDER BY CASE i.severity WHEN '致命' THEN 1 WHEN '重度' THEN 2 ELSE 3 END LIMIT 5 """, (medicine_name,)) high_risk_interactions = [] for row in cursor.fetchall(): interaction = DrugInteraction( id=row['id'], chinese_medicine_id=row['chinese_medicine_id'], western_medicine_id=row['western_medicine_id'], chinese_name=row['chinese_name'], western_name=row['western_name'], interaction_type=row['interaction_type'], severity=row['severity'], mechanism=row['mechanism'], clinical_effects=row['clinical_effects'], recommendation=row['recommendation'], evidence_level=row['evidence_level'], ref_sources=row['ref_sources'] ) high_risk_interactions.append(interaction) # 生成建議 recommendations = [] if severity_dist.get('致命', 0) > 0: recommendations.append("此藥物存在致命交互作用，禁止同時使用") if severity_dist.get('重度', 0) > 0: recommendations.append("此藥物存在重度交互作用，需要嚴密監測") if severity_dist.get('中度', 0) > 0: recommendations.append("此藥物存在中度交互作用，建議調整劑量或服藥時間") if not recommendations: recommendations.append("目前無已知嚴重交互作用，但仍需注意監測") return InteractionSummary( medicine_name=medicine_name, medicine_type=medicine_type, total_interactions=total_interactions, severity_distribution=severity_dist, high_risk_interactions=high_risk_interactions, recommendations=recommendations ) except Exception as e: self.logger.error(f"獲取交互作用摘要時發生錯誤: {e}") return None finally: if 'conn' in locals(): conn.close()