| """Speech transcription using Hugging Face Inference."""
|
|
|
| from __future__ import annotations
|
|
|
| import re
|
| from pathlib import Path
|
|
|
| from huggingface_hub import InferenceClient
|
|
|
|
|
| def format_transcript_items(question: str, items: list[str]) -> str:
|
| """Apply explicit deterministic ordering and delimiter instructions."""
|
| cleaned = [item.strip() for item in items if item.strip()]
|
| lowered = question.lower()
|
| if "alphabetical" in lowered:
|
| cleaned.sort(key=str.casefold)
|
| elif "numeric" in lowered or "ascending" in lowered:
|
| try:
|
| cleaned.sort(key=lambda item: float(item))
|
| except ValueError:
|
| pass
|
| if "comma-separated" in lowered or "comma separated" in lowered:
|
| return ", ".join(cleaned)
|
| return "\n".join(cleaned)
|
|
|
|
|
| def transcribe_audio(
|
| path: str | Path,
|
| *,
|
| token: str,
|
| model_id: str,
|
| provider: str | None = None,
|
| timeout: float = 120,
|
| ) -> str:
|
| """Transcribe an audio file, with credentials supplied by the caller."""
|
| client = InferenceClient(provider=provider, token=token, timeout=timeout)
|
|
|
|
|
| result = client.automatic_speech_recognition(
|
| Path(path).read_bytes(), model=model_id
|
| )
|
| text = getattr(result, "text", None)
|
| if not text:
|
| raise RuntimeError("Speech recognition returned no transcript")
|
| return str(text).strip()
|
|
|
|
|
| def answer_from_transcript(question: str, transcript: str) -> str | None:
|
| """Deterministically extract common list answers from an ASR transcript."""
|
| lowered = question.lower()
|
| if "page number" in lowered or "page numbers" in lowered:
|
| numbers: set[int] = set()
|
| for match in re.finditer(
|
| r"\bpages?\s+((?:\d+)(?:\s*(?:,|and)\s*\d+)*)",
|
| transcript,
|
| re.IGNORECASE,
|
| ):
|
| numbers.update(int(value) for value in re.findall(r"\d+", match.group(1)))
|
| if numbers:
|
| return ", ".join(str(value) for value in sorted(numbers))
|
|
|
| if "ingredient" in lowered:
|
| candidates: list[str] = []
|
| combine = re.search(r"\bcombine\s+(.+?)(?:\. |\n|$)", transcript, re.IGNORECASE)
|
| if combine:
|
| candidates.extend(
|
| re.split(r"\s*,\s*|\s+and\s+", combine.group(1), flags=re.IGNORECASE)
|
| )
|
| for match in re.finditer(
|
| r"\b(?:stir|mix|add)\s+in\s+(.+?)(?:\. |\n|$)",
|
| transcript,
|
| re.IGNORECASE,
|
| ):
|
| candidates.extend(
|
| re.split(r"\s*,\s*|\s+and\s+", match.group(1), flags=re.IGNORECASE)
|
| )
|
| cleaned = []
|
| for item in candidates:
|
| item = re.sub(
|
| r"^(?:a|an|one|two|three|\d+(?:\.\d+)?)\s+"
|
| r"(?:(?:pinch|dash|cup|cups|tablespoon|tablespoons|teaspoon|teaspoons)"
|
| r"\s+of\s+)?",
|
| "",
|
| item.strip().rstrip(". ;:"),
|
| flags=re.IGNORECASE,
|
| )
|
| if item and item.casefold() not in {value.casefold() for value in cleaned}:
|
| cleaned.append(item)
|
| if cleaned:
|
| return ", ".join(sorted(cleaned, key=str.casefold))
|
| return None
|
|
|
