File size: 3,413 Bytes
c641d5f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
"""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)
    # The current huggingface_hub ASR client accepts raw audio bytes. Passing a
    # pathlib.Path reaches multipart encoding and fails on Windows.
    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