""" Discussion Guide JSON Schema Validation Provides schema validation and utilities for structured discussion guides. """ from typing import Dict, List, Any, Optional, Union from dataclasses import dataclass import json @dataclass class DiscussionGuideActivity: """Represents an activity within a discussion guide section.""" id: str type: str # moderator_statement, open_question, probe_question, activity, etc. content: str time_limit: Optional[int] = None metadata: Optional[Dict[str, Any]] = None @dataclass class DiscussionGuideQuestion: """Represents a question within a discussion guide section.""" id: str type: str # open_question, probe_question, follow_up, etc. content: str time_limit: Optional[int] = None probes: Optional[List[str]] = None metadata: Optional[Dict[str, Any]] = None @dataclass class DiscussionGuideSubsection: """Represents a subsection within a main discussion section.""" id: str title: str duration: int questions: List[DiscussionGuideQuestion] activities: Optional[List[DiscussionGuideActivity]] = None metadata: Optional[Dict[str, Any]] = None @dataclass class DiscussionGuideSection: """Represents a main section of a discussion guide.""" id: str title: str duration: int type: str # introduction, warmup, main_content, conclusion content: Optional[str] = None questions: Optional[List[DiscussionGuideQuestion]] = None activities: Optional[List[DiscussionGuideActivity]] = None subsections: Optional[List[DiscussionGuideSubsection]] = None metadata: Optional[Dict[str, Any]] = None @dataclass class StructuredDiscussionGuide: """Represents a complete structured discussion guide.""" title: str total_duration: int sections: List[DiscussionGuideSection] metadata: Optional[Dict[str, Any]] = None class DiscussionGuideValidator: """Validates and processes discussion guide JSON structures.""" @staticmethod def create_visual_asset_metadata(filename: str, display_reference: str) -> Dict[str, Any]: """ Create visual asset metadata for questions/activities. Args: filename: The system filename (e.g., 'fg-123-abc.jpg') display_reference: User-friendly reference (e.g., 'Asset 1' or custom name) Returns: Visual asset metadata dictionary """ return { "visual_asset": { "filename": filename, "display_reference": display_reference } } @staticmethod def generate_display_reference(assets: List[Dict[str, Any]], asset_index: int) -> str: """ Generate a display reference for an asset based on user assignment or default numbering. Args: assets: List of asset metadata objects asset_index: Index of the current asset Returns: Display reference string """ asset = assets[asset_index] # Use user-assigned name if available, otherwise use numbered reference if asset.get("user_assigned_name"): return asset["user_assigned_name"] else: return f"Asset {asset_index + 1}" @staticmethod def validate_json_structure(guide_json: Dict[str, Any]) -> tuple[bool, List[str]]: """ Validate a discussion guide JSON structure. Args: guide_json: The JSON structure to validate Returns: Tuple of (is_valid, list_of_errors) """ errors = [] # Check required top-level fields required_fields = ['title', 'total_duration', 'sections'] for field in required_fields: if field not in guide_json: errors.append(f"Missing required field: {field}") if 'sections' in guide_json: if not isinstance(guide_json['sections'], list): errors.append("'sections' must be a list") elif len(guide_json['sections']) == 0: errors.append("'sections' cannot be empty") else: # Validate each section for i, section in enumerate(guide_json['sections']): section_errors = DiscussionGuideValidator._validate_section(section, i) errors.extend(section_errors) # Validate total duration matches sum of sections if 'sections' in guide_json and 'total_duration' in guide_json: try: total_section_duration = sum(section.get('duration', 0) for section in guide_json['sections']) if abs(total_section_duration - guide_json['total_duration']) > 5: # Allow 5 minute tolerance errors.append(f"Total duration ({guide_json['total_duration']}) doesn't match sum of sections ({total_section_duration})") except (TypeError, ValueError): errors.append("Invalid duration values in sections") return len(errors) == 0, errors @staticmethod def _validate_section(section: Dict[str, Any], index: int) -> List[str]: """Validate a single section.""" errors = [] section_prefix = f"Section {index + 1}" # Check required section fields required_fields = ['id', 'title', 'duration', 'type'] for field in required_fields: if field not in section: errors.append(f"{section_prefix}: Missing required field '{field}'") # Validate section type if 'type' in section: valid_types = ['introduction', 'warmup', 'main_content', 'conclusion', 'activity', 'break'] if section['type'] not in valid_types: errors.append(f"{section_prefix}: Invalid section type '{section['type']}'") # Validate duration if 'duration' in section: try: duration = int(section['duration']) if duration <= 0: errors.append(f"{section_prefix}: Duration must be positive") except (TypeError, ValueError): errors.append(f"{section_prefix}: Duration must be a number") # Validate questions if present if 'questions' in section and section['questions']: if not isinstance(section['questions'], list): errors.append(f"{section_prefix}: 'questions' must be a list") else: for j, question in enumerate(section['questions']): question_errors = DiscussionGuideValidator._validate_question(question, index, j) errors.extend(question_errors) # Validate activities if present if 'activities' in section and section['activities']: if not isinstance(section['activities'], list): errors.append(f"{section_prefix}: 'activities' must be a list") else: for j, activity in enumerate(section['activities']): activity_errors = DiscussionGuideValidator._validate_activity(activity, index, j) errors.extend(activity_errors) # Validate subsections if present if 'subsections' in section and section['subsections']: if not isinstance(section['subsections'], list): errors.append(f"{section_prefix}: 'subsections' must be a list") else: for j, subsection in enumerate(section['subsections']): subsection_errors = DiscussionGuideValidator._validate_subsection(subsection, index, j) errors.extend(subsection_errors) return errors @staticmethod def _validate_question(question: Dict[str, Any], section_index: int, question_index: int) -> List[str]: """Validate a single question.""" errors = [] question_prefix = f"Section {section_index + 1}, Question {question_index + 1}" # Check required question fields required_fields = ['id', 'type', 'content'] for field in required_fields: if field not in question: errors.append(f"{question_prefix}: Missing required field '{field}'") # Validate question type (accept any string) if 'type' in question and not isinstance(question['type'], str): errors.append(f"{question_prefix}: Question type must be a string") # Validate content if 'content' in question and not isinstance(question['content'], str): errors.append(f"{question_prefix}: 'content' must be a string") return errors @staticmethod def _validate_activity(activity: Dict[str, Any], section_index: int, activity_index: int) -> List[str]: """Validate a single activity.""" errors = [] activity_prefix = f"Section {section_index + 1}, Activity {activity_index + 1}" # Check required activity fields required_fields = ['id', 'type', 'content'] for field in required_fields: if field not in activity: errors.append(f"{activity_prefix}: Missing required field '{field}'") # Validate activity type (accept any string) if 'type' in activity and not isinstance(activity['type'], str): errors.append(f"{activity_prefix}: Activity type must be a string") return errors @staticmethod def _validate_subsection(subsection: Dict[str, Any], section_index: int, subsection_index: int) -> List[str]: """Validate a single subsection.""" errors = [] subsection_prefix = f"Section {section_index + 1}, Subsection {subsection_index + 1}" # Check required subsection fields required_fields = ['id', 'title', 'duration', 'questions'] for field in required_fields: if field not in subsection: errors.append(f"{subsection_prefix}: Missing required field '{field}'") # Validate questions if 'questions' in subsection and subsection['questions']: if not isinstance(subsection['questions'], list): errors.append(f"{subsection_prefix}: 'questions' must be a list") else: for j, question in enumerate(subsection['questions']): question_errors = DiscussionGuideValidator._validate_question(question, section_index, j) errors.extend(question_errors) return errors @staticmethod def create_fallback_structure(title: str, duration: int, content: str) -> Dict[str, Any]: """ Create a fallback discussion guide structure if JSON generation fails. Args: title: The focus group title duration: Total duration in minutes content: Raw content to structure Returns: A basic valid discussion guide structure """ # Calculate section durations intro_duration = max(5, int(duration * 0.1)) warmup_duration = max(5, int(duration * 0.15)) main_duration = max(20, int(duration * 0.6)) conclusion_duration = max(5, int(duration * 0.15)) return { "title": title, "total_duration": duration, "sections": [ { "id": "introduction", "title": "Introduction", "duration": intro_duration, "type": "introduction", "activities": [ { "id": "welcome", "type": "moderator_statement", "content": f"Welcome everyone to our focus group on {title}. Let's begin by introducing ourselves and the purpose of today's discussion." } ] }, { "id": "warmup", "title": "Warm-up Questions", "duration": warmup_duration, "type": "warmup", "questions": [ { "id": "intro_question", "type": "open_question", "content": "Let's start with brief introductions. Please share your name and one thing you're excited about today." } ] }, { "id": "main_discussion", "title": "Main Discussion", "duration": main_duration, "type": "main_content", "questions": [ { "id": "main_question", "type": "open_question", "content": "What are your initial thoughts on the topic we're discussing today?" } ] }, { "id": "conclusion", "title": "Conclusion", "duration": conclusion_duration, "type": "conclusion", "questions": [ { "id": "final_thoughts", "type": "open_question", "content": "Before we wrap up, are there any final thoughts or comments you'd like to share?" } ] } ] } @staticmethod def parse_from_json(json_string: str) -> tuple[Optional[StructuredDiscussionGuide], List[str]]: """ Parse a JSON string into a StructuredDiscussionGuide object. Args: json_string: The JSON string to parse Returns: Tuple of (parsed_guide, list_of_errors) """ try: guide_json = json.loads(json_string) except json.JSONDecodeError as e: return None, [f"Invalid JSON: {str(e)}"] # Validate structure is_valid, errors = DiscussionGuideValidator.validate_json_structure(guide_json) if not is_valid: return None, errors try: # Convert to structured objects sections = [] for section_data in guide_json['sections']: section = DiscussionGuideValidator._parse_section(section_data) sections.append(section) guide = StructuredDiscussionGuide( title=guide_json['title'], total_duration=guide_json['total_duration'], sections=sections, metadata=guide_json.get('metadata') ) return guide, [] except Exception as e: return None, [f"Error parsing structure: {str(e)}"] @staticmethod def _parse_section(section_data: Dict[str, Any]) -> DiscussionGuideSection: """Parse a section from JSON data.""" questions = [] if section_data.get('questions'): for q_data in section_data['questions']: question = DiscussionGuideQuestion( id=q_data['id'], type=q_data['type'], content=q_data['content'], time_limit=q_data.get('time_limit'), probes=q_data.get('probes'), metadata=q_data.get('metadata') ) questions.append(question) activities = [] if section_data.get('activities'): for a_data in section_data['activities']: activity = DiscussionGuideActivity( id=a_data['id'], type=a_data['type'], content=a_data['content'], time_limit=a_data.get('time_limit'), metadata=a_data.get('metadata') ) activities.append(activity) subsections = [] if section_data.get('subsections'): for s_data in section_data['subsections']: subsection_questions = [] for q_data in s_data.get('questions', []): question = DiscussionGuideQuestion( id=q_data['id'], type=q_data['type'], content=q_data['content'], time_limit=q_data.get('time_limit'), probes=q_data.get('probes'), metadata=q_data.get('metadata') ) subsection_questions.append(question) subsection_activities = [] if s_data.get('activities'): for a_data in s_data['activities']: activity = DiscussionGuideActivity( id=a_data['id'], type=a_data['type'], content=a_data['content'], time_limit=a_data.get('time_limit'), metadata=a_data.get('metadata') ) subsection_activities.append(activity) subsection = DiscussionGuideSubsection( id=s_data['id'], title=s_data['title'], duration=s_data['duration'], questions=subsection_questions, activities=subsection_activities if subsection_activities else None, metadata=s_data.get('metadata') ) subsections.append(subsection) return DiscussionGuideSection( id=section_data['id'], title=section_data['title'], duration=section_data['duration'], type=section_data['type'], content=section_data.get('content'), questions=questions if questions else None, activities=activities if activities else None, subsections=subsections if subsections else None, metadata=section_data.get('metadata') )