Spaces:
Running
Running
File size: 5,316 Bytes
2a8faae |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 |
"""
Side Effects Reporting Router
"""
import uuid
import time
from datetime import datetime
from fastapi import APIRouter, HTTPException
import sys
import os
# Add backend and src directories to path for imports
backend_path = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
src_path = os.path.join(backend_path, 'src')
sys.path.append(backend_path)
sys.path.append(src_path)
from api.models import SideEffectReport
from tools import side_effect_recording_tool
from github_storage import get_github_storage
router = APIRouter(prefix="/side-effects", tags=["side-effects"])
@router.post("/report")
async def report_side_effect(report: SideEffectReport):
"""
Submit a side effect report for pharmacovigilance
"""
try:
# Prepare the report text for the side effect tool
report_text = f"Drug: {report.drug_name}, Side effects: {report.side_effects}"
if report.patient_age:
report_text += f", Patient age: {report.patient_age}"
if report.patient_gender:
report_text += f", Patient gender: {report.patient_gender}"
if report.dosage:
report_text += f", Dosage: {report.dosage}"
if report.duration:
report_text += f", Duration: {report.duration}"
if report.severity:
report_text += f", Severity: {report.severity}"
if report.outcome:
report_text += f", Outcome: {report.outcome}"
if report.additional_details:
report_text += f", Additional details: {report.additional_details}"
# Process through the existing side effect recording tool
result = side_effect_recording_tool(report_text)
return {"result": result}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Error processing side effect report: {str(e)}"
)
@router.get("/reports/summary")
async def get_reports_summary():
"""
Get summary of side effect reports from GitHub repository
"""
try:
import pandas as pd
import os
# Try to get reports from GitHub first
github_storage = get_github_storage()
reports = github_storage.get_side_effects_reports()
if not reports:
# Fallback to local file if GitHub fails
csv_path = "side_effects_reports.csv"
if not os.path.exists(csv_path):
return {
"total_reports": 0,
"message": "No side effect reports found"
}
df = pd.read_csv(csv_path)
else:
# Convert GitHub reports to DataFrame
df = pd.DataFrame(reports)
if df.empty:
return {
"total_reports": 0,
"message": "No side effect reports found"
}
summary = {
"total_reports": len(df),
"unique_drugs": df['drug_name'].nunique() if 'drug_name' in df.columns else 0,
"recent_reports": len(df[df['timestamp'] >= (datetime.now() - pd.Timedelta(days=30)).strftime('%Y-%m-%d')]) if 'timestamp' in df.columns else 0,
"most_reported_drugs": df['drug_name'].value_counts().head(5).to_dict() if 'drug_name' in df.columns else {}
}
return summary
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Error retrieving reports summary: {str(e)}"
)
@router.get("/drug/{drug_name}/reports")
async def get_drug_reports(drug_name: str):
"""
Get side effect reports for a specific drug from GitHub repository
"""
try:
# Try to get reports from GitHub first
github_storage = get_github_storage()
reports = github_storage.get_drug_reports(drug_name)
if reports:
return {
"drug_name": drug_name,
"total_reports": len(reports),
"reports": reports
}
# Fallback to local file if GitHub fails or no reports found
import pandas as pd
import os
csv_path = "side_effects_reports.csv"
if not os.path.exists(csv_path):
return {
"drug_name": drug_name,
"reports": [],
"message": "No reports found"
}
df = pd.read_csv(csv_path)
# Filter reports for the specific drug (case-insensitive)
drug_reports = df[df['drug_name'].str.lower() == drug_name.lower()]
if drug_reports.empty:
return {
"drug_name": drug_name,
"reports": [],
"message": f"No reports found for {drug_name}"
}
# Convert to list of dictionaries
local_reports = drug_reports.to_dict('records')
return {
"drug_name": drug_name,
"total_reports": len(local_reports),
"reports": local_reports
}
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Error retrieving drug reports: {str(e)}"
)
|