Skip to content

Session Exporter

exporter

Classes

Exporter 

Exporter(session_repo: SessionRepository, segment_repo: SegmentRepository, utterance_repo: UtteranceRepository, store: Store)

Bases: LoggingMixin

Exporter for packaging session data and audio files.

Source code in audex/lib/exporter.py 
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    session_repo: SessionRepository,
    segment_repo: SegmentRepository,
    utterance_repo: UtteranceRepository,
    store: Store,
):
    super().__init__()
    self.session_repo = session_repo
    self.segment_repo = segment_repo
    self.utterance_repo = utterance_repo
    self.store = store
Functions
export_session_data async 
export_session_data(session_id: str) -> SessionExportData

Export session data as structured format.

Source code in audex/lib/exporter.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def export_session_data(self, session_id: str) -> SessionExportData:
    """Export session data as structured format."""
    # Get session
    session = await self.session_repo.read(session_id)
    if not session:
        raise ValueError(f"Session {session_id} not found")

    # Get utterances
    utt_filter = utterance_filter().session_id.eq(session_id).sequence.asc()
    utterances = await self.utterance_repo.list(utt_filter.build())

    # Get segments
    seg_filter = segment_filter().session_id.eq(session_id).sequence.asc()
    segments = await self.segment_repo.list(seg_filter.build())

    # Convert to typed dicts
    return SessionExportData(
        session=self._session_to_dict(session),
        utterances=[self._utterance_to_dict(u) for u in utterances],
        segments=[self._segment_to_dict(s) for s in segments],
    )
export_session_zip async 
export_session_zip(session_id: str) -> bytes

Export session as ZIP package.

Source code in audex/lib/exporter.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def export_session_zip(self, session_id: str) -> bytes:
    """Export session as ZIP package."""
    export_data = await self.export_session_data(session_id)
    zip_buffer = io.BytesIO()

    with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zipf:
        # Add conversation.json
        conversation: ConversationJSON = {
            "session": export_data["session"],
            "utterances": export_data["utterances"],
            "total_utterances": len(export_data["utterances"]),
            "total_segments": len(export_data["segments"]),
        }
        zipf.writestr(
            "conversation.json",
            json.dumps(conversation, ensure_ascii=False, indent=2),
        )

        # Add audio files
        audio_metadata_items: list[AudioMetadataItem] = []

        for idx, segment_dict in enumerate(export_data["segments"], start=1):
            audio_key = segment_dict["audio_key"]

            try:
                audio_data = await self.store.download(audio_key)
                ext = pathlib.Path(audio_key).suffix or ".mp3"
                filename = f"segment_{idx:03d}{ext}"

                zipf.writestr(f"audio/{filename}", audio_data)

                audio_metadata_items.append(
                    AudioMetadataItem(
                        filename=filename,
                        sequence=segment_dict["sequence"],
                        duration_ms=segment_dict["duration_ms"],
                        started_at=segment_dict["started_at"],
                        ended_at=segment_dict["ended_at"],
                    )
                )

                self.logger.debug(f"Added audio file: {filename}")

            except Exception as e:
                self.logger.error(f"Failed to add audio {audio_key}: {e}")

        # Add audio metadata
        if audio_metadata_items:
            audio_metadata: AudioMetadataJSON = {
                "session_id": session_id,
                "total_segments": len(audio_metadata_items),
                "segments": audio_metadata_items,
            }
            zipf.writestr(
                "audio/metadata.json",
                json.dumps(audio_metadata, ensure_ascii=False, indent=2),
            )

    zip_buffer.seek(0)
    return zip_buffer.getvalue()
export_multiple_sessions_zip async 
export_multiple_sessions_zip(session_ids: list[str]) -> bytes

Export multiple sessions as ZIP package.

Source code in audex/lib/exporter.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def export_multiple_sessions_zip(self, session_ids: list[str]) -> bytes:
    """Export multiple sessions as ZIP package."""
    zip_buffer = io.BytesIO()

    with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zipf:
        for session_id in session_ids:
            try:
                session_zip_data = await self.export_session_zip(session_id)

                with zipfile.ZipFile(io.BytesIO(session_zip_data), "r") as session_zipf:
                    for file_info in session_zipf.infolist():
                        file_data = session_zipf.read(file_info.filename)
                        new_path = f"{session_id}/{file_info.filename}"
                        zipf.writestr(new_path, file_data)

                self.logger.info(f"Added session {session_id} to export")

            except Exception as e:
                self.logger.error(f"Failed to export session {session_id}: {e}")

    zip_buffer.seek(0)
    return zip_buffer.getvalue()

Functions

options: show_root_heading: true show_source: true heading_level: 2 members_order: source show_signature_annotations: true separate_signature: true