| import logging
|
| import threading
|
| from datetime import datetime
|
| from typing import Dict
|
|
|
| logger = logging.getLogger(__name__)
|
|
|
|
|
| class TranscriptionStatus:
|
| """Simple transcription status tracker"""
|
|
|
| def __init__(self):
|
| self.is_busy = False
|
| self.current_operation = None
|
| self.current_filename = None
|
| self.started_at = None
|
| self.progress = 0.0
|
| self.lock = threading.Lock()
|
| self.total_completed = 0
|
|
|
| def start_transcription(self, operation_type: str, filename: str = None):
|
| """Mark transcription as started"""
|
| with self.lock:
|
| self.is_busy = True
|
| self.current_operation = operation_type
|
| self.current_filename = filename
|
| self.started_at = datetime.now()
|
| self.progress = 0.0
|
| logger.info(f"Started {operation_type} transcription for {filename or 'unknown file'}")
|
|
|
| def update_progress(self, progress: float):
|
| """Update transcription progress (0.0 to 1.0)"""
|
| with self.lock:
|
| self.progress = max(0.0, min(1.0, progress))
|
|
|
| def finish_transcription(self):
|
| """Mark transcription as finished"""
|
| with self.lock:
|
| self.is_busy = False
|
| self.current_operation = None
|
| self.current_filename = None
|
| self.started_at = None
|
| self.progress = 0.0
|
| self.total_completed += 1
|
| logger.info("Transcription finished")
|
|
|
| def get_status(self) -> Dict:
|
| """Get current status for API response"""
|
| with self.lock:
|
| status = {"is_busy": self.is_busy, "total_completed": self.total_completed}
|
|
|
| if self.is_busy:
|
| duration = (
|
| (datetime.now() - self.started_at).total_seconds()
|
| if self.started_at
|
| else 0
|
| )
|
| status.update(
|
| {
|
| "current_operation": self.current_operation,
|
| "current_filename": self.current_filename,
|
| "progress": self.progress,
|
| "duration_seconds": round(duration, 1),
|
| }
|
| )
|
|
|
| return status
|
|
|
|
|
|
|
| transcription_status = TranscriptionStatus()
|
|
|
