+ "value": "import os\nimport stat\nfrom pathlib import Path\nfrom typing import BinaryIO\n\nimport assemblyai as aai\n\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.io import BoolInput, DropdownInput, FileInput, MessageTextInput, Output, SecretStrInput\nfrom lfx.log.logger import logger\nfrom lfx.schema.data import Data\nfrom lfx.services.deps import get_storage_service\n\n\nclass AssemblyAITranscriptionJobCreator(Component):\n display_name = \"AssemblyAI Start Transcript\"\n description = \"Create a transcription job for an audio file using AssemblyAI with advanced options\"\n documentation = \"https://www.assemblyai.com/docs\"\n icon = \"AssemblyAI\"\n\n inputs = [\n SecretStrInput(\n name=\"api_key\",\n display_name=\"Assembly API Key\",\n info=\"Your AssemblyAI API key. You can get one from https://www.assemblyai.com/\",\n required=True,\n ),\n FileInput(\n name=\"audio_file\",\n display_name=\"Audio File\",\n file_types=[\n \"3ga\",\n \"8svx\",\n \"aac\",\n \"ac3\",\n \"aif\",\n \"aiff\",\n \"alac\",\n \"amr\",\n \"ape\",\n \"au\",\n \"dss\",\n \"flac\",\n \"flv\",\n \"m4a\",\n \"m4b\",\n \"m4p\",\n \"m4r\",\n \"mp3\",\n \"mpga\",\n \"ogg\",\n \"oga\",\n \"mogg\",\n \"opus\",\n \"qcp\",\n \"tta\",\n \"voc\",\n \"wav\",\n \"wma\",\n \"wv\",\n \"webm\",\n \"mts\",\n \"m2ts\",\n \"ts\",\n \"mov\",\n \"mp2\",\n \"mp4\",\n \"m4p\",\n \"m4v\",\n \"mxf\",\n ],\n info=\"The audio file to transcribe\",\n required=True,\n ),\n MessageTextInput(\n name=\"audio_file_url\",\n display_name=\"Audio File URL\",\n info=\"The URL of the audio file to transcribe (Can be used instead of a File)\",\n advanced=True,\n ),\n DropdownInput(\n name=\"speech_model\",\n display_name=\"Speech Model\",\n options=[\n \"best\",\n \"nano\",\n ],\n value=\"best\",\n info=\"The speech model to use for the transcription\",\n advanced=True,\n ),\n BoolInput(\n name=\"language_detection\",\n display_name=\"Automatic Language Detection\",\n info=\"Enable automatic language detection\",\n advanced=True,\n ),\n MessageTextInput(\n name=\"language_code\",\n display_name=\"Language\",\n info=(\n \"\"\"\n The language of the audio file. Can be set manually if automatic language detection is disabled.\n See https://www.assemblyai.com/docs/getting-started/supported-languages \"\"\"\n \"for a list of supported language codes.\"\n ),\n advanced=True,\n ),\n BoolInput(\n name=\"speaker_labels\",\n display_name=\"Enable Speaker Labels\",\n info=\"Enable speaker diarization\",\n ),\n MessageTextInput(\n name=\"speakers_expected\",\n display_name=\"Expected Number of Speakers\",\n info=\"Set the expected number of speakers (optional, enter a number)\",\n advanced=True,\n ),\n BoolInput(\n name=\"punctuate\",\n display_name=\"Punctuate\",\n info=\"Enable automatic punctuation\",\n advanced=True,\n value=True,\n ),\n BoolInput(\n name=\"format_text\",\n display_name=\"Format Text\",\n info=\"Enable text formatting\",\n advanced=True,\n value=True,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Transcript ID\", name=\"transcript_id\", method=\"create_transcription_job\"),\n ]\n\n @staticmethod\n def _is_safe_namespace_id(namespace_id: str) -> bool:\n \"\"\"Return whether a storage namespace is a single safe path segment.\"\"\"\n return (\n bool(namespace_id)\n and namespace_id not in {\".\", \"..\"}\n and not any(token in namespace_id for token in (\"/\", \"\\\\\", \"..\", \"\\x00\", \":\"))\n )\n\n def _open_uploaded_audio_file(self) -> BinaryIO | None:\n \"\"\"Open audio only when its file descriptor resolves inside a trusted storage namespace.\"\"\"\n storage_service = get_storage_service()\n namespace_ids = set()\n for namespace_id in (self.flow_id, self.user_id):\n normalized_id = str(namespace_id) if namespace_id else \"\"\n if self._is_safe_namespace_id(normalized_id):\n namespace_ids.add(normalized_id)\n if storage_service is None or not namespace_ids:\n return None\n\n try:\n trusted_roots = []\n for namespace_id in namespace_ids:\n namespace_path = Path(storage_service.build_full_path(namespace_id, \"\"))\n storage_root = namespace_path.parent.resolve()\n trusted_root = namespace_path.resolve()\n if trusted_root.is_relative_to(storage_root):\n trusted_roots.append(trusted_root)\n\n requested_path = Path(self.audio_file)\n audio_path = requested_path.resolve(strict=True)\n except (OSError, RuntimeError, TypeError, ValueError):\n return None\n\n if not any(audio_path.is_relative_to(root) for root in trusted_roots):\n return None\n\n audio_file = None\n try:\n audio_file = requested_path.open(\"rb\")\n opened_stat = os.fstat(audio_file.fileno())\n current_path = requested_path.resolve(strict=True)\n current_stat = current_path.stat()\n except (OSError, RuntimeError, TypeError, ValueError):\n if audio_file is not None:\n audio_file.close()\n return None\n\n if (\n not stat.S_ISREG(opened_stat.st_mode)\n or not os.path.samestat(opened_stat, current_stat)\n or not any(current_path.is_relative_to(root) for root in trusted_roots)\n ):\n audio_file.close()\n return None\n\n return audio_file\n\n def create_transcription_job(self) -> Data:\n aai.settings.api_key = self.api_key\n\n # Convert speakers_expected to int if it's not empty\n speakers_expected = None\n if self.speakers_expected and self.speakers_expected.strip():\n try:\n speakers_expected = int(self.speakers_expected)\n except ValueError:\n self.status = \"Error: Expected Number of Speakers must be a valid integer\"\n return Data(data={\"error\": \"Error: Expected Number of Speakers must be a valid integer\"})\n\n language_code = self.language_code or None\n\n config = aai.TranscriptionConfig(\n speech_model=self.speech_model,\n language_detection=self.language_detection,\n language_code=language_code,\n speaker_labels=self.speaker_labels,\n speakers_expected=speakers_expected,\n punctuate=self.punctuate,\n format_text=self.format_text,\n )\n\n audio = None\n uploaded_audio = None\n if self.audio_file:\n if self.audio_file_url:\n logger.warning(\"Both an audio file an audio URL were specified. The audio URL was ignored.\")\n\n uploaded_audio = self._open_uploaded_audio_file()\n if uploaded_audio is None:\n self.status = \"Error: Audio file not found\"\n return Data(data={\"error\": \"Error: Audio file not found\"})\n elif self.audio_file_url:\n audio = self.audio_file_url\n else:\n self.status = \"Error: Either an audio file or an audio URL must be specified\"\n return Data(data={\"error\": \"Error: Either an audio file or an audio URL must be specified\"})\n\n try:\n if uploaded_audio is not None:\n with uploaded_audio:\n transcript = aai.Transcriber().submit(uploaded_audio, config=config)\n else:\n transcript = aai.Transcriber().submit(audio, config=config)\n except Exception as e: # noqa: BLE001\n logger.debug(\"Error submitting transcription job\", exc_info=True)\n self.status = f\"An error occurred: {e}\"\n return Data(data={\"error\": f\"An error occurred: {e}\"})\n\n if transcript.error:\n self.status = transcript.error\n return Data(data={\"error\": transcript.error})\n result = Data(data={\"transcript_id\": transcript.id})\n self.status = result\n return result\n"
![autofix-ci[bot]](https://avatars.githubusercontent.com/in/243519?v=4&size=40){kind=avatar}
0 commit comments