Integrate TxAgent into CPS API: add TxAgent files, update requirements, fix imports and database collections

api/routes/txagent.py

@@ -415,4 +415,267 @@ async def get_chats(current_user: dict = Depends(get_current_user)):
         logger.error(f"Error getting chats: {e}")
         raise HTTPException(status_code=500, detail="Failed to get chats")
 
+@router.get("/patients/{patient_id}/analysis-reports/pdf")
+async def get_patient_analysis_reports_pdf(
+    patient_id: str = Path(...),
+    current_user: dict = Depends(get_current_user)
+):
+    """Generate PDF analysis reports for a specific patient"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can generate PDF reports")
+        
+        logger.info(f"Generating PDF analysis reports for patient {patient_id} by {current_user['email']}")
+        
+        # Import database collections
+        from db.mongo import db
+        analysis_collection = db.analysis_results
+        
+        # Find analysis results for the patient
+        analysis_results = await analysis_collection.find({"patient_id": patient_id}).to_list(length=None)
+        
+        if not analysis_results:
+            raise HTTPException(status_code=404, detail="No analysis results found for this patient")
+        
+        # Create a simple PDF report
+        from reportlab.lib.pagesizes import letter
+        from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
+        from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
+        from reportlab.lib.units import inch
+        from reportlab.lib import colors
+        import io
+        
+        # Create PDF buffer
+        buffer = io.BytesIO()
+        doc = SimpleDocTemplate(buffer, pagesize=letter)
+        styles = getSampleStyleSheet()
+        story = []
+        
+        # Title
+        title_style = ParagraphStyle(
+            'CustomTitle',
+            parent=styles['Heading1'],
+            fontSize=16,
+            spaceAfter=30,
+            alignment=1  # Center alignment
+        )
+        story.append(Paragraph("Patient Analysis Report", title_style))
+        story.append(Spacer(1, 12))
+        
+        # Patient Information
+        story.append(Paragraph("Patient Information", styles['Heading2']))
+        story.append(Spacer(1, 12))
+        
+        # Get patient info from first analysis result
+        first_result = analysis_results[0]
+        patient_info = [
+            ["Patient ID:", patient_id],
+            ["Analysis Date:", first_result.get('timestamp', 'N/A')],
+        ]
+        
+        patient_table = Table(patient_info, colWidths=[2*inch, 4*inch])
+        patient_table.setStyle(TableStyle([
+            ('BACKGROUND', (0, 0), (0, -1), colors.grey),
+            ('TEXTCOLOR', (0, 0), (0, -1), colors.whitesmoke),
+            ('ALIGN', (0, 0), (-1, -1), 'LEFT'),
+            ('FONTNAME', (0, 0), (-1, -1), 'Helvetica-Bold'),
+            ('FONTSIZE', (0, 0), (-1, -1), 10),
+            ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
+            ('BACKGROUND', (1, 0), (1, -1), colors.beige),
+            ('GRID', (0, 0), (-1, -1), 1, colors.black)
+        ]))
+        story.append(patient_table)
+        story.append(Spacer(1, 20))
+        
+        # Analysis Results
+        story.append(Paragraph("Analysis Results", styles['Heading2']))
+        story.append(Spacer(1, 12))
+        
+        for i, result in enumerate(analysis_results):
+            # Risk Assessment
+            suicide_risk = result.get('suicide_risk', {})
+            risk_level = suicide_risk.get('level', 'none') if isinstance(suicide_risk, dict) else 'none'
+            risk_score = suicide_risk.get('score', 0.0) if isinstance(suicide_risk, dict) else 0.0
+            risk_factors = suicide_risk.get('factors', []) if isinstance(suicide_risk, dict) else []
+            
+            story.append(Paragraph(f"Analysis #{i+1}", styles['Heading3']))
+            story.append(Spacer(1, 6))
+            
+            analysis_data = [
+                ["Risk Level:", risk_level.upper()],
+                ["Risk Score:", f"{risk_score:.2f}"],
+                ["Risk Factors:", ", ".join(risk_factors) if risk_factors else "None identified"],
+                ["Analysis Date:", result.get('timestamp', 'N/A')],
+            ]
+            
+            analysis_table = Table(analysis_data, colWidths=[2*inch, 4*inch])
+            analysis_table.setStyle(TableStyle([
+                ('BACKGROUND', (0, 0), (0, -1), colors.lightblue),
+                ('TEXTCOLOR', (0, 0), (0, -1), colors.black),
+                ('ALIGN', (0, 0), (-1, -1), 'LEFT'),
+                ('FONTNAME', (0, 0), (-1, -1), 'Helvetica'),
+                ('FONTSIZE', (0, 0), (-1, -1), 9),
+                ('BOTTOMPADDING', (0, 0), (-1, 0), 6),
+                ('BACKGROUND', (1, 0), (1, -1), colors.white),
+                ('GRID', (0, 0), (-1, -1), 1, colors.black)
+            ]))
+            story.append(analysis_table)
+            story.append(Spacer(1, 12))
+            
+            # Summary if available
+            if result.get('summary'):
+                story.append(Paragraph("Summary:", styles['Heading4']))
+                story.append(Paragraph(result['summary'], styles['Normal']))
+                story.append(Spacer(1, 12))
+        
+        # Build PDF
+        doc.build(story)
+        buffer.seek(0)
+        
+        return StreamingResponse(
+            buffer,
+            media_type="application/pdf",
+            headers={"Content-Disposition": f"attachment; filename=patient_{patient_id}_analysis_reports.pdf"}
+        )
+        
+    except Exception as e:
+        logger.error(f"Error generating PDF report for patient {patient_id}: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to generate PDF report: {str(e)}")
+
+@router.get("/patients/analysis-reports/all/pdf")
+async def get_all_patients_analysis_reports_pdf(
+    current_user: dict = Depends(get_current_user)
+):
+    """Generate PDF analysis reports for all patients"""
+    try:
+        if not any(role in current_user.get('roles', []) for role in ['doctor', 'admin']):
+            raise HTTPException(status_code=403, detail="Only doctors and admins can generate PDF reports")
+        
+        logger.info(f"Generating PDF analysis reports for all patients by {current_user['email']}")
+        
+        # Import database collections
+        from db.mongo import db
+        analysis_collection = db.analysis_results
+        
+        # Find all analysis results
+        analysis_results = await analysis_collection.find({}).to_list(length=None)
+        
+        if not analysis_results:
+            raise HTTPException(status_code=404, detail="No analysis results found")
+        
+        # Create a simple PDF report
+        from reportlab.lib.pagesizes import letter
+        from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
+        from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
+        from reportlab.lib.units import inch
+        from reportlab.lib import colors
+        import io
+        
+        # Create PDF buffer
+        buffer = io.BytesIO()
+        doc = SimpleDocTemplate(buffer, pagesize=letter)
+        styles = getSampleStyleSheet()
+        story = []
+        
+        # Title
+        title_style = ParagraphStyle(
+            'CustomTitle',
+            parent=styles['Heading1'],
+            fontSize=16,
+            spaceAfter=30,
+            alignment=1  # Center alignment
+        )
+        story.append(Paragraph("All Patients Analysis Reports", title_style))
+        story.append(Spacer(1, 12))
+        
+        # Summary
+        story.append(Paragraph("Summary", styles['Heading2']))
+        story.append(Spacer(1, 12))
+        
+        summary_data = [
+            ["Total Analysis Reports:", str(len(analysis_results))],
+            ["Generated Date:", datetime.now().strftime("%Y-%m-%d %H:%M:%S")],
+            ["Generated By:", current_user['email']],
+        ]
+        
+        summary_table = Table(summary_data, colWidths=[2*inch, 4*inch])
+        summary_table.setStyle(TableStyle([
+            ('BACKGROUND', (0, 0), (0, -1), colors.grey),
+            ('TEXTCOLOR', (0, 0), (0, -1), colors.whitesmoke),
+            ('ALIGN', (0, 0), (-1, -1), 'LEFT'),
+            ('FONTNAME', (0, 0), (-1, -1), 'Helvetica-Bold'),
+            ('FONTSIZE', (0, 0), (-1, -1), 10),
+            ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
+            ('BACKGROUND', (1, 0), (1, -1), colors.beige),
+            ('GRID', (0, 0), (-1, -1), 1, colors.black)
+        ]))
+        story.append(summary_table)
+        story.append(Spacer(1, 20))
+        
+        # Group results by patient
+        patient_results = {}
+        for result in analysis_results:
+            patient_id = result.get('patient_id', 'unknown')
+            if patient_id not in patient_results:
+                patient_results[patient_id] = []
+            patient_results[patient_id].append(result)
+        
+        # Patient Reports
+        for patient_id, results in patient_results.items():
+            story.append(Paragraph(f"Patient: {patient_id}", styles['Heading2']))
+            story.append(Spacer(1, 12))
+            
+            for i, result in enumerate(results):
+                # Risk Assessment
+                suicide_risk = result.get('suicide_risk', {})
+                risk_level = suicide_risk.get('level', 'none') if isinstance(suicide_risk, dict) else 'none'
+                risk_score = suicide_risk.get('score', 0.0) if isinstance(suicide_risk, dict) else 0.0
+                risk_factors = suicide_risk.get('factors', []) if isinstance(suicide_risk, dict) else []
+                
+                story.append(Paragraph(f"Analysis #{i+1}", styles['Heading3']))
+                story.append(Spacer(1, 6))
+                
+                analysis_data = [
+                    ["Risk Level:", risk_level.upper()],
+                    ["Risk Score:", f"{risk_score:.2f}"],
+                    ["Risk Factors:", ", ".join(risk_factors) if risk_factors else "None identified"],
+                    ["Analysis Date:", result.get('timestamp', 'N/A')],
+                ]
+                
+                analysis_table = Table(analysis_data, colWidths=[2*inch, 4*inch])
+                analysis_table.setStyle(TableStyle([
+                    ('BACKGROUND', (0, 0), (0, -1), colors.lightblue),
+                    ('TEXTCOLOR', (0, 0), (0, -1), colors.black),
+                    ('ALIGN', (0, 0), (-1, -1), 'LEFT'),
+                    ('FONTNAME', (0, 0), (-1, -1), 'Helvetica'),
+                    ('FONTSIZE', (0, 0), (-1, -1), 9),
+                    ('BOTTOMPADDING', (0, 0), (-1, 0), 6),
+                    ('BACKGROUND', (1, 0), (1, -1), colors.white),
+                    ('GRID', (0, 0), (-1, -1), 1, colors.black)
+                ]))
+                story.append(analysis_table)
+                story.append(Spacer(1, 12))
+                
+                # Summary if available
+                if result.get('summary'):
+                    story.append(Paragraph("Summary:", styles['Heading4']))
+                    story.append(Paragraph(result['summary'], styles['Normal']))
+                    story.append(Spacer(1, 12))
+            
+            story.append(Spacer(1, 20))
+        
+        # Build PDF
+        doc.build(story)
+        buffer.seek(0)
+        
+        return StreamingResponse(
+            buffer,
+            media_type="application/pdf",
+            headers={"Content-Disposition": f"attachment; filename=all_patients_analysis_reports_{datetime.now().strftime('%Y%m%d')}.pdf"}
+        )
+        
+    except Exception as e:
+        logger.error(f"Error generating PDF report for all patients: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to generate PDF report: {str(e)}")
+
 

requirements.txt

@@ -30,4 +30,5 @@ fitz
 python-docx
 pyfcm
 httpx
-jwt
+jwt
+reportlab>=3.6.0