""" 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)}" )