update file handler

Author: seanreed1111

packages/livekit-recording/src/livekit_recording/__init__.py

@@ -22,11 +22,13 @@
     StorageSettings,
 )
 from livekit_recording.transcript import (
+    LocalTranscriptStorage,
     S3Uploader,
     S3UploaderProtocol,
     TranscriptData,
     TranscriptEntry,
     TranscriptHandler,
+    TranscriptStorageProtocol,
 )
 
 __all__ = [
@@ -38,6 +40,7 @@
     "EgressManager",
     "LiveKitSettings",
     "LocalAudioRecorder",
+    "LocalTranscriptStorage",
     "S3AudioRecorder",
     "S3Settings",
     "S3Uploader",
@@ -48,6 +51,7 @@
     "TranscriptData",
     "TranscriptEntry",
     "TranscriptHandler",
+    "TranscriptStorageProtocol",
     "create_default_egress_manager",
     "create_default_s3_recorder",
 ]

packages/livekit-recording/src/livekit_recording/audio_storage.py

@@ -193,7 +193,7 @@ def on_local_track_published(
                     # Recording is active and not stopping, subscribe immediately
                     _task = asyncio.create_task(
                         self._subscribe_to_track(track, "agent")
-                    )  # noqa: RUF006
+                    )
                 elif not self._stop_event:
                     # Recording not started yet, queue the track
                     self._pending_agent_tracks.append(track)

packages/livekit-recording/src/livekit_recording/settings.py

@@ -14,6 +14,7 @@
     from livekit import rtc
 
     from livekit_recording.audio_storage import AudioRecorderProtocol
+    from livekit_recording.transcript import TranscriptStorageProtocol
 
 
 class StorageMode(str, Enum):
@@ -276,3 +277,48 @@ def create_audio_recorder(
 
         else:
             raise ValueError(f"Unknown storage mode: {self.storage.mode}")
+
+    def create_transcript_storage(
+        self,
+        bucket: str | None = None,
+        prefix: str | None = None,
+    ) -> TranscriptStorageProtocol:
+        """Create a transcript storage backend based on the current storage mode.
+
+        Args:
+            bucket: S3 bucket name (required for S3 mode, uses settings.s3.bucket if not provided)
+            prefix: S3 prefix (uses settings.s3.prefix if not provided)
+
+        Returns:
+            A TranscriptStorageProtocol implementation (LocalTranscriptStorage or S3Uploader)
+
+        Raises:
+            ValueError: If required configuration is missing for S3 mode
+        """
+        # Import here to avoid circular imports
+        from livekit_recording.transcript import LocalTranscriptStorage, S3Uploader
+
+        if self.storage.mode == StorageMode.LOCAL:
+            logger.info(
+                f"Creating LocalTranscriptStorage with output_dir={self.storage.local_output_dir}"
+            )
+            return LocalTranscriptStorage(output_dir=self.storage.local_output_dir)
+
+        elif self.storage.mode == StorageMode.S3:
+            # Use provided bucket/prefix or fall back to settings
+            s3_bucket = bucket or self.s3.bucket
+            s3_prefix = prefix if prefix is not None else self.s3.prefix
+
+            if not s3_bucket:
+                raise ValueError(
+                    "S3 bucket is required for S3 storage mode. "
+                    "Set STORAGE_MODE=local or provide bucket parameter."
+                )
+
+            logger.info(
+                f"Creating S3Uploader with bucket={s3_bucket}, prefix={s3_prefix}"
+            )
+            return S3Uploader.from_settings(self, bucket=s3_bucket, prefix=s3_prefix)
+
+        else:
+            raise ValueError(f"Unknown storage mode: {self.storage.mode}")

packages/livekit-recording/src/livekit_recording/transcript.py

@@ -1,10 +1,11 @@
-"""Transcript handler for capturing and storing STT output to S3."""
+"""Transcript handler for capturing and storing STT output to S3 or local files."""
 
 from __future__ import annotations
 
 import json
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
+from pathlib import Path
 from typing import TYPE_CHECKING, Protocol
 
 import boto3
@@ -52,10 +53,27 @@ def to_dict(self) -> dict:
         }
 
 
