Update app.py

app.py

@@ -243,10 +243,254 @@ class TTSDatasetCollector:
             logger.error(traceback.format_exc())
             return False, error_msg
 
-    # ... (Rest of the class remains unchanged)
+    def generate_filenames(self, dataset_name: str, speaker_id: str, sentence_text: str) -> Tuple[str, str]:
+        """Generate unique filenames for audio and text files"""
+        line_number = self.current_index + 1
+        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+
+        # Sanitize strings for filenames
+        def sanitize_filename(s):
+            return re.sub(r'[^a-zA-Z0-9_-]', '_', s)[:50]
+
+        dataset_name_safe = sanitize_filename(dataset_name)
+        speaker_id_safe = sanitize_filename(speaker_id)
+        sentence_excerpt = sanitize_filename(sentence_text[:20])
+        base_name = f"{dataset_name_safe}_{speaker_id_safe}_line{line_number}_{sentence_excerpt}_{timestamp}"
+        return f"{base_name}.wav", f"{base_name}.txt"
+
+    def save_recording(self, audio_file, speaker_id: str, dataset_name: str) -> Tuple[bool, str]:
+        """Save recording with enhanced error handling and logging"""
+        if not all([audio_file, speaker_id, dataset_name]):
+            missing = []
+            if not audio_file:
+                missing.append("audio recording")
+            if not speaker_id:
+                missing.append("speaker ID")
+            if not dataset_name:
+                missing.append("dataset name")
+            return False, f"Missing required information: {', '.join(missing)}"
+
+        # Check if sentences have been loaded
+        if not self.sentences:
+            return False, "No sentences have been loaded. Please load text before saving recordings."
+        if self.current_index >= len(self.sentences):
+            return False, "Current sentence index is out of range."
+
+        try:
+            # Validate inputs
+            if not speaker_id.strip().isalnum():
+                return False, "Speaker ID must contain only letters and numbers"
+            if not dataset_name.strip().isalnum():
+                return False, "Dataset name must contain only letters and numbers"
+
+            # Get current sentence text
+            sentence_text = self.sentences[self.current_index]
+
+            # Generate filenames
+            audio_name, text_name = self.generate_filenames(dataset_name, speaker_id, sentence_text)
+
+            # Create speaker directories
+            audio_dir = self.root_path / 'audio' / speaker_id
+            text_dir = self.root_path / 'transcriptions' / speaker_id
+            audio_dir.mkdir(parents=True, exist_ok=True)
+            text_dir.mkdir(parents=True, exist_ok=True)
+
+            # Save audio file
+            audio_path = audio_dir / audio_name
+
+            # Read the audio file using soundfile
+            audio_data, sampling_rate = sf.read(audio_file)
+
+            # Save audio file
+            sf.write(str(audio_path), audio_data, sampling_rate)
+
+            # Save transcription
+            text_path = text_dir / text_name
+            self.save_transcription(
+                text_path,
+                sentence_text,
+                {
+                    'speaker_id': speaker_id,
+                    'dataset_name': dataset_name,
+                    'timestamp': datetime.now().isoformat(),
+                    'audio_file': audio_name,
+                    'font_style': self.current_font
+                }
+            )
+
+            # Update metadata
+            self.update_metadata(speaker_id, dataset_name)
+
+            # Log success
+            self.log_operation(
+                f"Saved recording: Speaker={speaker_id}, Dataset={dataset_name}, "
+                f"Audio={audio_name}, Text={text_name}"
+            )
+
+            return True, f"Recording saved successfully as {audio_name}"
+
+        except Exception as e:
+            error_msg = f"Error saving recording: {str(e)}"
+            self.log_operation(error_msg, "error")
+            logger.error(traceback.format_exc())
+            return False, error_msg
+
+    def save_transcription(self, file_path: Path, text: str, metadata: Dict) -> None:
+        """Save transcription with metadata"""
+        content = f"""[METADATA]
+Recording_ID: {metadata['audio_file']}
+Speaker_ID: {metadata['speaker_id']}
+Dataset_Name: {metadata['dataset_name']}
+Timestamp: {metadata['timestamp']}
+Font_Style: {metadata['font_style']}
+
+[TEXT]
+{text}
+"""
+        with open(file_path, 'w', encoding='utf-8') as f:
+            f.write(content)
+
+    def update_metadata(self, speaker_id: str, dataset_name: str) -> None:
+        """Update dataset metadata with error handling"""
+        metadata_file = self.root_path / 'metadata' / 'dataset_info.json'
+
+        try:
+            if metadata_file.exists():
+                with open(metadata_file, 'r') as f:
+                    metadata = json.load(f)
+            else:
+                metadata = {'speakers': {}, 'last_updated': None}
+
+            # Update speaker data
+            if speaker_id not in metadata['speakers']:
+                metadata['speakers'][speaker_id] = {
+                    'total_recordings': 0,
+                    'datasets': {}
+                }
+
+            if dataset_name not in metadata['speakers'][speaker_id]['datasets']:
+                metadata['speakers'][speaker_id]['datasets'][dataset_name] = {
+                    'recordings': 0,
+                    'sentences': len(self.sentences),
+                    'recorded_sentences': [],
+                    'first_recording': datetime.now().isoformat(),
+                    'last_recording': None,
+                    'font_styles_used': []
+                }
+
+            # Update counts and timestamps
+            metadata['speakers'][speaker_id]['total_recordings'] += 1
+            metadata['speakers'][speaker_id]['datasets'][dataset_name]['recordings'] += 1
+            metadata['speakers'][speaker_id]['datasets'][dataset_name]['last_recording'] = \
+                datetime.now().isoformat()
+
+            # Add current index to recorded sentences
+            if self.current_index not in metadata['speakers'][speaker_id]['datasets'][dataset_name]['recorded_sentences']:
+                metadata['speakers'][speaker_id]['datasets'][dataset_name]['recorded_sentences'].append(self.current_index)
+
+            # Update font styles
+            if self.current_font not in metadata['speakers'][speaker_id]['datasets'][dataset_name]['font_styles_used']:
+                metadata['speakers'][speaker_id]['datasets'][dataset_name]['font_styles_used'].append(
+                    self.current_font
+                )
+
+            metadata['last_updated'] = datetime.now().isoformat()
+
+            # Save updated metadata
+            with open(metadata_file, 'w') as f:
+                json.dump(metadata, f, indent=2)
+
+            self.log_operation(f"Updated metadata for {speaker_id} in {dataset_name}")
+
+        except Exception as e:
+            error_msg = f"Error updating metadata: {str(e)}"
+            self.log_operation(error_msg, "error")
+            logger.error(traceback.format_exc())
+
+    def get_navigation_info(self) -> Dict[str, Optional[str]]:
+        """Get current and next sentence information"""
+        if not self.sentences:
+            return {
+                'current': None,
+                'next': None,
+                'progress': "No text loaded"
+            }
+
+        current = self.get_styled_text(self.sentences[self.current_index])
+        next_text = None
+
+        if self.current_index < len(self.sentences) - 1:
+            next_text = self.get_styled_text(self.sentences[self.current_index + 1])
+
+        progress = f"Sentence {self.current_index + 1} of {len(self.sentences)}"
+
+        return {
+            'current': current,
+            'next': next_text,
+            'progress': progress
+        }
+
+    def navigate(self, direction: str) -> Dict[str, Optional[str]]:
+        """Navigate through sentences"""
+        if not self.sentences:
+            return {
+                'current': None,
+                'next': None,
+                'progress': "No text loaded",
+                'status': "⚠️ Please load a text file first"
+            }
+
+        if direction == "next" and self.current_index < len(self.sentences) - 1:
+            self.current_index += 1
+        elif direction == "prev" and self.current_index > 0:
+            self.current_index -= 1
+
+        nav_info = self.get_navigation_info()
+        nav_info['status'] = "✅ Navigation successful"
+
+        return nav_info
+
+    def get_dataset_statistics(self) -> Dict:
+        """Get current dataset statistics"""
+        try:
+            metadata_file = self.root_path / 'metadata' / 'dataset_info.json'
+            if not metadata_file.exists():
+                return {}
+            with open(metadata_file, 'r') as f:
+                metadata = json.load(f)
+            # Flatten statistics for display
+            total_sentences = len(self.sentences)
+            recorded = sum(len(dataset.get('recorded_sentences', [])) for speaker in metadata['speakers'].values() for dataset in speaker['datasets'].values())
+            remaining = total_sentences - recorded
+            stats = {
+                "Total Sentences": total_sentences,
+                "Recorded Sentences": recorded,
+                "Remaining Sentences": remaining,
+                "Last Updated": metadata.get('last_updated', 'N/A')
+            }
+            return stats
+        except Exception as e:
+            logger.error(f"Error reading dataset statistics: {str(e)}")
+            return {}
+
+    def get_last_audio_path(self, speaker_id: str) -> Optional[str]:
+        """Get the path to the last saved audio file for downloading"""
+        audio_dir = self.root_path / 'audio' / speaker_id
+        audio_files = sorted(audio_dir.glob('*.wav'), key=lambda f: f.stat().st_mtime, reverse=True)
+        if audio_files:
+            return str(audio_files[0])
+        else:
+            return None
+
+    def get_last_transcript_path(self, speaker_id: str) -> Optional[str]:
+        """Get the path to the last saved transcription file for downloading"""
+        text_dir = self.root_path / 'transcriptions' / speaker_id
+        text_files = sorted(text_dir.glob('*.txt'), key=lambda f: f.stat().st_mtime, reverse=True)
+        if text_files:
+            return str(text_files[0])
+        else:
+            return None
 
-    # For brevity, the rest of the class methods are unchanged
-    # Please ensure to include the rest of the methods from the previous code
 
 def create_interface():
     """Create Gradio interface with enhanced features"""