tools.py•25.1 kB
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
中西藥交互作用MCP工具實現
實現7個核心MCP工具，用於查詢和管理中西藥交互作用資訊
"""
import logging
from typing import List, Dict, Any, Optional
from .database import DrugInteractionDB
from .models import ChineseMedicine, WesternMedicine, DrugInteraction, InteractionSummary, BatchCheckResult
from .common import handle_errors, validate_input, create_error_response, create_success_response
class DrugInteractionTools:
"""中西藥交互作用工具類"""
def __init__(self, db_path: str = "data/drug_interactions.db"):
"""初始化工具類"""
self.db = DrugInteractionDB(db_path)
self.logger = logging.getLogger(__name__)
# 確保資料庫已初始化
self._initialize_database()
def _initialize_database(self):
"""初始化資料庫"""
try:
if not self.db.initialize_database():
raise Exception("資料庫初始化失敗")
# 檢查是否有資料，如果沒有則載入測試資料
conn = self.db.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) as count FROM chinese_medicines")
if cursor.fetchone()['count'] == 0:
self.logger.info("載入測試資料...")
if not self.db.load_test_data():
raise Exception("載入測試資料失敗")
except Exception as e:
self.logger.error(f"資料庫初始化錯誤: {e}")
raise
finally:
self.db.close_connections()
@handle_errors()
def search_chinese_medicine(self, keyword: str, limit: int = 20) -> Dict[str, Any]:
"""搜尋中藥"""
# 驗證輸入
validation = validate_input(keyword=keyword)
if not validation['valid']:
return create_error_response(validation['errors'][0])
keyword = validation['data']['keyword'].strip()
self.logger.info(f"搜尋中藥: {keyword}")
medicines = self.db.search_chinese_medicine(keyword, limit)
if not medicines:
return create_success_response(
f"未找到包含「{keyword}」的中藥",
data=[],
count=0
)
# 轉換為字典格式
results = []
for medicine in medicines:
results.append({
"id": medicine.id,
"name": medicine.name,
"name_pinyin": medicine.name_pinyin,
"category": medicine.category,
"properties": medicine.properties,
"functions": medicine.functions,
"side_effects": medicine.side_effects,
"contraindications": medicine.contraindications,
"dosage": medicine.dosage
})
return create_success_response(
f"找到 {len(results)} 個中藥",
data=results,
count=len(results)
)
@handle_errors()
def search_western_medicine(self, keyword: str, limit: int = 20) -> Dict[str, Any]:
"""搜尋西藥"""
# 驗證輸入
validation = validate_input(keyword=keyword)
if not validation['valid']:
return create_error_response(validation['errors'][0])
keyword = validation['data']['keyword'].strip()
self.logger.info(f"搜尋西藥: {keyword}")
medicines = self.db.search_western_medicine(keyword, limit)
if not medicines:
return create_success_response(
f"未找到包含「{keyword}」的西藥",
data=[],
count=0
)
# 轉換為字典格式
results = []
for medicine in medicines:
results.append({
"id": medicine.id,
"name": medicine.name,
"generic_name": medicine.generic_name,
"brand_name": medicine.brand_name,
"category": medicine.category,
"therapeutic_class": medicine.therapeutic_class,
"mechanism": medicine.mechanism,
"side_effects": medicine.side_effects,
"contraindications": medicine.contraindications,
"dosage": medicine.dosage
})
return create_success_response(
f"找到 {len(results)} 個西藥",
data=results,
count=len(results)
)
def check_interactions(self, chinese_medicine: str, western_medicine: str) -> Dict[str, Any]:
"""檢查特定中藥和西藥的交互作用"""
try:
self.logger.info(f"檢查交互作用: {chinese_medicine} + {western_medicine}")
# 驗證輸入參數
if not chinese_medicine or not chinese_medicine.strip():
return {
"success": False,
"error": "請提供中藥名稱",
"data": None
}
if not western_medicine or not western_medicine.strip():
return {
"success": False,
"error": "請提供西藥名稱",
"data": None
}
chinese_medicine = chinese_medicine.strip()
western_medicine = western_medicine.strip()
# 查找中藥和西藥的ID
chinese_meds = self.db.search_chinese_medicine(chinese_medicine, 1)
if not chinese_meds:
return {
"success": False,
"error": f"未找到中藥: {chinese_medicine}",
"data": None
}
western_meds = self.db.search_western_medicine(western_medicine, 1)
if not western_meds:
return {
"success": False,
"error": f"未找到西藥: {western_medicine}",
"data": None
}
chinese_medicine_id = chinese_meds[0].id
western_medicine_id = western_meds[0].id
# 檢查交互作用
interaction = self.db.check_interaction(chinese_medicine_id, western_medicine_id)
if interaction:
result = {
"found": True,
"chinese_medicine": {
"name": interaction.chinese_name,
"id": interaction.chinese_medicine_id
},
"western_medicine": {
"name": interaction.western_name,
"id": interaction.western_medicine_id
},
"interaction": {
"type": interaction.interaction_type,
"severity": interaction.severity,
"mechanism": interaction.mechanism,
"clinical_effects": interaction.clinical_effects,
"recommendation": interaction.recommendation,
"evidence_level": interaction.evidence_level,
"references": interaction.ref_sources
}
}
return {
"success": True,
"message": f"發現 {interaction.severity} 程度交互作用",
"data": result
}
else:
return {
"success": True,
"message": f"未發現 {chinese_medicine} 與 {western_medicine} 的已知交互作用",
"data": {
"found": False,
"chinese_medicine": {"name": chinese_medicine},
"western_medicine": {"name": western_medicine},
"note": "請注意，這並不意味著完全安全，仍需謹慎使用"
}
}
except Exception as e:
self.logger.error(f"檢查交互作用時發生錯誤: {e}")
self.logger.error(traceback.format_exc())
return {
"success": False,
"error": f"檢查過程中發生錯誤: {str(e)}",
"data": None
}
def find_interactions_by_chinese_medicine(self, chinese_medicine: str) -> Dict[str, Any]:
"""根據中藥查找所有相關交互作用"""
try:
self.logger.info(f"查找中藥交互作用: {chinese_medicine}")
if not chinese_medicine or not chinese_medicine.strip():
return {
"success": False,
"error": "請提供中藥名稱",
"data": None
}
chinese_medicine = chinese_medicine.strip()
# 查找中藥ID
chinese_meds = self.db.search_chinese_medicine(chinese_medicine, 1)
if not chinese_meds:
return {
"success": False,
"error": f"未找到中藥: {chinese_medicine}",
"data": None
}
chinese_medicine_id = chinese_meds[0].id
# 查找所有交互作用
interactions = self.db.get_interactions_by_chinese_medicine(chinese_medicine_id)
if not interactions:
return {
"success": True,
"message": f"未找到 {chinese_medicine} 的已知交互作用",
"data": {
"chinese_medicine": {"name": chinese_medicine},
"interactions": [],
"count": 0
}
}
# 轉換為字典格式並按嚴重程度排序
results = []
severity_order = {"致命": 1, "重度": 2, "中度": 3, "輕度": 4}
interactions.sort(key=lambda x: severity_order.get(x.severity, 5))
for interaction in interactions:
results.append({
"id": interaction.id,
"western_medicine": {
"name": interaction.western_name,
"id": interaction.western_medicine_id
},
"interaction": {
"type": interaction.interaction_type,
"severity": interaction.severity,
"mechanism": interaction.mechanism,
"clinical_effects": interaction.clinical_effects,
"recommendation": interaction.recommendation,
"evidence_level": interaction.evidence_level,
"references": interaction.ref_sources
}
})
# 統計嚴重程度分布
severity_stats = {}
for interaction in interactions:
severity = interaction.severity
severity_stats[severity] = severity_stats.get(severity, 0) + 1
return {
"success": True,
"message": f"找到 {chinese_medicine} 的 {len(results)} 個交互作用",
"data": {
"chinese_medicine": {
"name": chinese_medicine,
"id": chinese_medicine_id
},
"interactions": results,
"count": len(results),
"severity_distribution": severity_stats
}
}
except Exception as e:
self.logger.error(f"查找中藥交互作用時發生錯誤: {e}")
self.logger.error(traceback.format_exc())
return {
"success": False,
"error": f"查詢過程中發生錯誤: {str(e)}",
"data": None
}
def find_interactions_by_western_medicine(self, western_medicine: str) -> Dict[str, Any]:
"""根據西藥查找所有相關交互作用"""
try:
self.logger.info(f"查找西藥交互作用: {western_medicine}")
if not western_medicine or not western_medicine.strip():
return {
"success": False,
"error": "請提供西藥名稱",
"data": None
}
western_medicine = western_medicine.strip()
# 查找西藥ID
western_meds = self.db.search_western_medicine(western_medicine, 1)
if not western_meds:
return {
"success": False,
"error": f"未找到西藥: {western_medicine}",
"data": None
}
western_medicine_id = western_meds[0].id
# 查找所有交互作用
interactions = self.db.get_interactions_by_western_medicine(western_medicine_id)
if not interactions:
return {
"success": True,
"message": f"未找到 {western_medicine} 的已知交互作用",
"data": {
"western_medicine": {"name": western_medicine},
"interactions": [],
"count": 0
}
}
# 轉換為字典格式並按嚴重程度排序
results = []
severity_order = {"致命": 1, "重度": 2, "中度": 3, "輕度": 4}
interactions.sort(key=lambda x: severity_order.get(x.severity, 5))
for interaction in interactions:
results.append({
"id": interaction.id,
"chinese_medicine": {
"name": interaction.chinese_name,
"id": interaction.chinese_medicine_id
},
"interaction": {
"type": interaction.interaction_type,
"severity": interaction.severity,
"mechanism": interaction.mechanism,
"clinical_effects": interaction.clinical_effects,
"recommendation": interaction.recommendation,
"evidence_level": interaction.evidence_level,
"references": interaction.ref_sources
}
})
# 統計嚴重程度分布
severity_stats = {}
for interaction in interactions:
severity = interaction.severity
severity_stats[severity] = severity_stats.get(severity, 0) + 1
return {
"success": True,
"message": f"找到 {western_medicine} 的 {len(results)} 個交互作用",
"data": {
"western_medicine": {
"name": western_medicine,
"id": western_medicine_id
},
"interactions": results,
"count": len(results),
"severity_distribution": severity_stats
}
}
except Exception as e:
self.logger.error(f"查找西藥交互作用時發生錯誤: {e}")
self.logger.error(traceback.format_exc())
return {
"success": False,
"error": f"查詢過程中發生錯誤: {str(e)}",
"data": None
}
def batch_check_interactions(self, chinese_medicines: List[str], western_medicines: List[str]) -> Dict[str, Any]:
"""批次檢查多個中藥和西藥的交互作用"""
try:
self.logger.info(f"批次檢查交互作用: {len(chinese_medicines)}個中藥 x {len(western_medicines)}個西藥")
# 驗證輸入參數
if not chinese_medicines:
return {
"success": False,
"error": "請提供至少一個中藥名稱",
"data": None
}
if not western_medicines:
return {
"success": False,
"error": "請提供至少一個西藥名稱",
"data": None
}
# 清理和驗證藥物名稱
chinese_medicines = [med.strip() for med in chinese_medicines if med.strip()]
western_medicines = [med.strip() for med in western_medicines if med.strip()]
if not chinese_medicines or not western_medicines:
return {
"success": False,
"error": "請提供有效的藥物名稱列表",
"data": None
}
# 執行批次檢查
result = self.db.batch_check_interactions(chinese_medicines, western_medicines)
# 轉換為字典格式
interactions_data = []
for interaction in result.interactions_found:
interactions_data.append({
"id": interaction.id,
"chinese_medicine": {
"name": interaction.chinese_name,
"id": interaction.chinese_medicine_id
},
"western_medicine": {
"name": interaction.western_name,
"id": interaction.western_medicine_id
},
"interaction": {
"type": interaction.interaction_type,
"severity": interaction.severity,
"mechanism": interaction.mechanism,
"clinical_effects": interaction.clinical_effects,
"recommendation": interaction.recommendation,
"evidence_level": interaction.evidence_level,
"references": interaction.ref_sources
}
})
# 按嚴重程度排序
severity_order = {"致命": 1, "重度": 2, "中度": 3, "輕度": 4}
interactions_data.sort(key=lambda x: severity_order.get(x["interaction"]["severity"], 5))
# 生成安全建議
safety_recommendations = []
if result.high_risk_pairs > 0:
safety_recommendations.append(f"⚠️ 發現 {result.high_risk_pairs} 個高風險交互作用組合，請避免同時使用")
safety_recommendations.append("建議諮詢醫師或藥師的專業建議")
if len(result.interactions_found) > 0:
safety_recommendations.append(f"總共發現 {len(result.interactions_found)} 個交互作用，需要特別注意用藥安全")
else:
safety_recommendations.append("目前未發現已知交互作用，但仍需謹慎使用並監測身體反應")
return {
"success": True,
"message": f"完成 {result.total_checked_pairs} 個藥物組合的檢查",
"data": {
"summary": {
"total_checked_pairs": result.total_checked_pairs,
"interactions_found": len(result.interactions_found),
"high_risk_pairs": result.high_risk_pairs,
"safety_level": "高風險" if result.high_risk_pairs > 0 else "低風險"
},
"chinese_medicines": chinese_medicines,
"western_medicines": western_medicines,
"interactions": interactions_data,
"warnings": result.warnings,
"safety_recommendations": safety_recommendations
}
}
except Exception as e:
self.logger.error(f"批次檢查交互作用時發生錯誤: {e}")
self.logger.error(traceback.format_exc())
return {
"success": False,
"error": f"批次檢查過程中發生錯誤: {str(e)}",
"data": None
}
def get_interaction_summary(self, medicine_name: str, medicine_type: str) -> Dict[str, Any]:
"""獲取藥物交互作用摘要"""
try:
self.logger.info(f"獲取交互作用摘要: {medicine_name} ({medicine_type})")
# 驗證輸入參數
if not medicine_name or not medicine_name.strip():
return {
"success": False,
"error": "請提供藥物名稱",
"data": None
}
if medicine_type not in ['chinese', 'western']:
return {
"success": False,
"error": "藥物類型必須是 'chinese' 或 'western'",
"data": None
}
medicine_name = medicine_name.strip()
medicine_type = medicine_type.strip()
# 獲取摘要資訊
summary = self.db.get_interaction_summary(medicine_name, medicine_type)
if not summary:
return {
"success": False,
"error": f"未找到藥物: {medicine_name}",
"data": None
}
# 轉換為字典格式
high_risk_interactions = []
for interaction in summary.high_risk_interactions:
high_risk_interactions.append({
"id": interaction.id,
"chinese_medicine": {
"name": interaction.chinese_name,
"id": interaction.chinese_medicine_id
},
"western_medicine": {
"name": interaction.western_name,
"id": interaction.western_medicine_id
},
"interaction": {
"type": interaction.interaction_type,
"severity": interaction.severity,
"mechanism": interaction.mechanism,
"clinical_effects": interaction.clinical_effects,
"recommendation": interaction.recommendation,
"evidence_level": interaction.evidence_level,
"references": interaction.ref_sources
}
})
# 生成風險評估
risk_assessment = "低風險"
if summary.severity_distribution.get('致命', 0) > 0:
risk_assessment = "極高風險"
elif summary.severity_distribution.get('重度', 0) > 0:
risk_assessment = "高風險"
elif summary.severity_distribution.get('中度', 0) > 0:
risk_assessment = "中風險"
return {
"success": True,
"message": f"獲取 {medicine_name} 的交互作用摘要",
"data": {
"medicine": {
"name": summary.medicine_name,
"type": summary.medicine_type,
"total_interactions": summary.total_interactions
},
"risk_assessment": {
"overall_risk_level": risk_assessment,
"severity_distribution": summary.severity_distribution,
"total_interactions": summary.total_interactions
},
"high_risk_interactions": high_risk_interactions,
"recommendations": summary.recommendations,
"usage_alert": f"此藥物目前記錄有 {summary.total_interactions} 個交互作用，請謹慎使用"
}
}
except Exception as e:
self.logger.error(f"獲取交互作用摘要時發生錯誤: {e}")
self.logger.error(traceback.format_exc())
return {
"success": False,
"error": f"獲取摘要過程中發生錯誤: {str(e)}",
"data": None
}