Spaces:
Running
Running
| """ | |
| 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"]) | |
| 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)}" | |
| ) | |
| 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)}" | |
| ) | |
| 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)}" | |
| ) | |