#!/usr/bin/env python3 """ Conversation Logger for Medical Assistant. Logs conversations with spiritual classification indicators for analysis. """ import json import os from datetime import datetime from typing import Dict, List, Any, Optional from dataclasses import dataclass, asdict from src.core.spiritual_state import SpiritualState, SpiritualAssessment @dataclass class ConversationEntry: """Single conversation entry with classification data.""" timestamp: str user_message: str assistant_response: str spiritual_classification: str # GREEN, YELLOW, RED classification_confidence: float classification_indicators: List[str] classification_reasoning: str session_id: str message_index: int @dataclass class ConversationSession: """Complete conversation session.""" session_id: str start_time: str end_time: Optional[str] patient_name: str total_messages: int entries: List[ConversationEntry] session_summary: Dict[str, Any] class ConversationLogger: """Logger for conversation sessions with spiritual classification data.""" def __init__(self, session_id: str = None, patient_name: str = "Anonymous"): """Initialize conversation logger.""" self.session_id = session_id or self._generate_session_id() self.patient_name = patient_name self.start_time = datetime.now().isoformat() self.entries: List[ConversationEntry] = [] self.message_counter = 0 # Create logs directory if it doesn't exist self.logs_dir = "conversation_logs" os.makedirs(self.logs_dir, exist_ok=True) def _generate_session_id(self) -> str: """Generate unique session ID.""" return f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}" def log_exchange( self, user_message: str, assistant_response: str, assessment: SpiritualAssessment ) -> None: """ Log a conversation exchange with spiritual classification. Args: user_message: User's message assistant_response: Assistant's response assessment: Spiritual assessment of the user message """ self.message_counter += 1 entry = ConversationEntry( timestamp=datetime.now().isoformat(), user_message=user_message, assistant_response=assistant_response, spiritual_classification=assessment.state.value.upper(), classification_confidence=assessment.confidence, classification_indicators=assessment.indicators, classification_reasoning=assessment.reasoning, session_id=self.session_id, message_index=self.message_counter ) self.entries.append(entry) # Auto-save after each entry self._save_session() def get_classification_indicator(self, state: SpiritualState) -> str: """Get colored emoji indicator for spiritual state.""" indicators = { SpiritualState.GREEN: "🟢", SpiritualState.YELLOW: "🟡", SpiritualState.RED: "🔴" } return indicators.get(state, "⚪") def get_classification_text(self, assessment: SpiritualAssessment) -> str: """Get formatted classification text for display.""" indicator = self.get_classification_indicator(assessment.state) confidence_percent = int(assessment.confidence * 100) classification_text = f"{indicator} **{assessment.state.value.upper()}** ({confidence_percent}%)" if assessment.indicators: indicators_text = ", ".join(assessment.indicators[:3]) # Show max 3 indicators if len(assessment.indicators) > 3: indicators_text += f" +{len(assessment.indicators) - 3} more" classification_text += f"\n*Indicators: {indicators_text}*" # Add reasoning in italics if assessment.reasoning: classification_text += f"\n*{assessment.reasoning}*" return classification_text def _save_session(self) -> None: """Save current session to JSON file.""" session = ConversationSession( session_id=self.session_id, start_time=self.start_time, end_time=None, # Will be set when session ends patient_name=self.patient_name, total_messages=self.message_counter, entries=self.entries, session_summary=self._generate_session_summary() ) filename = f"{self.session_id}.json" filepath = os.path.join(self.logs_dir, filename) try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(asdict(session), f, ensure_ascii=False, indent=2) except Exception as e: print(f"Error saving conversation log: {e}") def _generate_session_summary(self) -> Dict[str, Any]: """Generate summary statistics for the session.""" if not self.entries: return {} # Count classifications green_count = sum(1 for e in self.entries if e.spiritual_classification == "GREEN") yellow_count = sum(1 for e in self.entries if e.spiritual_classification == "YELLOW") red_count = sum(1 for e in self.entries if e.spiritual_classification == "RED") # Calculate average confidence avg_confidence = sum(e.classification_confidence for e in self.entries) / len(self.entries) # Collect all indicators all_indicators = [] for entry in self.entries: all_indicators.extend(entry.classification_indicators) # Count unique indicators indicator_counts = {} for indicator in all_indicators: indicator_counts[indicator] = indicator_counts.get(indicator, 0) + 1 return { "total_exchanges": len(self.entries), "classification_counts": { "green": green_count, "yellow": yellow_count, "red": red_count }, "classification_percentages": { "green": round(green_count / len(self.entries) * 100, 1), "yellow": round(yellow_count / len(self.entries) * 100, 1), "red": round(red_count / len(self.entries) * 100, 1) }, "average_confidence": round(avg_confidence, 3), "top_indicators": dict(sorted(indicator_counts.items(), key=lambda x: x[1], reverse=True)[:5]), "session_duration_minutes": self._calculate_session_duration() } def _calculate_session_duration(self) -> float: """Calculate session duration in minutes.""" if not self.entries: return 0.0 start = datetime.fromisoformat(self.start_time) last_entry = datetime.fromisoformat(self.entries[-1].timestamp) duration = (last_entry - start).total_seconds() / 60 return round(duration, 1) def end_session(self) -> str: """End the session and return final log file path.""" # Update end time in the last save if self.entries: self.entries[-1].timestamp = datetime.now().isoformat() self._save_session() filename = f"{self.session_id}.json" return os.path.join(self.logs_dir, filename) def get_session_summary(self) -> Dict[str, Any]: """Get current session summary.""" return self._generate_session_summary() def export_csv(self) -> str: """Export conversation to CSV format.""" import csv csv_filename = f"{self.session_id}.csv" csv_filepath = os.path.join(self.logs_dir, csv_filename) try: with open(csv_filepath, 'w', newline='', encoding='utf-8') as csvfile: fieldnames = [ 'timestamp', 'message_index', 'user_message', 'assistant_response', 'spiritual_classification', 'classification_confidence', 'classification_indicators', 'classification_reasoning' ] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for entry in self.entries: writer.writerow({ 'timestamp': entry.timestamp, 'message_index': entry.message_index, 'user_message': entry.user_message, 'assistant_response': entry.assistant_response, 'spiritual_classification': entry.spiritual_classification, 'classification_confidence': entry.classification_confidence, 'classification_indicators': '; '.join(entry.classification_indicators), 'classification_reasoning': entry.classification_reasoning }) return csv_filepath except Exception as e: print(f"Error exporting to CSV: {e}") return ""