From 973bab6781f13acaa51b5530b63a8b178775cf44 Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Tue, 11 Aug 2026 13:08:17 -0600 Subject: [PATCH] refactor(transcriber): move prerecorded transcription into prerecorded/v2 Relocates the async (prerecorded) product into a versioned subpackage mirroring sync/v1 and streaming/v3: - transcriber.py -> prerecorded/v2/{transcript,transcript_group,client}.py, one public class per module beside its private impl - transcript-only endpoints split out of api.py into prerecorded/v2/api.py; shared (upload_file, _get_error_message) and LeMUR pieces stay at root - flat transcriber.py and api.py silently re-export the old surface, so every existing import path keeps resolving to identical objects - adds test_transcriber_backwards_compat.py pinning old/new path identity (v0.64.34) Co-Authored-By: Claude Fable 5 --- assemblyai/__version__.py | 2 +- assemblyai/api.py | 246 +--- assemblyai/prerecorded/__init__.py | 0 assemblyai/prerecorded/v2/__init__.py | 13 + assemblyai/prerecorded/v2/api.py | 236 +++ assemblyai/prerecorded/v2/client.py | 439 ++++++ assemblyai/prerecorded/v2/transcript.py | 651 +++++++++ assemblyai/prerecorded/v2/transcript_group.py | 202 +++ assemblyai/transcriber.py | 1284 +---------------- assemblyai/types.py | 2 +- .../unit/test_transcriber_backwards_compat.py | 109 ++ 11 files changed, 1684 insertions(+), 1500 deletions(-) create mode 100644 assemblyai/prerecorded/__init__.py create mode 100644 assemblyai/prerecorded/v2/__init__.py create mode 100644 assemblyai/prerecorded/v2/api.py create mode 100644 assemblyai/prerecorded/v2/client.py create mode 100644 assemblyai/prerecorded/v2/transcript.py create mode 100644 assemblyai/prerecorded/v2/transcript_group.py create mode 100644 tests/unit/test_transcriber_backwards_compat.py diff --git a/assemblyai/__version__.py b/assemblyai/__version__.py index c7b8c61..bb14b99 100644 --- a/assemblyai/__version__.py +++ b/assemblyai/__version__.py @@ -1 +1 @@ -__version__ = "0.64.33" +__version__ = "0.64.34" diff --git a/assemblyai/api.py b/assemblyai/api.py index 3719879..25ad9f0 100644 --- a/assemblyai/api.py +++ b/assemblyai/api.py @@ -1,11 +1,9 @@ -from typing import BinaryIO, List, Optional, Union -from urllib.parse import urlencode +from typing import BinaryIO, Optional, Union import httpx from . import types -ENDPOINT_TRANSCRIPT = "/v2/transcript" ENDPOINT_UPLOAD = "/v2/upload" ENDPOINT_LEMUR_BASE = "/lemur/v3" ENDPOINT_LEMUR = f"{ENDPOINT_LEMUR_BASE}/generate" @@ -28,60 +26,6 @@ def _get_error_message(response: httpx.Response) -> str: return f"\nReason: {response.text}\nRequest: {response.request}" -def create_transcript( - client: httpx.Client, - request: types.TranscriptRequest, -) -> types.TranscriptResponse: - response = client.post( - ENDPOINT_TRANSCRIPT, - json=request.dict( - exclude_none=True, - by_alias=True, - ), - ) - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to transcribe url {request.audio_url}: {_get_error_message(response)}", - response.status_code, - ) - - return types.TranscriptResponse.parse_obj(response.json()) - - -def get_transcript( - client: httpx.Client, - transcript_id: str, -) -> types.TranscriptResponse: - response = client.get( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}", - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.TranscriptResponse.parse_obj(response.json()) - - -def delete_transcript( - client: httpx.Client, - transcript_id: str, -) -> types.TranscriptResponse: - response = client.delete( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}", - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to delete transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.TranscriptResponse.parse_obj(response.json()) - - def upload_file( client: httpx.Client, audio_file: Union[bytes, BinaryIO], @@ -110,177 +54,6 @@ def upload_file( return response.json()["upload_url"] -def export_subtitles_srt( - client: httpx.Client, - transcript_id: str, - chars_per_caption: Optional[int], -) -> str: - params = {} - - if chars_per_caption: - params = { - "chars_per_caption": chars_per_caption, - } - - response = client.get( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/srt", - params=params, - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to export SRT for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return response.text - - -def export_subtitles_vtt( - client: httpx.Client, - transcript_id: str, - chars_per_caption: Optional[int], -) -> str: - params = {} - - if chars_per_caption: - params = { - "chars_per_caption": chars_per_caption, - } - - response = client.get( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/vtt", - params=params, - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to export VTT for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return response.text - - -def word_search( - client: httpx.Client, - transcript_id: str, - words: List[str], -) -> types.WordSearchMatchResponse: - response = client.get( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/word-search", - params=urlencode( - { - "words": ",".join(words), - } - ), - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to search words in transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.WordSearchMatchResponse.parse_obj(response.json()) - - -def get_redacted_audio( - client: httpx.Client, transcript_id: str -) -> types.RedactedAudioResponse: - """ - Retrieves the object containing the redacted audio URL for the given transcript. - - Raises: - RedactedAudioIncompleteError: If response indicates that the redacted audio is still processing - RedactedAudioUnavailableError: If response indicates that the redacted audio is not available - TranscriptError: If we fail to get a valid response from the API at all - - Returns: - `RedactedAudioResponse`, which contains the URL of the redacted audio - """ - - response = client.get(f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/redacted-audio") - - if response.status_code == httpx.codes.ACCEPTED: - raise types.RedactedAudioIncompleteError( - f"redacted audio for transcript {transcript_id} is not ready yet", - response.status_code, - ) - - if response.status_code == httpx.codes.BAD_REQUEST: - raise types.RedactedAudioExpiredError( - f"redacted audio for transcript {transcript_id} is no longer available", - response.status_code, - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve redacted audio for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.RedactedAudioResponse.parse_obj(response.json()) - - -def get_sentences( - client: httpx.Client, - transcript_id: str, -) -> types.SentencesResponse: - response = client.get( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/sentences", - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve sentences for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.SentencesResponse.parse_obj(response.json()) - - -def get_paragraphs( - client: httpx.Client, - transcript_id: str, -) -> types.ParagraphsResponse: - response = client.get( - f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/paragraphs", - ) - - if response.status_code != httpx.codes.OK: - raise types.TranscriptError( - f"failed to retrieve paragraphs for transcript {transcript_id}: {_get_error_message(response)}", - response.status_code, - ) - - return types.ParagraphsResponse.parse_obj(response.json()) - - -def list_transcripts( - client: httpx.Client, - params: Optional[types.ListTranscriptParameters], -) -> types.ListTranscriptResponse: - response = client.get( - ENDPOINT_TRANSCRIPT, - params=( - params.dict( - exclude_none=True, - ) - if params - else None - ), - ) - - if response.status_code != httpx.codes.OK: - raise types.AssemblyAIError( - f"failed to retrieve transcripts: {_get_error_message(response)}", - response.status_code, - ) - - return types.ListTranscriptResponse.parse_obj(response.json()) - - def lemur_question( client: httpx.Client, request: types.LemurQuestionRequest, @@ -413,3 +186,20 @@ def lemur_get_response_data( return types.LemurQuestionResponse.parse_obj(json_data) return types.LemurStringResponse.parse_obj(json_data) + + +# Canonical location for the prerecorded transcript endpoints is +# ``assemblyai.prerecorded.v2.api``. +from .prerecorded.v2.api import ( # noqa: E402, F401 + ENDPOINT_TRANSCRIPT, + create_transcript, + delete_transcript, + export_subtitles_srt, + export_subtitles_vtt, + get_paragraphs, + get_redacted_audio, + get_sentences, + get_transcript, + list_transcripts, + word_search, +) diff --git a/assemblyai/prerecorded/__init__.py b/assemblyai/prerecorded/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/assemblyai/prerecorded/v2/__init__.py b/assemblyai/prerecorded/v2/__init__.py new file mode 100644 index 0000000..91e8eb1 --- /dev/null +++ b/assemblyai/prerecorded/v2/__init__.py @@ -0,0 +1,13 @@ +"""Prerecorded (async job) transcription against the v2 transcript API.""" + +from ...types import TranscriptionConfig +from .client import Transcriber +from .transcript import Transcript +from .transcript_group import TranscriptGroup + +__all__ = [ + "Transcriber", + "Transcript", + "TranscriptGroup", + "TranscriptionConfig", +] diff --git a/assemblyai/prerecorded/v2/api.py b/assemblyai/prerecorded/v2/api.py new file mode 100644 index 0000000..7cf04e4 --- /dev/null +++ b/assemblyai/prerecorded/v2/api.py @@ -0,0 +1,236 @@ +"""HTTP calls against the v2 transcript API endpoints.""" + +from typing import List, Optional +from urllib.parse import urlencode + +import httpx + +from ... import api as _root_api +from ... import types + +ENDPOINT_TRANSCRIPT = "/v2/transcript" + + +def create_transcript( + client: httpx.Client, + request: types.TranscriptRequest, +) -> types.TranscriptResponse: + response = client.post( + ENDPOINT_TRANSCRIPT, + json=request.dict( + exclude_none=True, + by_alias=True, + ), + ) + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to transcribe url {request.audio_url}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.TranscriptResponse.parse_obj(response.json()) + + +def get_transcript( + client: httpx.Client, + transcript_id: str, +) -> types.TranscriptResponse: + response = client.get( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}", + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to retrieve transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.TranscriptResponse.parse_obj(response.json()) + + +def delete_transcript( + client: httpx.Client, + transcript_id: str, +) -> types.TranscriptResponse: + response = client.delete( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}", + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to delete transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.TranscriptResponse.parse_obj(response.json()) + + +def export_subtitles_srt( + client: httpx.Client, + transcript_id: str, + chars_per_caption: Optional[int], +) -> str: + params = {} + + if chars_per_caption: + params = { + "chars_per_caption": chars_per_caption, + } + + response = client.get( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/srt", + params=params, + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to export SRT for transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return response.text + + +def export_subtitles_vtt( + client: httpx.Client, + transcript_id: str, + chars_per_caption: Optional[int], +) -> str: + params = {} + + if chars_per_caption: + params = { + "chars_per_caption": chars_per_caption, + } + + response = client.get( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/vtt", + params=params, + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to export VTT for transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return response.text + + +def word_search( + client: httpx.Client, + transcript_id: str, + words: List[str], +) -> types.WordSearchMatchResponse: + response = client.get( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/word-search", + params=urlencode( + { + "words": ",".join(words), + } + ), + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to search words in transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.WordSearchMatchResponse.parse_obj(response.json()) + + +def get_redacted_audio( + client: httpx.Client, transcript_id: str +) -> types.RedactedAudioResponse: + """ + Retrieves the object containing the redacted audio URL for the given transcript. + + Raises: + RedactedAudioIncompleteError: If response indicates that the redacted audio is still processing + RedactedAudioUnavailableError: If response indicates that the redacted audio is not available + TranscriptError: If we fail to get a valid response from the API at all + + Returns: + `RedactedAudioResponse`, which contains the URL of the redacted audio + """ + + response = client.get(f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/redacted-audio") + + if response.status_code == httpx.codes.ACCEPTED: + raise types.RedactedAudioIncompleteError( + f"redacted audio for transcript {transcript_id} is not ready yet", + response.status_code, + ) + + if response.status_code == httpx.codes.BAD_REQUEST: + raise types.RedactedAudioExpiredError( + f"redacted audio for transcript {transcript_id} is no longer available", + response.status_code, + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to retrieve redacted audio for transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.RedactedAudioResponse.parse_obj(response.json()) + + +def get_sentences( + client: httpx.Client, + transcript_id: str, +) -> types.SentencesResponse: + response = client.get( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/sentences", + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to retrieve sentences for transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.SentencesResponse.parse_obj(response.json()) + + +def get_paragraphs( + client: httpx.Client, + transcript_id: str, +) -> types.ParagraphsResponse: + response = client.get( + f"{ENDPOINT_TRANSCRIPT}/{transcript_id}/paragraphs", + ) + + if response.status_code != httpx.codes.OK: + raise types.TranscriptError( + f"failed to retrieve paragraphs for transcript {transcript_id}: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.ParagraphsResponse.parse_obj(response.json()) + + +def list_transcripts( + client: httpx.Client, + params: Optional[types.ListTranscriptParameters], +) -> types.ListTranscriptResponse: + response = client.get( + ENDPOINT_TRANSCRIPT, + params=( + params.dict( + exclude_none=True, + ) + if params + else None + ), + ) + + if response.status_code != httpx.codes.OK: + raise types.AssemblyAIError( + f"failed to retrieve transcripts: {_root_api._get_error_message(response)}", + response.status_code, + ) + + return types.ListTranscriptResponse.parse_obj(response.json()) diff --git a/assemblyai/prerecorded/v2/client.py b/assemblyai/prerecorded/v2/client.py new file mode 100644 index 0000000..0a9050a --- /dev/null +++ b/assemblyai/prerecorded/v2/client.py @@ -0,0 +1,439 @@ +"""The ``Transcriber`` entry point for prerecorded transcription.""" + +from __future__ import annotations + +import concurrent.futures +import os +from typing import BinaryIO, List, Optional, Set, Tuple, Union +from urllib.parse import urlparse + +from ... import api as _root_api +from ... import client as _client +from ... import types +from . import api +from .transcript import Transcript +from .transcript_group import TranscriptGroup + + +class _TranscriberImpl: + """ + Implementation of the Transcriber class. + """ + + def __init__( + self, + *, + client: _client.Client, + config: types.TranscriptionConfig, + ) -> None: + self._client = client + self.config = config + + def upload_file(self, data: Union[str, bytes, BinaryIO]) -> str: + if isinstance(data, str): + with open(data, "rb") as audio_file: + return _root_api.upload_file( + client=self._client.http_client, + audio_file=audio_file, + ) + else: + return _root_api.upload_file( + client=self._client.http_client, + audio_file=data, + ) + + def transcribe_url( + self, + *, + url: str, + config: types.TranscriptionConfig, + poll: bool, + ) -> Transcript: + transcript_request = types.TranscriptRequest( + audio_url=url, + **config.raw.dict(exclude_none=True), + ) + # No try-except - if there is an HTTP error raise it to the user + transcript = Transcript.from_response( + client=self._client, + response=api.create_transcript( + client=self._client.http_client, + request=transcript_request, + ), + ) + + if poll: + return transcript.wait_for_completion() + + return transcript + + def transcribe_file( + self, + *, + data: Union[str, bytes, BinaryIO], + config: types.TranscriptionConfig, + poll: bool, + ) -> Transcript: + # Note: If uploading fails, it should raise an Exception to the user, hence no try-except here. + audio_url = self.upload_file(data) + + return self.transcribe_url( + url=audio_url, + config=config, + poll=poll, + ) + + def transcribe( + self, + data: Union[str, bytes, BinaryIO], + config: Optional[types.TranscriptionConfig], + poll: bool, + ) -> Transcript: + if config is None: + config = self.config + + if isinstance(data, str) and urlparse(data).scheme in {"http", "https"}: + return self.transcribe_url( + url=data, + config=config, + poll=poll, + ) + + return self.transcribe_file( + data=data, + config=config, + poll=poll, + ) + + def transcribe_group( + self, + *, + data: List[Union[str, bytes, BinaryIO]], + config: Optional[types.TranscriptionConfig], + poll: bool, + return_failures: Optional[bool] = False, + ) -> Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]]: + if config is None: + config = self.config + + future_transcripts: Set[concurrent.futures.Future[Transcript]] = set() + + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: + for d in data: + transcript_future = executor.submit( + self.transcribe, + data=d, + config=config, + poll=False, + ) + + future_transcripts.add(transcript_future) + + finished_futures, _ = concurrent.futures.wait(future_transcripts) + + transcript_group = TranscriptGroup( + client=self._client, + ) + failures: List[types.AssemblyAIError] = [] + + for future in finished_futures: + try: + transcript_group.add_transcript(future.result()) + except types.TranscriptError as e: + failures.append(e) + + if poll is True and return_failures is True: + res = transcript_group.wait_for_completion(return_failures=return_failures) + if not isinstance(res, tuple): + raise ValueError( + "return_failures was set but did not receive failures object" + ) + transcript_group, completion_failures = res + failures.extend(completion_failures) + elif poll: + res = transcript_group.wait_for_completion(return_failures=return_failures) + if not isinstance(res, TranscriptGroup): + raise ValueError( + "return_failures was not set but did receive failures object" + ) + transcript_group = res + + if return_failures is True: + return transcript_group, failures + else: + return transcript_group + + def list_transcripts( + self, + params: Optional[types.ListTranscriptParameters], + ) -> types.ListTranscriptResponse: + return api.list_transcripts(client=self._client.http_client, params=params) + + +class Transcriber: + """ + A transcriber used for transcribing URLs or local audio files. + """ + + def __init__( + self, + *, + client: Optional[_client.Client] = None, + config: Optional[types.TranscriptionConfig] = None, + max_workers: Optional[int] = None, + ) -> None: + """ + Initializes the `Transcriber` with the given parameters. + + Args: + `client`: The `Client` to use for the `Transcriber`. If `None` is given, the + default settings for the `Client` will be used. + `config`: The default configuration for the `Transcriber`. If `None` is given, + the default configuration of a `TranscriptionConfig` will be used. + `max_workers`: The maximum number of parallel jobs when using the `_async` + methods on the `Transcriber`. By default it uses `os.cpu_count() - 1` + + Example: + To use the `Transcriber` with the default settings, you can simply do: + ``` + transcriber = aai.Transcriber() + ``` + + To use the `Transcriber` with a custom configuration, you can do: + ``` + config = aai.TranscriptionConfig(punctuate=False, format_text=False) + + transcriber = aai.Transcriber(config=config) + ``` + """ + self._client = client or _client.Client.get_default() + + self._impl = _TranscriberImpl( + client=self._client, + config=config or types.TranscriptionConfig(), + ) + + if not max_workers: + cpu_count = os.cpu_count() + if not cpu_count: + max_workers = 1 + else: + max_workers = max(1, cpu_count - 1) + + self._executor = concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers, + ) + + @property + def config(self) -> types.TranscriptionConfig: + """ + Returns the default configuration of the `Transcriber`. + """ + return self._impl.config + + @config.setter + def config(self, config: types.TranscriptionConfig) -> None: + """ + Sets the default configuration of the `Transcriber`. + + Args: + `config`: The new default configuration. + """ + self._impl.config = config + + def upload_file(self, data: Union[str, bytes, BinaryIO]) -> str: + """ + Uploads an audio file which can be specified as local path or binary object. + + Args: + `data`: A local file (as path), or a binary object. + + Returns: The URL of the uploaded audio file. + """ + return self._impl.upload_file(data=data) + + def upload_file_async( + self, data: Union[str, bytes, BinaryIO] + ) -> concurrent.futures.Future[str]: + """ + Uploads an audio file which can be specified as local path or binary object. + + Args: + `data`: A local file (as path), or a binary object. + + Returns: The URL of the uploaded audio file. + """ + return self._executor.submit( + self._impl.upload_file, + data=data, + ) + + def submit( + self, + data: Union[str, bytes, BinaryIO], + config: Optional[types.TranscriptionConfig] = None, + ) -> Transcript: + """ + Submits a transcription job without waiting for its completion. + + Args: + data: An URL, a local file (as path), raw `bytes`, or a binary object. + config: Transcription options and features. If `None` is given, the Transcriber's + default configuration will be used. + """ + return self._impl.transcribe( + data=data, + config=config, + poll=False, + ) + + def submit_group( + self, + data: List[Union[str, bytes, BinaryIO]], + config: Optional[types.TranscriptionConfig] = None, + return_failures: Optional[bool] = False, + ) -> Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]]: + """ + Submits multiple transcription jobs without waiting for their completion. + + Args: + data: A list of local paths, URLs, raw `bytes`, or binary objects (can be mixed). + config: Transcription options and features. If `None` is given, the Transcriber's + default configuration will be used. + return_failures: Whether to include a list of errors for transcriptions that failed due to HTTP errors + """ + return self._impl.transcribe_group( + data=data, + config=config, + poll=False, + return_failures=return_failures, + ) + + def transcribe( + self, + data: Union[str, bytes, BinaryIO], + config: Optional[types.TranscriptionConfig] = None, + ) -> Transcript: + """ + Transcribes an audio file which can be specified as local path, URL, raw `bytes`, or binary object. + + Args: + data: An URL, a local file (as path), raw `bytes`, or a binary object. + config: Transcription options and features. If `None` is given, the Transcriber's + default configuration will be used. + """ + + return self._impl.transcribe( + data=data, + config=config, + poll=True, + ) + + def transcribe_async( + self, + data: Union[str, bytes, BinaryIO], + config: Optional[types.TranscriptionConfig] = None, + ) -> concurrent.futures.Future[Transcript]: + """ + Transcribes an audio file which can be specified as local path, URL, raw `bytes`, or binary object. + + Args: + data: An URL, a local file (as path), raw `bytes`, or a binary object. + config: Transcription options and features. If `None` is given, the Transcriber's + default configuration will be used. + """ + + return self._executor.submit( + self._impl.transcribe, + data=data, + config=config, + poll=True, + ) + + def transcribe_group( + self, + data: List[Union[str, bytes, BinaryIO]], + config: Optional[types.TranscriptionConfig] = None, + return_failures: Optional[bool] = False, + ) -> Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]]: + """ + Transcribes a list of files (as local paths, URLs, or binary objects). + + Args: + data: A list of local paths, URLs, raw `bytes`, or binary objects (can be mixed). + config: Transcription options and features. If `None` is given, the Transcriber's + default configuration will be used. + return_failures: Whether to include a list of errors for transcriptions that failed due to HTTP errors + """ + + return self._impl.transcribe_group( + data=data, + config=config, + poll=True, + return_failures=return_failures, + ) + + def transcribe_group_async( + self, + data: List[Union[str, bytes, BinaryIO]], + config: Optional[types.TranscriptionConfig] = None, + return_failures: Optional[bool] = False, + ) -> concurrent.futures.Future[ + Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]] + ]: + """ + Transcribes a list of files (as local paths, URLs, or binary objects) asynchronously. + + Args: + data: A list of local paths, URLs, raw `bytes`, or binary objects (can be mixed). + config: Transcription options and features. If `None` is given, the Transcriber's + default configuration will be used. + return_failures: Whether to include a list of errors for transcriptions that failed due to HTTP errors + """ + + return self._executor.submit( + self._impl.transcribe_group, + data=data, + config=config, + poll=True, + return_failures=return_failures, + ) + + def list_transcripts( + self, + params: Optional[types.ListTranscriptParameters] = None, + ) -> types.ListTranscriptResponse: + """ + Retrieve a list of transcripts that were created. Transcripts are sorted from newest to oldest. + + Args: + params: The parameters to filter the transcript list by. + + Returns: A page with a list of transcripts along with page details. + + To paginate over all pages, you can set the `ListTranscriptParameters.before_id` + to the `before_id` of the `prev_url`. Example: + ``` + transcriber = aai.Transcriber() + params = aai.ListTranscriptParameters() + page = transcriber.list_transcripts(params) + while page.page_details.before_id_of_prev_url is not None: + params.before_id = page.page_details.before_id_of_prev_url + page = transcriber.list_transcripts(params) + ``` + """ + return self._impl.list_transcripts(params=params) + + def list_transcripts_async( + self, + params: Optional[types.ListTranscriptParameters] = None, + ) -> concurrent.futures.Future[types.ListTranscriptResponse]: + """ + Retrieve a list of transcripts that were created. Transcripts are sorted from newest to oldest. + + Args: + params: The parameters to filter the transcript list by. + + Returns: A page with a list of transcripts along with page details. + """ + return self._executor.submit(self._impl.list_transcripts, params=params) diff --git a/assemblyai/prerecorded/v2/transcript.py b/assemblyai/prerecorded/v2/transcript.py new file mode 100644 index 0000000..1ad7d4b --- /dev/null +++ b/assemblyai/prerecorded/v2/transcript.py @@ -0,0 +1,651 @@ +"""The ``Transcript`` result object for prerecorded transcription.""" + +from __future__ import annotations + +import concurrent.futures +import functools +import time +from typing import Dict, List, Optional, Union + +import httpx +from typing_extensions import Self + +from ... import client as _client +from ... import lemur, types +from . import api + + +class _TranscriptImpl: + def __init__( + self, + *, + client: _client.Client, + transcript_id: Optional[str], + ) -> None: + self._client = client + self.transcript_id = transcript_id + + self.transcript: Optional[types.TranscriptResponse] = None + + @property + def config(self) -> types.TranscriptionConfig: + "Returns the configuration from the internal Transcript object" + if self.transcript is None: + raise ValueError( + "Cannot access the configuration. The internal Transcript object is None." + ) + + return types.TranscriptionConfig( + **self.transcript.dict( + include=set(types.RawTranscriptionConfig.__fields__), + exclude_none=True, + ) + ) + + @classmethod + def from_response( + cls, + *, + client: _client.Client, + response: types.TranscriptResponse, + ) -> Self: + self = cls( + client=client, + transcript_id=response.id, + ) + self.transcript = response + + return self + + def wait_for_completion(self) -> Self: + """ + polls the given transcript until we have a status other than `processing` or `queued` + """ + if not self.transcript_id: + raise ValueError( + "Cannot wait for completion. The internal transcript ID is None." + ) + + while True: + # No try-except - if there is an HTTP error then surface it to user + self.transcript = api.get_transcript( + self._client.http_client, + self.transcript_id, + ) + + if self.transcript.status in ( + types.TranscriptStatus.completed, + types.TranscriptStatus.error, + ): + break + + time.sleep(self._client.settings.polling_interval) + + return self + + def export_subtitles_srt( + self, + *, + chars_per_caption: Optional[int], + ) -> str: + if not self.transcript or not self.transcript.id: + raise ValueError( + "Cannot export subtitles. The internal Transcript object is None." + ) + + return api.export_subtitles_srt( + client=self._client.http_client, + transcript_id=self.transcript.id, + chars_per_caption=chars_per_caption, + ) + + def export_subtitles_vtt( + self, + *, + chars_per_caption: Optional[int], + ) -> str: + if not self.transcript or not self.transcript.id: + raise ValueError( + "Cannot export subtitles. The internal Transcript object is None." + ) + + return api.export_subtitles_vtt( + client=self._client.http_client, + transcript_id=self.transcript.id, + chars_per_caption=chars_per_caption, + ) + + def word_search( + self, + *, + words: List[str], + ) -> List[types.WordSearchMatch]: + if not self.transcript or not self.transcript.id: + raise ValueError( + "Cannot perform word search. The internal Transcript object is None." + ) + + response = api.word_search( + client=self._client.http_client, + transcript_id=self.transcript.id, + words=words, + ) + + return response.matches + + def get_sentences(self) -> List[types.Sentence]: + if not self.transcript or not self.transcript.id: + raise ValueError( + "Cannot get sentences. The internal Transcript object is None." + ) + + response = api.get_sentences( + client=self._client.http_client, + transcript_id=self.transcript.id, + ) + + return response.sentences + + def get_paragraphs(self) -> List[types.Paragraph]: + if not self.transcript or not self.transcript.id: + raise ValueError( + "Cannot get paragraphs. The internal Transcript object is None." + ) + + response = api.get_paragraphs( + client=self._client.http_client, + transcript_id=self.transcript.id, + ) + + return response.paragraphs + + @functools.lru_cache + def get_redacted_audio_url(self) -> str: + """ + Retrieve the URL for the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig`. + Subsequent calls will return cached URL rather than requesting it from the API again. + + Returns: The URL of the redacted audio file. + """ + if not self.config.redact_pii or not self.config.redact_pii_audio: + raise ValueError( + "Redacted audio is only available when `redact_pii` and `redact_pii_audio` are set to `True`." + ) + + if not self.transcript_id: + raise ValueError( + "Cannot get redacted audio url. The internal transcript ID is None." + ) + + while True: + try: + return api.get_redacted_audio( + client=self._client.http_client, + transcript_id=self.transcript_id, + ).redacted_audio_url + except types.RedactedAudioIncompleteError: + time.sleep(self._client.settings.polling_interval) + + def save_redacted_audio(self, filepath: str): + """ + Retrieve the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig` + + Args: + filepath: The path to save the redacted audio file to. + """ + with httpx.stream(method="GET", url=self.get_redacted_audio_url()) as response: + if response.status_code not in (httpx.codes.OK, httpx.codes.NOT_MODIFIED): + raise types.RedactedAudioUnavailableError( + f"Fetching redacted audio failed with status code {response.status_code}", + response.status_code, + ) + with open(filepath, "wb") as f: + for chunk in response.iter_bytes(): + f.write(chunk) + + @classmethod + def delete_by_id(cls, transcript_id: str) -> types.Transcript: + client = _client.Client.get_default() + response = api.delete_transcript( + client=client.http_client, transcript_id=transcript_id + ) + + return Transcript.from_response(client=client, response=response) + + +class Transcript(types.Sourcable): + """ + Transcript object to perform operations on the actual transcript. + """ + + def __init__( + self, + transcript_id: Optional[str], + client: Optional[_client.Client] = None, + ) -> None: + self._client = client or _client.Client.get_default() + + self._impl = _TranscriptImpl( + client=self._client, + transcript_id=transcript_id, + ) + self._executor = concurrent.futures.ThreadPoolExecutor() + + def wait_for_completion(self) -> Self: + self._impl.wait_for_completion() + + return self + + def wait_for_completion_async( + self, + ) -> concurrent.futures.Future[Self]: + return self._executor.submit(self.wait_for_completion) + + @classmethod + def from_response( + cls, + *, + client: _client.Client, + response: types.TranscriptResponse, + ) -> Self: + _impl = _TranscriptImpl.from_response(client=client, response=response) + + self = cls( + client=client, + transcript_id=response.id, + ) + + self._impl = _impl + + return self + + @classmethod + def get_by_id(cls, transcript_id: str) -> Self: + """Fetch an existing transcript. Blocks until the transcript is completed. + + Args: + transcript_id: the id of the transcript to fetch + + Returns: + The transcript object identified by the given id. + """ + return cls(transcript_id=transcript_id).wait_for_completion() + + @classmethod + def get_by_id_async(cls, transcript_id: str) -> concurrent.futures.Future[Self]: + """Fetch an existing transcript asynchronously. + + Args: + transcript_id: the id of the transcript to fetch + + Returns: + A future that will resolve to the transcript object identified by the given id. + """ + return cls(transcript_id=transcript_id).wait_for_completion_async() + + @classmethod + def delete_by_id(cls, transcript_id: str) -> types.Transcript: + """Delete an existing transcript. Blocks until the transcript is completed. + + Args: + transcript_id: the id of the transcript to delete + + Returns: + A transcript object identified by the given id, with relevant fields/attributes cleared. + """ + return _TranscriptImpl.delete_by_id(transcript_id) + + @classmethod + def delete_by_id_async( + cls, transcript_id: str + ) -> concurrent.futures.Future[types.Transcript]: + """Delete an existing transcript asynchronously. + + Args: + transcript_id: the id of the transcript to delete + + Returns: + A future that will resolve to a transcript object identified by the given id, with relevant fields/attributes cleared. + """ + + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + future_transcript = executor.submit( + _TranscriptImpl.delete_by_id, transcript_id + ) + return future_transcript + + @property + def id(self) -> Optional[str]: + "The unique identifier of your transcription" + + return self._impl.transcript_id + + @property + def config(self) -> types.TranscriptionConfig: + "Return the corresponding configurations for the given transcript." + + return self._impl.config + + @property + def json_response(self) -> Optional[dict]: + "The full JSON response associated with the transcript." + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.dict() + + @property + def audio_url(self) -> str: + "The corresponding audio url" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.audio_url + + @property + def speech_model(self) -> Optional[str]: + "The speech model used for the transcription" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.speech_model + + @property + def speech_model_used(self) -> Optional[str]: + "The actual speech model that was used for the transcription" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.speech_model_used + + @property + def text(self) -> Optional[str]: + "The text transcription of your media file" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.text + + @property + def translated_texts(self) -> Optional[Dict[str, str]]: + "The translated texts transcription of your media file" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.translated_texts + + @property + def speech_understanding(self) -> Optional[types.SpeechUnderstandingResponse]: + "The text transcription of your media file" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.speech_understanding + + @property + def summary(self) -> Optional[str]: + "The summarization of the transcript" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.summary + + @property + def chapters(self) -> Optional[List[types.Chapter]]: + "The list of auto-chapters results" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.chapters + + @property + def content_safety(self) -> Optional[types.ContentSafetyResponse]: + "The results from the content safety analysis" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.content_safety_labels + + @property + def sentiment_analysis(self) -> Optional[List[types.Sentiment]]: + "The list of sentiment analysis results" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.sentiment_analysis_results + + @property + def entities(self) -> Optional[List[types.Entity]]: + "The list of entity detection results" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.entities + + @property + def iab_categories(self) -> Optional[types.IABResponse]: + "The results from the IAB category detection" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.iab_categories_result + + @property + def auto_highlights(self) -> Optional[types.AutohighlightResponse]: + "The results from the auto-highlights model" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.auto_highlights_result + + @property + def status(self) -> types.TranscriptStatus: + "The current status of the transcript" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.status + + @property + def error(self) -> Optional[str]: + "The error message in case the transcription fails" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.error + + @property + def words(self) -> Optional[List[types.Word]]: + "The list of words in the transcript" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.words + + @property + def utterances(self) -> Optional[List[types.Utterance]]: + """ + When `dual_channel` or `speaker_labels` is enabled, + a list of utterances in the transcript. + """ + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.utterances + + @property + def unredacted_text(self) -> Optional[str]: + "The unredacted transcript text, when `redact_pii_return_unredacted` was enabled." + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.unredacted_text + + @property + def unredacted_words(self) -> Optional[List[types.Word]]: + "The unredacted list of words, when `redact_pii_return_unredacted` was enabled." + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.unredacted_words + + @property + def unredacted_utterances(self) -> Optional[List[types.Utterance]]: + "The unredacted list of utterances, when `redact_pii_return_unredacted` was enabled." + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.unredacted_utterances + + @property + def confidence(self) -> Optional[float]: + "The confidence our model has in the transcribed text, between 0 and 1" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.confidence + + @property + def audio_duration(self) -> Optional[int]: + "The duration of the audio in seconds" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.audio_duration + + @property + def webhook_status_code(self) -> Optional[int]: + "The status code we received from your server when delivering your webhook" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.webhook_status_code + + @property + def webhook_auth(self) -> Optional[bool]: + "Whether the webhook was sent with an HTTP authentication header" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.webhook_auth + + @property + def language_code(self) -> Optional[Union[str, types.LanguageCode]]: + "The language code of the transcript" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.language_code + + @property + def language_codes(self) -> Optional[List[Union[str, types.LanguageCode]]]: + "The list of language codes for multilingual/code-switching audio" + if not self._impl.transcript: + raise ValueError("The internal Transcript object is None.") + + return self._impl.transcript.language_codes + + @property + def lemur(self) -> lemur.Lemur: + """ + Access AssemblyAI's LeMUR features. + """ + + return lemur.Lemur( + client=self._client, + sources=[types.LemurSource(self)], + ) + + def export_subtitles_srt( + self, + chars_per_caption: Optional[int] = None, + ) -> str: + """ + You can export your complete transcripts in SRT format, + to be plugged into a video player for subtitles and closed captions. + + Args: + chars_per_caption: To control the maximum number of characters per caption + + Returns: A string containing the all subtitles in SRT format. + """ + + return self._impl.export_subtitles_srt( + chars_per_caption=chars_per_caption, + ) + + def export_subtitles_vtt( + self, + chars_per_caption: Optional[int] = None, + ) -> str: + """ + You can export your complete transcripts in VTT format, + to be plugged into a video player for subtitles and closed captions. + + Args: + chars_per_caption: To control the maximum number of characters per caption + + Returns: A string containing the all subtitles in VTT format. + """ + + return self._impl.export_subtitles_vtt( + chars_per_caption=chars_per_caption, + ) + + def word_search( + self, + words: List[str], + ) -> List[types.WordSearchMatch]: + """ + Once a transcript has been completed, you can search through the transcript for a specific set of keywords. + You can search for individual words, numbers, or phrases containing up to five words or numbers. + + Args: + words: A list of words, numbers, or phrases (containing up to five words or numbers) + + Returns: A list of matches + """ + + return self._impl.word_search( + words=words, + ) + + def get_sentences( + self, + ) -> List[types.Sentence]: + """ + Semantically segment your transcript into sentences to create more reader-friendly transcripts. + + Returns: A list of sentence objects. + """ + + return self._impl.get_sentences() + + def get_paragraphs( + self, + ) -> List[types.Paragraph]: + """ + Semantically segment your transcript into paragraphs to create more reader-friendly transcripts. + + Returns: A list of paragraph objects. + """ + + return self._impl.get_paragraphs() + + def get_redacted_audio_url(self) -> str: + """ + Retrieve the URL for the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig`. + Subsequent calls will return cached URL rather than requesting it from the API again. + + Returns: The URL of the redacted audio file. + """ + return self._impl.get_redacted_audio_url() + + def save_redacted_audio(self, filepath: str): + """ + Retrieve the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig` + + Args: + filepath: The path to save the redacted audio file to. + """ + return self._impl.save_redacted_audio(filepath=filepath) diff --git a/assemblyai/prerecorded/v2/transcript_group.py b/assemblyai/prerecorded/v2/transcript_group.py new file mode 100644 index 0000000..e44f3e7 --- /dev/null +++ b/assemblyai/prerecorded/v2/transcript_group.py @@ -0,0 +1,202 @@ +"""The ``TranscriptGroup`` collection over several prerecorded transcripts.""" + +from __future__ import annotations + +import concurrent.futures +from typing import Iterator, List, Optional, Set, Tuple, Union + +from typing_extensions import Self + +from ... import client as _client +from ... import lemur, types +from .transcript import Transcript + + +class _TranscriptGroupImpl: + def __init__( + self, + *, + transcript_ids: List[str], + client: _client.Client, + ) -> None: + self._client = client + self.transcripts: List[Transcript] = [] + + for transcript_id in transcript_ids: + self.add_transcript(transcript_id) + + @property + def transcript_ids(self) -> List[str]: + if any(t.id is None for t in self.transcripts): + raise ValueError("All transcripts must have a transcript ID.") + return [ + t.id for t in self.transcripts if t.id + ] # include the if check for mypy type checker + + def add_transcript(self, transcript: Union[Transcript, str]) -> None: + if isinstance(transcript, Transcript): + self.transcripts.append(transcript) + elif isinstance(transcript, str): + self.transcripts.append( + Transcript( + client=self._client, + transcript_id=transcript, + ) + ) + else: + raise TypeError("Unsupported type for `transcript`") + + def wait_for_completion( + self, return_failures + ) -> Union[None, List[types.AssemblyAIError]]: + transcripts: List[Transcript] = [] + failures: List[types.AssemblyAIError] = [] + + future_transcripts: Set[concurrent.futures.Future[Transcript]] = set() + + for transcript in self.transcripts: + future = transcript.wait_for_completion_async() + future_transcripts.add(future) + + finished_futures, _ = concurrent.futures.wait(future_transcripts) + + for future in finished_futures: + try: + transcripts.append(future.result()) + except types.TranscriptError as e: + failures.append(e) + + self.transcripts = transcripts + + if return_failures is True: + return failures + return None + + +class TranscriptGroup: + """ + A group of transcripts. + + Used when transcribing multiple transcripts at once. + """ + + def __init__( + self, + transcript_ids: List[str] = [], + client: Optional[_client.Client] = None, + ) -> None: + self._client = client or _client.Client.get_default() + + self._impl = _TranscriptGroupImpl( + transcript_ids=transcript_ids, + client=self._client, + ) + self._executor = concurrent.futures.ThreadPoolExecutor() + + @property + def transcripts(self) -> List[Transcript]: + """ + Returns the list of the transcripts within the `TranscriptGroup` + """ + + return self._impl.transcripts + + def __iter__(self) -> Iterator[Transcript]: + """ + Iterate over the transcripts within the `TranscriptGroup` + """ + + return iter(self.transcripts) + + @classmethod + def get_by_ids( + cls, transcript_ids: List[str] + ) -> Union[Self, Tuple[Self, List[types.AssemblyAIError]]]: + return cls(transcript_ids=transcript_ids).wait_for_completion() + + @classmethod + def get_by_ids_async( + cls, transcript_ids: List[str] + ) -> concurrent.futures.Future[ + Union[Self, Tuple[Self, List[types.AssemblyAIError]]] + ]: + return cls(transcript_ids=transcript_ids).wait_for_completion_async() + + @property + def status(self) -> types.TranscriptStatus: + """ + Return the status of the `TranscriptGroup`. + + e.g. if any of the transcripts is in `error` status, the whole `TranscriptGroup` will be in `error` status. + """ + + all_status = {t.status for t in self.transcripts} + + if any(s == types.TranscriptStatus.error for s in all_status): + return types.TranscriptStatus.error + elif any(s == types.TranscriptStatus.queued for s in all_status): + return types.TranscriptStatus.queued + elif any(s == types.TranscriptStatus.processing for s in all_status): + return types.TranscriptStatus.processing + elif all(s == types.TranscriptStatus.completed for s in all_status): + return types.TranscriptStatus.completed + else: + raise ValueError(f"Unexpected status type: {all_status}") + + @property + def lemur(self) -> lemur.Lemur: + """ + Access AssemblyAI's LeMUR functionality. + """ + + return lemur.Lemur( + client=self._impl._client, + sources=[types.LemurSource(t) for t in self.transcripts], + ) + + def add_transcript( + self, + transcript: Union[Transcript, str], + ) -> Self: + """ + Adds a transcript to the given `TranscriptGroup` + + Args: + transcript: A `Transcript` object or the ID as a `str` + """ + self._impl.add_transcript(transcript) + + return self + + def wait_for_completion( + self, + return_failures: Optional[bool] = False, + ) -> Union[Self, Tuple[Self, List[types.AssemblyAIError]]]: + """ + Polls each transcript within the `TranscriptGroup`. + + Note - if an HTTP error is encountered when waiting for a Transcript in the TranscriptGroup, it will be popped from the group and added to the list of failures. + You can return this list of failures with `return_failures=True`. + + Args: + return_failures: Whether to return a list of errors for transcripts that failed due to HTTP errors. + """ + if return_failures is True: + failures = self._impl.wait_for_completion(return_failures=return_failures) + if failures is None: + raise ValueError("return_failures was set but failures object is None") + return self, failures + + self._impl.wait_for_completion(return_failures=return_failures) + + return self + + def wait_for_completion_async( + self, + return_failures: Optional[bool] = False, + ) -> concurrent.futures.Future[ + Union[Self, Tuple[Self, List[types.AssemblyAIError]]], + ]: + return self._executor.submit( + self.wait_for_completion, return_failures=return_failures + ) diff --git a/assemblyai/transcriber.py b/assemblyai/transcriber.py index 43d766d..3c8b54d 100644 --- a/assemblyai/transcriber.py +++ b/assemblyai/transcriber.py @@ -1,1273 +1,17 @@ -from __future__ import annotations +"""Backwards-compatible re-exports of the prerecorded transcription surface. -import concurrent.futures -import functools -import os -import time -from typing import ( - BinaryIO, - Dict, - Iterator, - List, - Optional, - Set, - Tuple, - Union, -) -from urllib.parse import urlparse - -import httpx -from typing_extensions import Self - -from . import api, lemur, types -from . import client as _client - - -class _TranscriptImpl: - def __init__( - self, - *, - client: _client.Client, - transcript_id: Optional[str], - ) -> None: - self._client = client - self.transcript_id = transcript_id - - self.transcript: Optional[types.TranscriptResponse] = None - - @property - def config(self) -> types.TranscriptionConfig: - "Returns the configuration from the internal Transcript object" - if self.transcript is None: - raise ValueError( - "Cannot access the configuration. The internal Transcript object is None." - ) - - return types.TranscriptionConfig( - **self.transcript.dict( - include=set(types.RawTranscriptionConfig.__fields__), - exclude_none=True, - ) - ) - - @classmethod - def from_response( - cls, - *, - client: _client.Client, - response: types.TranscriptResponse, - ) -> Self: - self = cls( - client=client, - transcript_id=response.id, - ) - self.transcript = response - - return self - - def wait_for_completion(self) -> Self: - """ - polls the given transcript until we have a status other than `processing` or `queued` - """ - if not self.transcript_id: - raise ValueError( - "Cannot wait for completion. The internal transcript ID is None." - ) - - while True: - # No try-except - if there is an HTTP error then surface it to user - self.transcript = api.get_transcript( - self._client.http_client, - self.transcript_id, - ) - - if self.transcript.status in ( - types.TranscriptStatus.completed, - types.TranscriptStatus.error, - ): - break - - time.sleep(self._client.settings.polling_interval) - - return self - - def export_subtitles_srt( - self, - *, - chars_per_caption: Optional[int], - ) -> str: - if not self.transcript or not self.transcript.id: - raise ValueError( - "Cannot export subtitles. The internal Transcript object is None." - ) - - return api.export_subtitles_srt( - client=self._client.http_client, - transcript_id=self.transcript.id, - chars_per_caption=chars_per_caption, - ) - - def export_subtitles_vtt( - self, - *, - chars_per_caption: Optional[int], - ) -> str: - if not self.transcript or not self.transcript.id: - raise ValueError( - "Cannot export subtitles. The internal Transcript object is None." - ) - - return api.export_subtitles_vtt( - client=self._client.http_client, - transcript_id=self.transcript.id, - chars_per_caption=chars_per_caption, - ) - - def word_search( - self, - *, - words: List[str], - ) -> List[types.WordSearchMatch]: - if not self.transcript or not self.transcript.id: - raise ValueError( - "Cannot perform word search. The internal Transcript object is None." - ) - - response = api.word_search( - client=self._client.http_client, - transcript_id=self.transcript.id, - words=words, - ) - - return response.matches - - def get_sentences(self) -> List[types.Sentence]: - if not self.transcript or not self.transcript.id: - raise ValueError( - "Cannot get sentences. The internal Transcript object is None." - ) - - response = api.get_sentences( - client=self._client.http_client, - transcript_id=self.transcript.id, - ) - - return response.sentences - - def get_paragraphs(self) -> List[types.Paragraph]: - if not self.transcript or not self.transcript.id: - raise ValueError( - "Cannot get paragraphs. The internal Transcript object is None." - ) - - response = api.get_paragraphs( - client=self._client.http_client, - transcript_id=self.transcript.id, - ) - - return response.paragraphs - - @functools.lru_cache - def get_redacted_audio_url(self) -> str: - """ - Retrieve the URL for the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig`. - Subsequent calls will return cached URL rather than requesting it from the API again. - - Returns: The URL of the redacted audio file. - """ - if not self.config.redact_pii or not self.config.redact_pii_audio: - raise ValueError( - "Redacted audio is only available when `redact_pii` and `redact_pii_audio` are set to `True`." - ) - - if not self.transcript_id: - raise ValueError( - "Cannot get redacted audio url. The internal transcript ID is None." - ) - - while True: - try: - return api.get_redacted_audio( - client=self._client.http_client, - transcript_id=self.transcript_id, - ).redacted_audio_url - except types.RedactedAudioIncompleteError: - time.sleep(self._client.settings.polling_interval) - - def save_redacted_audio(self, filepath: str): - """ - Retrieve the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig` - - Args: - filepath: The path to save the redacted audio file to. - """ - with httpx.stream(method="GET", url=self.get_redacted_audio_url()) as response: - if response.status_code not in (httpx.codes.OK, httpx.codes.NOT_MODIFIED): - raise types.RedactedAudioUnavailableError( - f"Fetching redacted audio failed with status code {response.status_code}", - response.status_code, - ) - with open(filepath, "wb") as f: - for chunk in response.iter_bytes(): - f.write(chunk) - - @classmethod - def delete_by_id(cls, transcript_id: str) -> types.Transcript: - client = _client.Client.get_default() - response = api.delete_transcript( - client=client.http_client, transcript_id=transcript_id - ) - - return Transcript.from_response(client=client, response=response) - - -class Transcript(types.Sourcable): - """ - Transcript object to perform operations on the actual transcript. - """ - - def __init__( - self, - transcript_id: Optional[str], - client: Optional[_client.Client] = None, - ) -> None: - self._client = client or _client.Client.get_default() - - self._impl = _TranscriptImpl( - client=self._client, - transcript_id=transcript_id, - ) - self._executor = concurrent.futures.ThreadPoolExecutor() - - def wait_for_completion(self) -> Self: - self._impl.wait_for_completion() - - return self - - def wait_for_completion_async( - self, - ) -> concurrent.futures.Future[Self]: - return self._executor.submit(self.wait_for_completion) - - @classmethod - def from_response( - cls, - *, - client: _client.Client, - response: types.TranscriptResponse, - ) -> Self: - _impl = _TranscriptImpl.from_response(client=client, response=response) - - self = cls( - client=client, - transcript_id=response.id, - ) - - self._impl = _impl - - return self - - @classmethod - def get_by_id(cls, transcript_id: str) -> Self: - """Fetch an existing transcript. Blocks until the transcript is completed. - - Args: - transcript_id: the id of the transcript to fetch - - Returns: - The transcript object identified by the given id. - """ - return cls(transcript_id=transcript_id).wait_for_completion() - - @classmethod - def get_by_id_async(cls, transcript_id: str) -> concurrent.futures.Future[Self]: - """Fetch an existing transcript asynchronously. - - Args: - transcript_id: the id of the transcript to fetch - - Returns: - A future that will resolve to the transcript object identified by the given id. - """ - return cls(transcript_id=transcript_id).wait_for_completion_async() - - @classmethod - def delete_by_id(cls, transcript_id: str) -> types.Transcript: - """Delete an existing transcript. Blocks until the transcript is completed. - - Args: - transcript_id: the id of the transcript to delete - - Returns: - A transcript object identified by the given id, with relevant fields/attributes cleared. - """ - return _TranscriptImpl.delete_by_id(transcript_id) - - @classmethod - def delete_by_id_async( - cls, transcript_id: str - ) -> concurrent.futures.Future[types.Transcript]: - """Delete an existing transcript asynchronously. - - Args: - transcript_id: the id of the transcript to delete - - Returns: - A future that will resolve to a transcript object identified by the given id, with relevant fields/attributes cleared. - """ - - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: - future_transcript = executor.submit( - _TranscriptImpl.delete_by_id, transcript_id - ) - return future_transcript - - @property - def id(self) -> Optional[str]: - "The unique identifier of your transcription" - - return self._impl.transcript_id - - @property - def config(self) -> types.TranscriptionConfig: - "Return the corresponding configurations for the given transcript." - - return self._impl.config - - @property - def json_response(self) -> Optional[dict]: - "The full JSON response associated with the transcript." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.dict() - - @property - def audio_url(self) -> str: - "The corresponding audio url" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.audio_url - - @property - def speech_model(self) -> Optional[str]: - "The speech model used for the transcription" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.speech_model - - @property - def speech_model_used(self) -> Optional[str]: - "The actual speech model that was used for the transcription" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.speech_model_used - - @property - def text(self) -> Optional[str]: - "The text transcription of your media file" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.text - - @property - def translated_texts(self) -> Optional[Dict[str, str]]: - "The translated texts transcription of your media file" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.translated_texts - - @property - def speech_understanding(self) -> Optional[types.SpeechUnderstandingResponse]: - "The text transcription of your media file" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.speech_understanding - - @property - def summary(self) -> Optional[str]: - "The summarization of the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.summary - - @property - def chapters(self) -> Optional[List[types.Chapter]]: - "The list of auto-chapters results" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.chapters - - @property - def content_safety(self) -> Optional[types.ContentSafetyResponse]: - "The results from the content safety analysis" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.content_safety_labels - - @property - def sentiment_analysis(self) -> Optional[List[types.Sentiment]]: - "The list of sentiment analysis results" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.sentiment_analysis_results - - @property - def entities(self) -> Optional[List[types.Entity]]: - "The list of entity detection results" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.entities - - @property - def iab_categories(self) -> Optional[types.IABResponse]: - "The results from the IAB category detection" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.iab_categories_result - - @property - def auto_highlights(self) -> Optional[types.AutohighlightResponse]: - "The results from the auto-highlights model" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.auto_highlights_result - - @property - def status(self) -> types.TranscriptStatus: - "The current status of the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.status - - @property - def error(self) -> Optional[str]: - "The error message in case the transcription fails" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.error - - @property - def words(self) -> Optional[List[types.Word]]: - "The list of words in the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.words - - @property - def utterances(self) -> Optional[List[types.Utterance]]: - """ - When `dual_channel` or `speaker_labels` is enabled, - a list of utterances in the transcript. - """ - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.utterances - - @property - def unredacted_text(self) -> Optional[str]: - "The unredacted transcript text, when `redact_pii_return_unredacted` was enabled." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.unredacted_text - - @property - def unredacted_words(self) -> Optional[List[types.Word]]: - "The unredacted list of words, when `redact_pii_return_unredacted` was enabled." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.unredacted_words - - @property - def unredacted_utterances(self) -> Optional[List[types.Utterance]]: - "The unredacted list of utterances, when `redact_pii_return_unredacted` was enabled." - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.unredacted_utterances - - @property - def confidence(self) -> Optional[float]: - "The confidence our model has in the transcribed text, between 0 and 1" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.confidence - - @property - def audio_duration(self) -> Optional[int]: - "The duration of the audio in seconds" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.audio_duration - - @property - def webhook_status_code(self) -> Optional[int]: - "The status code we received from your server when delivering your webhook" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.webhook_status_code - - @property - def webhook_auth(self) -> Optional[bool]: - "Whether the webhook was sent with an HTTP authentication header" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.webhook_auth - - @property - def language_code(self) -> Optional[Union[str, types.LanguageCode]]: - "The language code of the transcript" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.language_code - - @property - def language_codes(self) -> Optional[List[Union[str, types.LanguageCode]]]: - "The list of language codes for multilingual/code-switching audio" - if not self._impl.transcript: - raise ValueError("The internal Transcript object is None.") - - return self._impl.transcript.language_codes - - @property - def lemur(self) -> lemur.Lemur: - """ - Access AssemblyAI's LeMUR features. - """ - - return lemur.Lemur( - client=self._client, - sources=[types.LemurSource(self)], - ) - - def export_subtitles_srt( - self, - chars_per_caption: Optional[int] = None, - ) -> str: - """ - You can export your complete transcripts in SRT format, - to be plugged into a video player for subtitles and closed captions. - - Args: - chars_per_caption: To control the maximum number of characters per caption +The canonical location is ``assemblyai.prerecorded.v2``. +""" - Returns: A string containing the all subtitles in SRT format. - """ - - return self._impl.export_subtitles_srt( - chars_per_caption=chars_per_caption, - ) - - def export_subtitles_vtt( - self, - chars_per_caption: Optional[int] = None, - ) -> str: - """ - You can export your complete transcripts in VTT format, - to be plugged into a video player for subtitles and closed captions. - - Args: - chars_per_caption: To control the maximum number of characters per caption - - Returns: A string containing the all subtitles in VTT format. - """ - - return self._impl.export_subtitles_vtt( - chars_per_caption=chars_per_caption, - ) - - def word_search( - self, - words: List[str], - ) -> List[types.WordSearchMatch]: - """ - Once a transcript has been completed, you can search through the transcript for a specific set of keywords. - You can search for individual words, numbers, or phrases containing up to five words or numbers. - - Args: - words: A list of words, numbers, or phrases (containing up to five words or numbers) - - Returns: A list of matches - """ - - return self._impl.word_search( - words=words, - ) - - def get_sentences( - self, - ) -> List[types.Sentence]: - """ - Semantically segment your transcript into sentences to create more reader-friendly transcripts. - - Returns: A list of sentence objects. - """ - - return self._impl.get_sentences() - - def get_paragraphs( - self, - ) -> List[types.Paragraph]: - """ - Semantically segment your transcript into paragraphs to create more reader-friendly transcripts. - - Returns: A list of paragraph objects. - """ - - return self._impl.get_paragraphs() - - def get_redacted_audio_url(self) -> str: - """ - Retrieve the URL for the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig`. - Subsequent calls will return cached URL rather than requesting it from the API again. - - Returns: The URL of the redacted audio file. - """ - return self._impl.get_redacted_audio_url() - - def save_redacted_audio(self, filepath: str): - """ - Retrieve the PII-redacted audio file, if `redact_pii_audio` was enabled on the `TranscriptionConfig` - - Args: - filepath: The path to save the redacted audio file to. - """ - return self._impl.save_redacted_audio(filepath=filepath) - - -class _TranscriptGroupImpl: - def __init__( - self, - *, - transcript_ids: List[str], - client: _client.Client, - ) -> None: - self._client = client - self.transcripts: List[Transcript] = [] - - for transcript_id in transcript_ids: - self.add_transcript(transcript_id) - - @property - def transcript_ids(self) -> List[str]: - if any(t.id is None for t in self.transcripts): - raise ValueError("All transcripts must have a transcript ID.") - return [ - t.id for t in self.transcripts if t.id - ] # include the if check for mypy type checker - - def add_transcript(self, transcript: Union[Transcript, str]) -> None: - if isinstance(transcript, Transcript): - self.transcripts.append(transcript) - elif isinstance(transcript, str): - self.transcripts.append( - Transcript( - client=self._client, - transcript_id=transcript, - ) - ) - else: - raise TypeError("Unsupported type for `transcript`") - - def wait_for_completion( - self, return_failures - ) -> Union[None, List[types.AssemblyAIError]]: - transcripts: List[Transcript] = [] - failures: List[types.AssemblyAIError] = [] - - future_transcripts: Set[concurrent.futures.Future[Transcript]] = set() - - for transcript in self.transcripts: - future = transcript.wait_for_completion_async() - future_transcripts.add(future) - - finished_futures, _ = concurrent.futures.wait(future_transcripts) - - for future in finished_futures: - try: - transcripts.append(future.result()) - except types.TranscriptError as e: - failures.append(e) - - self.transcripts = transcripts - - if return_failures is True: - return failures - return None - - -class TranscriptGroup: - """ - A group of transcripts. - - Used when transcribing multiple transcripts at once. - """ - - def __init__( - self, - transcript_ids: List[str] = [], - client: Optional[_client.Client] = None, - ) -> None: - self._client = client or _client.Client.get_default() - - self._impl = _TranscriptGroupImpl( - transcript_ids=transcript_ids, - client=self._client, - ) - self._executor = concurrent.futures.ThreadPoolExecutor() - - @property - def transcripts(self) -> List[Transcript]: - """ - Returns the list of the transcripts within the `TranscriptGroup` - """ - - return self._impl.transcripts - - def __iter__(self) -> Iterator[Transcript]: - """ - Iterate over the transcripts within the `TranscriptGroup` - """ - - return iter(self.transcripts) - - @classmethod - def get_by_ids( - cls, transcript_ids: List[str] - ) -> Union[Self, Tuple[Self, List[types.AssemblyAIError]]]: - return cls(transcript_ids=transcript_ids).wait_for_completion() - - @classmethod - def get_by_ids_async( - cls, transcript_ids: List[str] - ) -> concurrent.futures.Future[ - Union[Self, Tuple[Self, List[types.AssemblyAIError]]] - ]: - return cls(transcript_ids=transcript_ids).wait_for_completion_async() - - @property - def status(self) -> types.TranscriptStatus: - """ - Return the status of the `TranscriptGroup`. - - e.g. if any of the transcripts is in `error` status, the whole `TranscriptGroup` will be in `error` status. - """ - - all_status = {t.status for t in self.transcripts} - - if any(s == types.TranscriptStatus.error for s in all_status): - return types.TranscriptStatus.error - elif any(s == types.TranscriptStatus.queued for s in all_status): - return types.TranscriptStatus.queued - elif any(s == types.TranscriptStatus.processing for s in all_status): - return types.TranscriptStatus.processing - elif all(s == types.TranscriptStatus.completed for s in all_status): - return types.TranscriptStatus.completed - else: - raise ValueError(f"Unexpected status type: {all_status}") - - @property - def lemur(self) -> lemur.Lemur: - """ - Access AssemblyAI's LeMUR functionality. - """ - - return lemur.Lemur( - client=self._impl._client, - sources=[types.LemurSource(t) for t in self.transcripts], - ) - - def add_transcript( - self, - transcript: Union[Transcript, str], - ) -> Self: - """ - Adds a transcript to the given `TranscriptGroup` - - Args: - transcript: A `Transcript` object or the ID as a `str` - """ - self._impl.add_transcript(transcript) - - return self - - def wait_for_completion( - self, - return_failures: Optional[bool] = False, - ) -> Union[Self, Tuple[Self, List[types.AssemblyAIError]]]: - """ - Polls each transcript within the `TranscriptGroup`. - - Note - if an HTTP error is encountered when waiting for a Transcript in the TranscriptGroup, it will be popped from the group and added to the list of failures. - You can return this list of failures with `return_failures=True`. - - Args: - return_failures: Whether to return a list of errors for transcripts that failed due to HTTP errors. - """ - if return_failures is True: - failures = self._impl.wait_for_completion(return_failures=return_failures) - if failures is None: - raise ValueError("return_failures was set but failures object is None") - return self, failures - - self._impl.wait_for_completion(return_failures=return_failures) - - return self - - def wait_for_completion_async( - self, - return_failures: Optional[bool] = False, - ) -> concurrent.futures.Future[ - Union[Self, Tuple[Self, List[types.AssemblyAIError]]], - ]: - return self._executor.submit( - self.wait_for_completion, return_failures=return_failures - ) - - -class _TranscriberImpl: - """ - Implementation of the Transcriber class. - """ - - def __init__( - self, - *, - client: _client.Client, - config: types.TranscriptionConfig, - ) -> None: - self._client = client - self.config = config - - def upload_file(self, data: Union[str, bytes, BinaryIO]) -> str: - if isinstance(data, str): - with open(data, "rb") as audio_file: - return api.upload_file( - client=self._client.http_client, - audio_file=audio_file, - ) - else: - return api.upload_file( - client=self._client.http_client, - audio_file=data, - ) - - def transcribe_url( - self, - *, - url: str, - config: types.TranscriptionConfig, - poll: bool, - ) -> Transcript: - transcript_request = types.TranscriptRequest( - audio_url=url, - **config.raw.dict(exclude_none=True), - ) - # No try-except - if there is an HTTP error raise it to the user - transcript = Transcript.from_response( - client=self._client, - response=api.create_transcript( - client=self._client.http_client, - request=transcript_request, - ), - ) - - if poll: - return transcript.wait_for_completion() - - return transcript - - def transcribe_file( - self, - *, - data: Union[str, bytes, BinaryIO], - config: types.TranscriptionConfig, - poll: bool, - ) -> Transcript: - # Note: If uploading fails, it should raise an Exception to the user, hence no try-except here. - audio_url = self.upload_file(data) - - return self.transcribe_url( - url=audio_url, - config=config, - poll=poll, - ) - - def transcribe( - self, - data: Union[str, bytes, BinaryIO], - config: Optional[types.TranscriptionConfig], - poll: bool, - ) -> Transcript: - if config is None: - config = self.config - - if isinstance(data, str) and urlparse(data).scheme in {"http", "https"}: - return self.transcribe_url( - url=data, - config=config, - poll=poll, - ) - - return self.transcribe_file( - data=data, - config=config, - poll=poll, - ) - - def transcribe_group( - self, - *, - data: List[Union[str, bytes, BinaryIO]], - config: Optional[types.TranscriptionConfig], - poll: bool, - return_failures: Optional[bool] = False, - ) -> Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]]: - if config is None: - config = self.config - - future_transcripts: Set[concurrent.futures.Future[Transcript]] = set() - - with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: - for d in data: - transcript_future = executor.submit( - self.transcribe, - data=d, - config=config, - poll=False, - ) - - future_transcripts.add(transcript_future) - - finished_futures, _ = concurrent.futures.wait(future_transcripts) - - transcript_group = TranscriptGroup( - client=self._client, - ) - failures: List[types.AssemblyAIError] = [] - - for future in finished_futures: - try: - transcript_group.add_transcript(future.result()) - except types.TranscriptError as e: - failures.append(e) - - if poll is True and return_failures is True: - res = transcript_group.wait_for_completion(return_failures=return_failures) - if not isinstance(res, tuple): - raise ValueError( - "return_failures was set but did not receive failures object" - ) - transcript_group, completion_failures = res - failures.extend(completion_failures) - elif poll: - res = transcript_group.wait_for_completion(return_failures=return_failures) - if not isinstance(res, TranscriptGroup): - raise ValueError( - "return_failures was not set but did receive failures object" - ) - transcript_group = res - - if return_failures is True: - return transcript_group, failures - else: - return transcript_group - - def list_transcripts( - self, - params: Optional[types.ListTranscriptParameters], - ) -> types.ListTranscriptResponse: - return api.list_transcripts(client=self._client.http_client, params=params) - - -class Transcriber: - """ - A transcriber used for transcribing URLs or local audio files. - """ - - def __init__( - self, - *, - client: Optional[_client.Client] = None, - config: Optional[types.TranscriptionConfig] = None, - max_workers: Optional[int] = None, - ) -> None: - """ - Initializes the `Transcriber` with the given parameters. - - Args: - `client`: The `Client` to use for the `Transcriber`. If `None` is given, the - default settings for the `Client` will be used. - `config`: The default configuration for the `Transcriber`. If `None` is given, - the default configuration of a `TranscriptionConfig` will be used. - `max_workers`: The maximum number of parallel jobs when using the `_async` - methods on the `Transcriber`. By default it uses `os.cpu_count() - 1` - - Example: - To use the `Transcriber` with the default settings, you can simply do: - ``` - transcriber = aai.Transcriber() - ``` - - To use the `Transcriber` with a custom configuration, you can do: - ``` - config = aai.TranscriptionConfig(punctuate=False, format_text=False) - - transcriber = aai.Transcriber(config=config) - ``` - """ - self._client = client or _client.Client.get_default() - - self._impl = _TranscriberImpl( - client=self._client, - config=config or types.TranscriptionConfig(), - ) - - if not max_workers: - cpu_count = os.cpu_count() - if not cpu_count: - max_workers = 1 - else: - max_workers = max(1, cpu_count - 1) - - self._executor = concurrent.futures.ThreadPoolExecutor( - max_workers=max_workers, - ) - - @property - def config(self) -> types.TranscriptionConfig: - """ - Returns the default configuration of the `Transcriber`. - """ - return self._impl.config - - @config.setter - def config(self, config: types.TranscriptionConfig) -> None: - """ - Sets the default configuration of the `Transcriber`. - - Args: - `config`: The new default configuration. - """ - self._impl.config = config - - def upload_file(self, data: Union[str, bytes, BinaryIO]) -> str: - """ - Uploads an audio file which can be specified as local path or binary object. - - Args: - `data`: A local file (as path), or a binary object. - - Returns: The URL of the uploaded audio file. - """ - return self._impl.upload_file(data=data) - - def upload_file_async( - self, data: Union[str, bytes, BinaryIO] - ) -> concurrent.futures.Future[str]: - """ - Uploads an audio file which can be specified as local path or binary object. - - Args: - `data`: A local file (as path), or a binary object. - - Returns: The URL of the uploaded audio file. - """ - return self._executor.submit( - self._impl.upload_file, - data=data, - ) - - def submit( - self, - data: Union[str, bytes, BinaryIO], - config: Optional[types.TranscriptionConfig] = None, - ) -> Transcript: - """ - Submits a transcription job without waiting for its completion. - - Args: - data: An URL, a local file (as path), raw `bytes`, or a binary object. - config: Transcription options and features. If `None` is given, the Transcriber's - default configuration will be used. - """ - return self._impl.transcribe( - data=data, - config=config, - poll=False, - ) - - def submit_group( - self, - data: List[Union[str, bytes, BinaryIO]], - config: Optional[types.TranscriptionConfig] = None, - return_failures: Optional[bool] = False, - ) -> Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]]: - """ - Submits multiple transcription jobs without waiting for their completion. - - Args: - data: A list of local paths, URLs, raw `bytes`, or binary objects (can be mixed). - config: Transcription options and features. If `None` is given, the Transcriber's - default configuration will be used. - return_failures: Whether to include a list of errors for transcriptions that failed due to HTTP errors - """ - return self._impl.transcribe_group( - data=data, - config=config, - poll=False, - return_failures=return_failures, - ) - - def transcribe( - self, - data: Union[str, bytes, BinaryIO], - config: Optional[types.TranscriptionConfig] = None, - ) -> Transcript: - """ - Transcribes an audio file which can be specified as local path, URL, raw `bytes`, or binary object. - - Args: - data: An URL, a local file (as path), raw `bytes`, or a binary object. - config: Transcription options and features. If `None` is given, the Transcriber's - default configuration will be used. - """ - - return self._impl.transcribe( - data=data, - config=config, - poll=True, - ) - - def transcribe_async( - self, - data: Union[str, bytes, BinaryIO], - config: Optional[types.TranscriptionConfig] = None, - ) -> concurrent.futures.Future[Transcript]: - """ - Transcribes an audio file which can be specified as local path, URL, raw `bytes`, or binary object. - - Args: - data: An URL, a local file (as path), raw `bytes`, or a binary object. - config: Transcription options and features. If `None` is given, the Transcriber's - default configuration will be used. - """ - - return self._executor.submit( - self._impl.transcribe, - data=data, - config=config, - poll=True, - ) - - def transcribe_group( - self, - data: List[Union[str, bytes, BinaryIO]], - config: Optional[types.TranscriptionConfig] = None, - return_failures: Optional[bool] = False, - ) -> Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]]: - """ - Transcribes a list of files (as local paths, URLs, or binary objects). - - Args: - data: A list of local paths, URLs, raw `bytes`, or binary objects (can be mixed). - config: Transcription options and features. If `None` is given, the Transcriber's - default configuration will be used. - return_failures: Whether to include a list of errors for transcriptions that failed due to HTTP errors - """ - - return self._impl.transcribe_group( - data=data, - config=config, - poll=True, - return_failures=return_failures, - ) - - def transcribe_group_async( - self, - data: List[Union[str, bytes, BinaryIO]], - config: Optional[types.TranscriptionConfig] = None, - return_failures: Optional[bool] = False, - ) -> concurrent.futures.Future[ - Union[TranscriptGroup, Tuple[TranscriptGroup, List[types.AssemblyAIError]]] - ]: - """ - Transcribes a list of files (as local paths, URLs, or binary objects) asynchronously. - - Args: - data: A list of local paths, URLs, raw `bytes`, or binary objects (can be mixed). - config: Transcription options and features. If `None` is given, the Transcriber's - default configuration will be used. - return_failures: Whether to include a list of errors for transcriptions that failed due to HTTP errors - """ - - return self._executor.submit( - self._impl.transcribe_group, - data=data, - config=config, - poll=True, - return_failures=return_failures, - ) - - def list_transcripts( - self, - params: Optional[types.ListTranscriptParameters] = None, - ) -> types.ListTranscriptResponse: - """ - Retrieve a list of transcripts that were created. Transcripts are sorted from newest to oldest. - - Args: - params: The parameters to filter the transcript list by. - - Returns: A page with a list of transcripts along with page details. - - To paginate over all pages, you can set the `ListTranscriptParameters.before_id` - to the `before_id` of the `prev_url`. Example: - ``` - transcriber = aai.Transcriber() - params = aai.ListTranscriptParameters() - page = transcriber.list_transcripts(params) - while page.page_details.before_id_of_prev_url is not None: - params.before_id = page.page_details.before_id_of_prev_url - page = transcriber.list_transcripts(params) - ``` - """ - return self._impl.list_transcripts(params=params) - - def list_transcripts_async( - self, - params: Optional[types.ListTranscriptParameters] = None, - ) -> concurrent.futures.Future[types.ListTranscriptResponse]: - """ - Retrieve a list of transcripts that were created. Transcripts are sorted from newest to oldest. - - Args: - params: The parameters to filter the transcript list by. +from .prerecorded.v2.client import Transcriber, _TranscriberImpl # noqa: F401 +from .prerecorded.v2.transcript import Transcript, _TranscriptImpl # noqa: F401 +from .prerecorded.v2.transcript_group import ( # noqa: F401 + TranscriptGroup, + _TranscriptGroupImpl, +) - Returns: A page with a list of transcripts along with page details. - """ - return self._executor.submit(self._impl.list_transcripts, params=params) +__all__ = [ + "Transcriber", + "Transcript", + "TranscriptGroup", +] diff --git a/assemblyai/types.py b/assemblyai/types.py index 4b44f71..0e7ba5e 100644 --- a/assemblyai/types.py +++ b/assemblyai/types.py @@ -17,7 +17,7 @@ from warnings import warn if TYPE_CHECKING: - from .transcriber import Transcript + from .prerecorded.v2.transcript import Transcript try: # pydantic v2 import diff --git a/tests/unit/test_transcriber_backwards_compat.py b/tests/unit/test_transcriber_backwards_compat.py new file mode 100644 index 0000000..14ac415 --- /dev/null +++ b/tests/unit/test_transcriber_backwards_compat.py @@ -0,0 +1,109 @@ +"""Backwards-compatibility tests for the prerecorded transcription surface. + +``assemblyai.transcriber`` re-exports the prerecorded surface whose canonical +location is ``assemblyai.prerecorded.v2``. Every import that names the flat +module must keep working silently: + +- ``from assemblyai.transcriber import Transcriber`` (the flat module path) is + preserved by re-exports in ``transcriber.py``. +- ``import assemblyai as aai; aai.Transcriber`` is unchanged. +- The canonical path is ``assemblyai.prerecorded.v2``. +""" + +import warnings + +import assemblyai as aai +from assemblyai import transcriber as transcriber_module +from assemblyai.prerecorded import v2 +from assemblyai.prerecorded.v2 import client as v2_client +from assemblyai.prerecorded.v2 import transcript as v2_transcript +from assemblyai.prerecorded.v2 import transcript_group as v2_transcript_group + + +def test_old_module_path_still_imports_prerecorded_classes(): + """``from assemblyai.transcriber import ...`` resolves to the same classes.""" + from assemblyai.transcriber import Transcriber, Transcript, TranscriptGroup + + assert Transcriber is v2.Transcriber is v2_client.Transcriber + assert Transcript is v2.Transcript is v2_transcript.Transcript + assert TranscriptGroup is v2.TranscriptGroup is v2_transcript_group.TranscriptGroup + + +def test_top_level_exports_are_unchanged(): + """``aai.Transcriber`` and friends are the classes from the new package.""" + assert aai.Transcriber is v2.Transcriber + assert aai.Transcript is v2.Transcript + assert aai.TranscriptGroup is v2.TranscriptGroup + + +def test_old_module_surface_is_preserved(): + """Every name the flat ``transcriber.py`` exposes matches its canonical module.""" + for name, module in ( + ("Transcriber", v2_client), + ("Transcript", v2_transcript), + ("TranscriptGroup", v2_transcript_group), + ("_TranscriberImpl", v2_client), + ("_TranscriptGroupImpl", v2_transcript_group), + ("_TranscriptImpl", v2_transcript), + ): + assert hasattr(transcriber_module, name), ( + f"assemblyai.transcriber.{name} is gone" + ) + assert getattr(transcriber_module, name) is getattr(module, name), ( + f"assemblyai.transcriber.{name} does not match {module.__name__}.{name}" + ) + + +def test_old_module_path_is_silent(): + """The compatibility re-exports must not emit deprecation warnings.""" + with warnings.catch_warnings(): + warnings.simplefilter("error") + from assemblyai.prerecorded.v2 import ( # noqa: F401 + Transcriber as V2Transcriber, + ) + from assemblyai.transcriber import ( # noqa: F401 + Transcriber, + ) + + getattr(transcriber_module, "Transcriber") + + +def test_root_api_module_reexports_prerecorded_endpoints(): + """Every prerecorded endpoint is importable from ``assemblyai.api`` unchanged.""" + with warnings.catch_warnings(): + warnings.simplefilter("error") + from assemblyai import api as root_api + from assemblyai.prerecorded.v2 import api as v2_api + + for name in ( + "ENDPOINT_TRANSCRIPT", + "create_transcript", + "delete_transcript", + "export_subtitles_srt", + "export_subtitles_vtt", + "get_paragraphs", + "get_redacted_audio", + "get_sentences", + "get_transcript", + "list_transcripts", + "word_search", + ): + assert getattr(root_api, name) is getattr(v2_api, name), ( + f"assemblyai.api.{name} does not match prerecorded.v2.api.{name}" + ) + + for name in ("ENDPOINT_UPLOAD", "upload_file", "lemur_task", "_get_error_message"): + assert hasattr(root_api, name), f"assemblyai.api.{name} is gone" + + +def test_v2_package_exports_full_prerecorded_surface(): + """The ``prerecorded.v2`` package exposes the client, result types, and config.""" + assert v2.Transcriber is v2_client.Transcriber + for name in ( + "Transcriber", + "Transcript", + "TranscriptGroup", + "TranscriptionConfig", + ): + assert hasattr(v2, name), f"assemblyai.prerecorded.v2.{name} missing" + assert name in v2.__all__