-class S3UploaderProtocol(Protocol):
-    """Protocol for S3 upload operations."""
+class TranscriptStorageProtocol(Protocol):
+    """Protocol for transcript storage operations.
 
-    def upload_transcript(self, transcript: TranscriptData, key: str) -> bool: ...
+    Implementations can save transcripts to various backends (local files, S3, etc.).
+    """
+
+    def save_transcript(self, transcript: TranscriptData, key: str) -> bool:
+        """Save a transcript to storage.
+
+        Args:
+            transcript: The transcript data to save
+            key: The storage key/path for the transcript
+
+        Returns:
+            True if save succeeded, False otherwise
+        """
+        ...
+
+
+# Keep S3UploaderProtocol for backward compatibility
+S3UploaderProtocol = TranscriptStorageProtocol
 
 
 class S3Uploader:
@@ -136,28 +154,92 @@ def upload_transcript(self, transcript: TranscriptData, key: str) -> bool:
             logger.error(f"Failed to upload transcript to S3: {e}")
             return False
 
+    def save_transcript(self, transcript: TranscriptData, key: str) -> bool:
+        """Save transcript JSON to S3 (alias for upload_transcript).
+
+        Args:
+            transcript: The transcript data to save
+            key: The S3 object key (path within bucket)
+
+        Returns:
+            True if save succeeded, False otherwise
+        """
+        return self.upload_transcript(transcript, key)
+
+
+class LocalTranscriptStorage:
+    """Handles saving transcripts to local filesystem."""
+
+    def __init__(self, output_dir: str | Path = "temp"):
+        """Initialize the local transcript storage.
+
+        Args:
+            output_dir: Directory to save transcript files (default: temp/)
+        """
+        self.output_dir = Path(output_dir)
+        logger.debug(f"LocalTranscriptStorage initialized with output_dir={output_dir}")
+
+    def save_transcript(self, transcript: TranscriptData, key: str) -> bool:
+        """Save transcript JSON to local filesystem.
+
+        Args:
+            transcript: The transcript data to save
+            key: The file path relative to output_dir (e.g., transcripts/room-session.json)
+
+        Returns:
+            True if save succeeded, False otherwise
+        """
+        # Build full path
+        output_path = self.output_dir / key
+
+        try:
+            # Create parent directories if needed
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+
+            # Write JSON file
+            json_content = json.dumps(transcript.to_dict(), indent=2)
+            output_path.write_text(json_content, encoding="utf-8")
+
+            logger.info(f"Saved transcript to {output_path.absolute()}")
+            return True
+        except OSError as e:
+            logger.error(f"Failed to save transcript to local file: {e}")
+            return False
+
+    # Alias for backward compatibility
+    def upload_transcript(self, transcript: TranscriptData, key: str) -> bool:
+        """Alias for save_transcript for backward compatibility."""
+        return self.save_transcript(transcript, key)
+
 
 class TranscriptHandler:
     """Handles capturing and storing conversation transcripts."""
 
     def __init__(
         self,
         room_name: str,
-        s3_uploader: S3UploaderProtocol | None = None,
+        storage: TranscriptStorageProtocol | None = None,
         session_id: str | None = None,
+        *,
+        # Backward compatibility alias
+        s3_uploader: TranscriptStorageProtocol | None = None,
     ):
         """Initialize the transcript handler.
 
         Args:
             room_name: Name of the LiveKit room
-            s3_uploader: S3 uploader instance for storing transcripts
+            storage: Storage instance for saving transcripts (local or S3)
             session_id: Unique session identifier for matching audio/transcript files
+            s3_uploader: Deprecated alias for storage parameter (backward compatibility)
         """
         self.transcript = TranscriptData(
             room_name=room_name,
             session_start=datetime.now(UTC).isoformat(),
         )
-        self.s3_uploader = s3_uploader
+        # Support both 'storage' and deprecated 's3_uploader' parameter
+        self.storage = storage or s3_uploader
+        # Keep s3_uploader as alias for backward compatibility
+        self.s3_uploader = self.storage
         # Use provided session_id or generate one
         self.session_id = session_id or datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
 
@@ -199,22 +281,30 @@ def add_agent_transcript(self, text: str, is_final: bool = True) -> None:
         self.transcript.entries.append(entry)
         logger.debug(f"Agent transcript: {text}")
 
-    async def finalize_and_upload(self) -> bool:
-        """Finalize the transcript and upload to S3.
+    async def finalize_and_save(self) -> bool:
+        """Finalize the transcript and save to storage (local or S3).
 
         Returns:
