generated from john/python-template
134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
"""Prompt loading and provider-backed transcription service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import mimetypes
|
|
from pathlib import Path
|
|
|
|
from transcription.config import Settings, get_settings
|
|
from transcription.errors import AppError, ErrorCategory
|
|
from transcription.providers import (
|
|
ProviderAuthError,
|
|
ProviderError,
|
|
ProviderResponseError,
|
|
TranscriptionProvider,
|
|
TranscriptionResult,
|
|
get_transcription_provider,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_PROMPT_FILE = "transcribe_document.md"
|
|
SUPPORTED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff", ".pdf"}
|
|
|
|
|
|
class PromptLoadError(AppError):
|
|
"""Raised when prompt artifacts cannot be loaded safely."""
|
|
|
|
|
|
class TranscriptionError(AppError):
|
|
"""Raised when transcription execution fails."""
|
|
|
|
|
|
def load_prompt_text(*, prompt_name: str = DEFAULT_PROMPT_FILE, settings: Settings | None = None) -> str:
|
|
"""Load and validate prompt text from PROMPT_DIR."""
|
|
runtime_settings = settings or get_settings()
|
|
prompt_path = runtime_settings.prompt_dir / prompt_name
|
|
|
|
if not prompt_path.exists() or not prompt_path.is_file():
|
|
raise PromptLoadError(
|
|
f"Prompt file not found: {prompt_path}",
|
|
category=ErrorCategory.INFRA_PERSISTENT,
|
|
suggestion="Verify PROMPT_DIR and prompt file configuration, then retry.",
|
|
)
|
|
|
|
prompt_text = prompt_path.read_text(encoding="utf-8").strip()
|
|
if not prompt_text:
|
|
raise PromptLoadError(
|
|
f"Prompt file is empty: {prompt_path}",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Populate the prompt file with valid instructions and retry.",
|
|
)
|
|
|
|
logger.info("Loaded prompt artifact: %s", prompt_path)
|
|
return prompt_text
|
|
|
|
|
|
def load_image_payload(image_path: str | Path) -> tuple[bytes, str]:
|
|
"""Read image bytes and detect mime type for supported uploads."""
|
|
path = Path(image_path)
|
|
|
|
if not path.exists() or not path.is_file():
|
|
raise TranscriptionError(
|
|
f"Image file not found: {path}",
|
|
category=ErrorCategory.NOT_FOUND,
|
|
suggestion="Verify the uploaded file exists and retry from the jobs page.",
|
|
)
|
|
|
|
suffix = path.suffix.lower()
|
|
if suffix not in SUPPORTED_EXTENSIONS:
|
|
raise TranscriptionError(
|
|
f"Unsupported file type: {suffix}",
|
|
category=ErrorCategory.USER_INPUT,
|
|
suggestion="Use JPG, JPEG, PNG, TIFF, or PDF files.",
|
|
)
|
|
|
|
mime_type, _ = mimetypes.guess_type(path.name)
|
|
if suffix in {".tif", ".tiff"}:
|
|
mime_type = "image/tiff"
|
|
if not mime_type:
|
|
raise TranscriptionError(
|
|
f"Unable to determine MIME type for: {path}",
|
|
category=ErrorCategory.VALIDATION,
|
|
suggestion="Re-save the file in a supported format and retry.",
|
|
)
|
|
|
|
return path.read_bytes(), mime_type
|
|
|
|
|
|
def transcribe_document_image(
|
|
image_path: str | Path,
|
|
*,
|
|
prompt_name: str = DEFAULT_PROMPT_FILE,
|
|
settings: Settings | None = None,
|
|
provider: TranscriptionProvider | None = None,
|
|
) -> TranscriptionResult:
|
|
"""Transcribe a local image using the configured prompt and provider."""
|
|
runtime_settings = settings or get_settings()
|
|
prompt_text = load_prompt_text(prompt_name=prompt_name, settings=runtime_settings)
|
|
image_bytes, mime_type = load_image_payload(image_path)
|
|
|
|
adapter = provider or get_transcription_provider(settings=runtime_settings)
|
|
logger.info("Starting transcription for image=%s mime_type=%s", image_path, mime_type)
|
|
|
|
try:
|
|
result = adapter.transcribe(
|
|
prompt_text=prompt_text,
|
|
image_bytes=image_bytes,
|
|
mime_type=mime_type,
|
|
)
|
|
except ProviderAuthError as exc:
|
|
raise TranscriptionError(
|
|
"Provider authentication failed",
|
|
category=ErrorCategory.INFRA_PERSISTENT,
|
|
suggestion="Verify provider API credentials and retry.",
|
|
) from exc
|
|
except ProviderResponseError as exc:
|
|
raise TranscriptionError(
|
|
"Provider returned an invalid response",
|
|
category=ErrorCategory.EXTERNAL_PROVIDER,
|
|
suggestion="Retry once. If it persists, try a different provider model or inspect provider status.",
|
|
retriable=True,
|
|
) from exc
|
|
except ProviderError as exc:
|
|
raise TranscriptionError(
|
|
"Provider transcription failed",
|
|
category=ErrorCategory.EXTERNAL_PROVIDER,
|
|
suggestion="Retry the transcription from jobs. If repeated, check provider availability.",
|
|
retriable=True,
|
|
) from exc
|
|
|
|
logger.info("Transcription completed for image=%s provider=%s", image_path, result.provider)
|
|
return result
|