""" Export Utilities - Handle various export formats """ import json import csv import io from typing import Dict, List from datetime import datetime def save_json_file(data: Dict, prefix: str = "export") -> str: """ Save data to JSON file and return filepath. Args: data: Data to save prefix: Filename prefix Returns: Path to saved file """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{prefix}_{timestamp}.json" with open(filename, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) return filename def survey_to_csv(survey_data: Dict) -> str: """ Convert survey to CSV format (one row per question). Args: survey_data: Survey dictionary Returns: Path to CSV file """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"survey_{timestamp}.csv" with open(filename, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) # Write header writer.writerow(['Question ID', 'Question Text', 'Type', 'Options', 'Required', 'Help Text']) # Write questions for q in survey_data.get('questions', []): writer.writerow([ q.get('id', ''), q.get('question_text', ''), q.get('question_type', ''), '; '.join(q.get('options', [])) if q.get('options') else '', 'Yes' if q.get('required', False) else 'No', q.get('help_text', '') ]) return filename def responses_to_csv(responses: List[Dict], filename_prefix: str = "responses") -> str: """ Convert responses to CSV format. Args: responses: List of response dictionaries filename_prefix: Prefix for filename Returns: Path to CSV file """ if not responses: return None timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{filename_prefix}_{timestamp}.csv" # Get all unique keys from all responses all_keys = set() for response in responses: if isinstance(response, dict): all_keys.update(response.keys()) fieldnames = sorted(all_keys) with open(filename, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for response in responses: if isinstance(response, dict): writer.writerow(response) return filename def analysis_to_markdown_file(analysis_report: str, prefix: str = "analysis_report") -> str: """ Save analysis report to markdown file. Args: analysis_report: Markdown formatted report prefix: Filename prefix Returns: Path to markdown file """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{prefix}_{timestamp}.md" with open(filename, 'w', encoding='utf-8') as f: f.write(analysis_report) return filename def create_survey_package(survey_data: Dict) -> Dict[str, str]: """ Create a complete package of survey files (JSON, CSV, etc.). Args: survey_data: Survey dictionary Returns: Dictionary mapping format to filepath """ package = {} # Save JSON package['json'] = save_json_file(survey_data, "survey") # Save CSV package['csv'] = survey_to_csv(survey_data) return package def conversation_to_transcript(conversation_session) -> str: """ Export conversation session as readable text transcript. Args: conversation_session: ConversationSession object Returns: Path to transcript file """ transcript = conversation_session.get_transcript() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"conversation_transcript_{timestamp}.txt" with open(filename, 'w', encoding='utf-8') as f: f.write(transcript) return filename def conversation_to_json(conversation_session) -> str: """ Export conversation session as JSON. Args: conversation_session: ConversationSession object Returns: Path to JSON file """ return save_json_file(conversation_session.to_dict(), "conversation_session") def conversation_to_csv(conversation_session) -> str: """ Export conversation turns as CSV. Args: conversation_session: ConversationSession object Returns: Path to CSV file """ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"conversation_{timestamp}.csv" with open(filename, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) # Write header writer.writerow(['Turn', 'Speaker', 'Timestamp', 'Content', 'Node ID', 'Summary']) # Write turns for i, turn in enumerate(conversation_session.conversation_history, 1): speaker = "AI Moderator" if turn.role == "ai" else "Respondent" writer.writerow([ i, speaker, turn.timestamp, turn.content, turn.node_id or '', turn.summary or '' ]) return filename def flow_to_markdown(conversation_flow) -> str: """ Export conversation flow as markdown document. Args: conversation_flow: ConversationFlow object Returns: Path to markdown file """ lines = [] lines.append(f"# {conversation_flow.name}\n") lines.append(f"**Description:** {conversation_flow.description}\n") lines.append(f"**Created:** {conversation_flow.created_at}") lines.append(f"**Updated:** {conversation_flow.updated_at}\n") lines.append("\n## Conversation Flow\n") for i, node in enumerate(conversation_flow.nodes, 1): lines.append(f"### Step {i}: {node.type.capitalize()}\n") lines.append(f"**Content:** {node.content}\n") if node.next: lines.append(f"**Next Node:** {node.next}\n") if node.branches: lines.append(f"**Branches:** {len(node.branches)}\n") lines.append("") content = "\n".join(lines) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"conversation_flow_{timestamp}.md" with open(filename, 'w', encoding='utf-8') as f: f.write(content) return filename