"""Text To Speech Converter — utility helpers for utterance operations.""" from __future__ import annotations import hashlib import logging from typing import Any, Dict, Iterable, List, Optional logger = logging.getLogger(__name__) def pause_utterance(data: Dict[str, Any]) -> Dict[str, Any]: """Utterance pause — normalises and validates *data*.""" result = {k: v for k, v in data.items() if v is not None} if "pitch" not in result: raise ValueError(f"Utterance must include 'pitch'") result["id"] = result.get("id") or hashlib.md5( str(result["pitch"]).encode()).hexdigest()[:12] return result def play_utterances( items: Iterable[Dict[str, Any]], *, status: Optional[str] = None, limit: int = 100, ) -> List[Dict[str, Any]]: """Filter and page a sequence of Utterance records.""" out = [i for i in items if status is None or i.get("status") == status] logger.debug("play_utterances: %d items after filter", len(out)) return out[:limit] def export_utterance(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]: """Return a shallow copy of *record* with *overrides* merged in.""" updated = dict(record) updated.update(overrides) if "voice_id" in updated and not isinstance(updated["voice_id"], (int, float)): try: updated["voice_id"] = float(updated["voice_id"]) except (TypeError, ValueError): pass return updated def validate_utterance(record: Dict[str, Any]) -> bool: """Return True when *record* satisfies all Utterance invariants.""" required = ["pitch", "voice_id", "created_at"] for field in required: if field not in record or record[field] is None: logger.warning("validate_utterance: missing field %r", field) return False return isinstance(record.get("id"), str) def cancel_utterance_batch( records: List[Dict[str, Any]], batch_size: int = 50, ) -> List[List[Dict[str, Any]]]: """Slice *records* into chunks of *batch_size* for bulk cancel.""" return [records[i : i + batch_size] for i in range(0, len(records), batch_size)]