-            True if upload succeeded or no uploader configured, False on failure
+            True if save succeeded or no storage configured, False on failure
         """
         self.transcript.session_end = datetime.now(UTC).isoformat()
 
-        if not self.s3_uploader:
-            logger.warning("No S3 uploader configured, transcript not saved")
+        if not self.storage:
+            logger.warning("No transcript storage configured, transcript not saved")
             return True
 
         # Use session_id for filename to match audio recording
         key = f"transcripts/{self.transcript.room_name}-{self.session_id}.json"
 
-        return self.s3_uploader.upload_transcript(self.transcript, key)
+        return self.storage.save_transcript(self.transcript, key)
+
+    async def finalize_and_upload(self) -> bool:
+        """Alias for finalize_and_save (backward compatibility).
+
+        Returns:
+            True if save succeeded or no storage configured, False on failure
+        """
+        return await self.finalize_and_save()
 
     def get_transcript_text(self) -> str:
         """Get the transcript as plain text.

src/agent.py

@@ -18,7 +18,6 @@
 from livekit.plugins.turn_detector.multilingual import MultilingualModel
 from livekit_recording import (
     AudioRecorderProtocol,
-    S3Uploader,
     Settings,
     StorageMode,
     TranscriptHandler,
@@ -106,19 +105,21 @@ async def my_agent(ctx: JobContext):
     except Exception as e:
         logger.error(f"Failed to initialize audio recorder: {e}")
 
-    # Initialize transcript handler for saving STT output
+    # Initialize transcript handler for saving STT output (local or S3 based on storage mode)
     transcript_handler = None
     try:
         logger.info("Initializing transcript handler...")
-        s3_uploader = S3Uploader.from_settings(
-            settings, bucket=S3_BUCKET, prefix=S3_PREFIX
+        transcript_storage = settings.create_transcript_storage(
+            bucket=S3_BUCKET, prefix=S3_PREFIX
         )
         transcript_handler = TranscriptHandler(
             room_name=room_name,
-            s3_uploader=s3_uploader,
+            storage=transcript_storage,
             session_id=session_id,
         )
-        logger.info("Transcript handler initialized successfully")
+        logger.info(
+            f"Transcript handler initialized successfully (mode={settings.storage.mode.value})"
+        )
     except Exception as e:
         logger.error(f"Failed to initialize transcript handler: {e}")
 

temp/transcripts/playground-ggC8-V5BN-20251212-211022.json

@@ -0,0 +1,43 @@
+{
+  "room_name": "playground-ggC8-V5BN",
+  "session_start": "2025-12-12T21:10:22.555334+00:00",
+  "session_end": "2025-12-12T21:11:46.970180+00:00",
+  "entries": [
+    {
+      "timestamp": "2025-12-12T21:10:25.345108+00:00",
+      "speaker": "agent",
+      "text": "Hello, how can I assist you?",
+      "is_final": true
+    },
+    {
+      "timestamp": "2025-12-12T21:10:29.422298+00:00",
+      "speaker": "user",
+      "text": "Yes.",
+      "is_final": true
+    },
+    {
+      "timestamp": "2025-12-12T21:10:36.456330+00:00",
+      "speaker": "user",
+      "text": "Are there any planets that Rotate in the opposite direction than the sun.",
+      "is_final": true
+    },
+    {
+      "timestamp": "2025-12-12T21:10:56.089449+00:00",
+      "speaker": "agent",
+      "text": "Yes, Venus and Uranus rotate in the opposite direction to the Sun. This is called retrograde rotation. Venus spins clockwise on its axis, while the Sun spins counterclockwise when viewed from above their north poles. Uranus is unique because it rotates on its side with an axial tilt of about 98 degrees, which makes its rotation appear retrograde as well. Interesting, right?",
+      "is_final": true
+    },
+    {
+      "timestamp": "2025-12-12T21:11:09.955625+00:00",
+      "speaker": "user",
+      "text": "How long did the roman empire last",
+      "is_final": true
+    },
+    {
+      "timestamp": "2025-12-12T21:11:30.215535+00:00",
+      "speaker": "agent",
+      "text": "The Roman Empire lasted about 500 years if you start from 27 BC when Augustus became the first emperor, until 476 AD when the Western Roman Empire fell. If you include the Eastern Roman Empire, also called the Byzantine Empire, it lasted until 1453 AD, so that\u2019s almost 1500 years in total. Quite a long run!",
+      "is_final": true
+    }
+  ]
+}