"""Mistral text generation provider.""" from __future__ import annotations from pathlib import Path from typing import override from mistralai import Mistral, models from bulkgen.config import IMAGE_EXTENSIONS, TargetConfig from bulkgen.providers import Provider class TextProvider(Provider): """Generates text via the Mistral API.""" _api_key: str def __init__(self, api_key: str) -> None: self._api_key = api_key @override async def generate( self, target_name: str, target_config: TargetConfig, resolved_prompt: str, resolved_model: str, project_dir: Path, ) -> None: output_path = project_dir / target_name content_parts: list[str] = [resolved_prompt] for input_name in target_config.inputs: input_path = project_dir / input_name suffix = input_path.suffix.lower() if suffix in IMAGE_EXTENSIONS: content_parts.append(f"\n[Attached image: {input_name}]") else: file_content = input_path.read_text() content_parts.append( f"\n--- Contents of {input_name} ---\n{file_content}" ) full_prompt = "\n".join(content_parts) async with Mistral(api_key=self._api_key) as client: response = await client.chat.complete_async( model=resolved_model, messages=[models.UserMessage(content=full_prompt)], ) if not response.choices: msg = f"Mistral API returned no choices for target '{target_name}'" raise RuntimeError(msg) content = response.choices[0].message.content if content is None: msg = f"Mistral API returned empty content for target '{target_name}'" raise RuntimeError(msg) text = content if isinstance(content, str) else str(content) _ = output_path.write_text(text)