diff --git a/sapat/process.py b/sapat/process.py index 5a3c7c4..93ff79b 100644 --- a/sapat/process.py +++ b/sapat/process.py @@ -32,6 +32,7 @@ def process_file( ext = fmt.value converted_file = input_path.with_suffix(f".{ext}") txt_file = input_path.with_suffix(".txt") + created_temp_audio = False click.echo(click.style(f"\nProcessing: {input_file}", fg="cyan", bold=True)) @@ -40,19 +41,28 @@ def process_file( with spinner_context(f"Converting to {ext.upper()}...") as spinner: try: convert_audio(str(input_path), str(converted_file), quality, fmt) + created_temp_audio = True spinner.succeed(f"Conversion to {ext.upper()} completed") except Exception as e: spinner.fail(f"Conversion failed: {e}") return else: - click.echo(click.style(f"{ext.upper()} file already exists, skipping", fg="yellow")) + click.echo( + click.style(f"{ext.upper()} file already exists, skipping", fg="yellow") + ) # Transcribe (with splitting if needed) max_size = provider.config.max_file_size_mb if should_split_file(str(converted_file), max_size_mb=max_size): - click.echo(click.style(f"File is large (>{max_size}MB), splitting into chunks...", fg="yellow")) + click.echo( + click.style( + f"File is large (>{max_size}MB), splitting into chunks...", fg="yellow" + ) + ) try: - result = _process_large_audio(str(converted_file), provider, model, language, prompt, temperature) + result = _process_large_audio( + str(converted_file), provider, model, language, prompt, temperature + ) except Exception as e: click.echo(click.style(f"Error processing large file: {e}", fg="red")) return @@ -87,8 +97,9 @@ def process_file( click.echo(click.style(f"Transcription saved to: {txt_file}", fg="green")) # Cleanup - converted_file.unlink() - click.echo(click.style("Temporary audio file removed", fg="yellow")) + if created_temp_audio and converted_file.exists(): + converted_file.unlink() + click.echo(click.style("Temporary audio file removed", fg="yellow")) def _process_large_audio( @@ -119,10 +130,13 @@ def _process_large_audio( ) all_texts.append(result.text.strip()) except Exception as e: - click.echo(click.style(f"Warning: chunk {i+1} failed: {e}", fg="yellow")) + click.echo( + click.style(f"Warning: chunk {i+1} failed: {e}", fg="yellow") + ) all_texts.append(f"[Chunk {i+1} transcription failed]") from sapat.providers.base import TranscriptionResult + return TranscriptionResult(text=" ".join(all_texts)) finally: diff --git a/sapat/providers/siliconflow.py b/sapat/providers/siliconflow.py new file mode 100644 index 0000000..e159e24 --- /dev/null +++ b/sapat/providers/siliconflow.py @@ -0,0 +1,34 @@ +# ABOUTME: SiliconFlow transcription provider +# ABOUTME: Uses SiliconFlow's OpenAI-compatible audio transcription endpoint + +from sapat.providers import register +from sapat.providers.base import AudioFormat, ProviderConfig +from sapat.providers.openai_compat import OpenAICompatProvider + + +@register +class SiliconFlowProvider(OpenAICompatProvider): + name = "siliconflow" + base_url = "https://api.siliconflow.cn/v1/audio/transcriptions" + _env_key_for_auth = "SILICONFLOW_API_KEY" + config = ProviderConfig( + required_env_vars=["SILICONFLOW_API_KEY"], + max_file_size_mb=50.0, + preferred_format=AudioFormat.MP3, + supports_correction=False, + default_model="FunAudioLLM/SenseVoiceSmall", + ) + + def _build_data( + self, model: str, language: str, prompt, temperature: float, **kwargs + ) -> dict: + return {"model": model} + + def resolve_model(self, model_alias: str) -> str: + aliases = { + "sensevoice": "FunAudioLLM/SenseVoiceSmall", + "sensevoice-small": "FunAudioLLM/SenseVoiceSmall", + "teleai": "TeleAI/TeleSpeechASR", + "telespeech": "TeleAI/TeleSpeechASR", + } + return aliases.get(model_alias, model_alias) diff --git a/tests/providers/test_siliconflow.py b/tests/providers/test_siliconflow.py new file mode 100644 index 0000000..d6f201d --- /dev/null +++ b/tests/providers/test_siliconflow.py @@ -0,0 +1,97 @@ +# ABOUTME: Tests for the SiliconFlow transcription provider +# ABOUTME: Verifies request shape, model aliases, and availability checks + +import os +from unittest.mock import patch + +import pytest + +from sapat.providers.base import TranscriptionResult + + +class FakeResponse: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self.payload = payload or {} + self.text = text + + def json(self): + return self.payload + + +class TestSiliconFlowProvider: + @patch.dict(os.environ, {"SILICONFLOW_API_KEY": "test-key"}, clear=False) + @patch("sapat.providers.openai_compat.requests.post") + def test_transcribe_sends_correct_request(self, mock_post, tmp_path): + audio_file = tmp_path / "sample.mp3" + audio_file.write_bytes(b"fake audio") + mock_post.return_value = FakeResponse(payload={"text": "hello siliconflow"}) + + from sapat.providers.siliconflow import SiliconFlowProvider + + provider = SiliconFlowProvider() + result = provider.transcribe( + str(audio_file), + model="FunAudioLLM/SenseVoiceSmall", + language="zh", + prompt="Product names: Sapat, Daytona", + temperature=0.1, + ) + + assert isinstance(result, TranscriptionResult) + assert result.text == "hello siliconflow" + + _, kwargs = mock_post.call_args + assert mock_post.call_args.args[0] == ( + "https://api.siliconflow.cn/v1/audio/transcriptions" + ) + assert kwargs["headers"]["Authorization"] == "Bearer test-key" + assert kwargs["data"]["model"] == "FunAudioLLM/SenseVoiceSmall" + assert "language" not in kwargs["data"] + assert "prompt" not in kwargs["data"] + assert "temperature" not in kwargs["data"] + assert "file" in kwargs["files"] + + @patch.dict(os.environ, {"SILICONFLOW_API_KEY": "test-key"}, clear=False) + def test_default_model_and_limits(self): + from sapat.providers.siliconflow import SiliconFlowProvider + + assert SiliconFlowProvider.config.default_model == "FunAudioLLM/SenseVoiceSmall" + assert SiliconFlowProvider.config.max_file_size_mb == 50.0 + assert SiliconFlowProvider.config.preferred_format.value == "mp3" + + @patch.dict(os.environ, {"SILICONFLOW_API_KEY": "test-key"}, clear=False) + def test_resolve_model_aliases(self): + from sapat.providers.siliconflow import SiliconFlowProvider + + provider = SiliconFlowProvider() + assert provider.resolve_model("sensevoice") == "FunAudioLLM/SenseVoiceSmall" + assert ( + provider.resolve_model("sensevoice-small") == "FunAudioLLM/SenseVoiceSmall" + ) + assert provider.resolve_model("teleai") == "TeleAI/TeleSpeechASR" + assert provider.resolve_model("custom/model") == "custom/model" + + @patch.dict(os.environ, {}, clear=True) + def test_not_available_without_key(self): + from sapat.providers.siliconflow import SiliconFlowProvider + + assert SiliconFlowProvider.is_available() is False + + @patch.dict(os.environ, {"SILICONFLOW_API_KEY": "bad-key"}, clear=False) + @patch("sapat.providers.openai_compat.requests.post") + def test_raises_on_api_error(self, mock_post, tmp_path): + audio_file = tmp_path / "sample.mp3" + audio_file.write_bytes(b"fake audio") + mock_post.return_value = FakeResponse(status_code=401, text="unauthorized") + + from sapat.providers.siliconflow import SiliconFlowProvider + + provider = SiliconFlowProvider() + with pytest.raises( + RuntimeError, match="siliconflow transcription failed \\(401\\)" + ): + provider.transcribe( + str(audio_file), + model="FunAudioLLM/SenseVoiceSmall", + ) diff --git a/tests/test_process.py b/tests/test_process.py new file mode 100644 index 0000000..e1c698a --- /dev/null +++ b/tests/test_process.py @@ -0,0 +1,98 @@ +# ABOUTME: Tests for the file processing orchestration layer +# ABOUTME: Verifies conversion cleanup does not remove user-owned source audio + +from contextlib import contextmanager + +import pytest + +from sapat.process import process_file +from sapat.providers.base import ( + AudioFormat, + ProviderConfig, + TranscriptionProvider, + TranscriptionResult, +) + + +class DummyProvider(TranscriptionProvider): + name = "dummy" + config = ProviderConfig(preferred_format=AudioFormat.MP3) + + def transcribe( + self, audio_file, model, language="en", prompt=None, temperature=0, **kwargs + ): + return TranscriptionResult(text=f"transcribed {audio_file}") + + +class DummySpinner: + def succeed(self, message): + pass + + def fail(self, message): + pass + + +@contextmanager +def dummy_spinner_context(message): + yield DummySpinner() + + +@pytest.fixture(autouse=True) +def no_spinner(monkeypatch): + monkeypatch.setattr("sapat.process.spinner_context", dummy_spinner_context) + + +def test_process_file_keeps_existing_preferred_audio(tmp_path, monkeypatch): + audio_file = tmp_path / "sample.mp3" + audio_file.write_bytes(b"already mp3") + + def fail_convert(*args, **kwargs): + raise AssertionError("conversion should not run for existing mp3") + + monkeypatch.setattr("sapat.process.convert_audio", fail_convert) + + process_file( + str(audio_file), + DummyProvider(), + model="model", + language="en", + prompt=None, + temperature=0, + quality="M", + correct=False, + ) + + assert audio_file.exists() + assert (tmp_path / "sample.txt").read_text(encoding="utf-8") == ( + f"transcribed {audio_file}" + ) + + +def test_process_file_removes_converted_temp_audio(tmp_path, monkeypatch): + video_file = tmp_path / "sample.wav" + video_file.write_bytes(b"source audio") + converted_file = tmp_path / "sample.mp3" + + def fake_convert(input_file, output_file, quality, fmt): + assert input_file == str(video_file) + assert output_file == str(converted_file) + converted_file.write_bytes(b"converted audio") + + monkeypatch.setattr("sapat.process.convert_audio", fake_convert) + + process_file( + str(video_file), + DummyProvider(), + model="model", + language="en", + prompt=None, + temperature=0, + quality="M", + correct=False, + ) + + assert video_file.exists() + assert not converted_file.exists() + assert (tmp_path / "sample.txt").read_text(encoding="utf-8") == ( + f"transcribed {converted_file}" + )