Trascrizione audio (STT) con Vosk su Raspberry Pi
+ Gateway API su Hostinger

cd /home/pi

sudo a2enmod proxy proxy_http ssl headers rewrite
sudo systemctl restart apache2

cd /home/pi/stt_vosk

./venv/bin/python -V
./venv/bin/pip show flask vosk | sed -n "1,12p"

#!/usr/bin/env python3
import os
import re
import json
import time
import tempfile
import subprocess
import wave
import urllib.request

from flask import Flask, request, jsonify, make_response
from vosk import Model, KaldiRecognizer

APP_HOST = "0.0.0.0"
APP_PORT = 5005

API_KEY = "QueSTa_MiaChiaVE_PERTraduZIOne!3"

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
MODEL_PATH = os.path.join(BASE_DIR, "model")  # symlink -> /media/pi/SARDO1/vosk-model-it-0.22

# ✅ URL API Hostinger che restituisce i piatti
# Esempio: "https://tuodominio.it/api/stt_piatti.php"
DISH_API_URL = os.getenv("STT_DISH_API_URL", "https://mydigitaltools.it/api_stt/api/stt_piatti.php")

# Azienda di default se non passa header
DEFAULT_AZIENDA_ID = int(os.getenv("STT_AZIENDA_ID", "1"))

# Cache phrases (secondi)
PHRASES_TTL_SECONDS = int(os.getenv("STT_PHRASES_TTL", "120"))

# Limiti (per non far esplodere la phrase_list)
LIMIT_FULL_DISH_PHRASES = int(os.getenv("STT_LIMIT_FULL_DISH_PHRASES", "250"))  # quante frasi "ugo togli "
MAX_QTY_VARIANTS = int(os.getenv("STT_MAX_QTY_VARIANTS", "3"))                  # "ugo togli 2 " ecc.

print("Caricamento modello da:", MODEL_PATH)
model = Model(MODEL_PATH)
print("Modello caricato correttamente.")

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 15 * 1024 * 1024

def _options_204():
    return make_response("", 204)

_re_keep = re.compile(r"[^a-z0-9àèéìòù\s]", re.IGNORECASE)

def norm_phrase(s: str) -> str:
    s = (s or "").strip().lower()
    s = s.replace("’", " ")
    s = s.replace("'", " ")
    s = _re_keep.sub(" ", s)
    s = " ".join(s.split())
    return s

def get_azienda_id_from_request():
    h = (request.headers.get("X-AZIENDA-ID") or "").strip()
    if h.isdigit() and int(h) > 0:
        return int(h)
    q = (request.args.get("azienda_id") or "").strip()
    if q.isdigit() and int(q) > 0:
        return int(q)
    return DEFAULT_AZIENDA_ID

_phrases_cache = {}  # azienda_id -> {"at": ts, "phrases": [...]}

def base_words():
    nums_words = [
        "zero","uno","due","tre","quattro","cinque","sei","sette","otto","nove","dieci",
        "undici","dodici","tredici","quattordici","quindici","sedici","diciassette","diciotto","diciannove","venti","trenta"
    ]
    digits = [str(i) for i in range(0, 51)]
    return [
        "ugo","apri","chiudi","azzera","togli","tavolo",
        "il","lo","la","i","gli","le","al","allo","alla","ai","agli","alle"
    ] + nums_words + digits

def fetch_dishes_from_api(azienda_id: int):
    url = f"{DISH_API_URL}?azienda_id={azienda_id}"
    req = urllib.request.Request(
        url,
        headers={
            "X-API-KEY": API_KEY,
            "X-AZIENDA-ID": str(azienda_id),
            "User-Agent": "stt_vosk/1.0"
        },
        method="GET"
    )
    with urllib.request.urlopen(req, timeout=6) as resp:
        raw = resp.read().decode("utf-8", errors="replace")
        data = json.loads(raw)

    if not isinstance(data, dict) or not data.get("ok"):
        raise RuntimeError("API piatti non valida: " + str(data))

    dishes = data.get("dishes", [])
    if not isinstance(dishes, list):
        return []

    out = []
    for d in dishes:
        n = norm_phrase(str(d))
        if n:
            out.append(n)

    out = sorted(list(set(out)))
    print(f"[STT] Piatti da API azienda_id={azienda_id}: {len(out)}")
    return out

def build_phrases(azienda_id: int, force: bool = False):
    now = int(time.time())
    cached = _phrases_cache.get(azienda_id)

    if (not force) and cached and (now - cached["at"] < PHRASES_TTL_SECONDS):
        return cached["phrases"]

    phrases = []
    phrases.extend(base_words())

    phrases.extend([
        "ugo chiudi",
        "ugo azzera",
        "ugo apri tavolo",
        "ugo togli",
    ])

    tavoli_words = [
        ("uno", 1), ("due", 2), ("tre", 3), ("quattro", 4), ("cinque", 5),
        ("sei", 6), ("sette", 7), ("otto", 8), ("nove", 9), ("dieci", 10),
        ("undici", 11), ("dodici", 12), ("tredici", 13), ("quattordici", 14), ("quindici", 15),
        ("sedici", 16), ("diciassette", 17), ("diciotto", 18), ("diciannove", 19), ("venti", 20),
        ("trenta", 30),
    ]
    for w, n in tavoli_words:
        phrases.append(f"ugo apri tavolo {w}")
        phrases.append(f"ugo apri tavolo {n}")

    try:
        dishes = fetch_dishes_from_api(azienda_id)

        dish_words = set()
        for d in dishes:
            for w in d.split(" "):
                if len(w) >= 2:
                    dish_words.add(w)
        phrases.extend(sorted(list(dish_words)))

        for d in dishes[:LIMIT_FULL_DISH_PHRASES]:
            phrases.append("ugo togli " + d)

        qty_words = ["uno", "due", "tre", "quattro", "cinque"]
        qty_digits = ["1", "2", "3", "4", "5"]
        qN = max(0, min(MAX_QTY_VARIANTS, 5))

        for d in dishes[:min(LIMIT_FULL_DISH_PHRASES, 200)]:
            for i in range(qN):
                phrases.append(f"ugo togli {qty_words[i]} {d}")
                phrases.append(f"ugo togli {qty_digits[i]} {d}")

    except Exception as e:
        print("[STT] ERRORE lettura piatti da API:", e)

    phrases = [norm_phrase(p) for p in phrases if p]
    phrases = list(dict.fromkeys(phrases))

    _phrases_cache[azienda_id] = {"at": now, "phrases": phrases}
    print(f"[STT] phrase_list pronta azienda_id={azienda_id}: {len(phrases)} (ttl={PHRASES_TTL_SECONDS}s)")
    return phrases

def convert_to_wav(in_path, out_path):
    cmd = [
        "ffmpeg",
        "-y",
        "-hide_banner",
        "-loglevel", "error",
        "-i", in_path,
        "-ac", "1",
        "-ar", "16000",
        "-f", "wav",
        out_path
    ]
    subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

def transcribe_wav(wav_path, azienda_id: int):
    phrases = build_phrases(azienda_id, force=False)
    grammar = json.dumps({"phrase_list": phrases}, ensure_ascii=False)

    with wave.open(wav_path, "rb") as wf:
        sample_rate = wf.getframerate()
        rec = KaldiRecognizer(model, sample_rate, grammar)
        rec.SetWords(False)

        while True:
            data = wf.readframes(4000)
            if len(data) == 0:
                break
            rec.AcceptWaveform(data)

        result = json.loads(rec.FinalResult())
        text = (result.get("text") or "").strip().lower()
        print(f"[STT] Transcript azienda_id={azienda_id}:", text)
        return text

@app.get("/health")
def health():
    return jsonify({"ok": True})

@app.route("/health", methods=["OPTIONS"])
def health_options():
    return _options_204()

@app.post("/reload")
def reload_phrases():
    key = request.headers.get("X-API-KEY", "")
    if key != API_KEY:
        return jsonify({"ok": False, "error": "Unauthorized"}), 401
    azienda_id = get_azienda_id_from_request()
    phrases = build_phrases(azienda_id, force=True)
    return jsonify({"ok": True, "azienda_id": azienda_id, "count": len(phrases)})

@app.route("/reload", methods=["OPTIONS"])
def reload_options():
    return _options_204()

# --- testo -> testo (compatibile con Web Speech API del browser) ---
@app.post("/transcribe_text")
def transcribe_text():
    key = request.headers.get("X-API-KEY", "")
    if key != API_KEY:
        return jsonify({"ok": False, "error": "Unauthorized"}), 401

    data = request.get_json(silent=True) or {}
    txt = data.get("text") if isinstance(data, dict) else ""
    if not isinstance(txt, str):
        return jsonify({"ok": False, "error": "Invalid text"}), 400

    txt = txt.strip()
    if not txt:
        return jsonify({"ok": False, "error": "Missing text"}), 400

    azienda_id = get_azienda_id_from_request()
    out = norm_phrase(txt)
    return jsonify({"ok": True, "azienda_id": azienda_id, "transcript": out})

@app.route("/transcribe_text", methods=["OPTIONS"])
def transcribe_text_options():
    return _options_204()

# --- audio -> testo (Vosk) ---
@app.post("/transcribe")
def transcribe():
    key = request.headers.get("X-API-KEY", "")
    if key != API_KEY:
        return jsonify({"ok": False, "error": "Unauthorized"}), 401

    if "audio" not in request.files:
        return jsonify({"ok": False, "error": "Missing file"}), 400

    azienda_id = get_azienda_id_from_request()
    f = request.files["audio"]

    with tempfile.TemporaryDirectory() as tmpdir:
        in_path = os.path.join(tmpdir, "input.webm")
        wav_path = os.path.join(tmpdir, "audio.wav")

        f.save(in_path)
        convert_to_wav(in_path, wav_path)
        text = transcribe_wav(wav_path, azienda_id)

        return jsonify({"ok": True, "azienda_id": azienda_id, "transcript": text})

@app.route("/transcribe", methods=["OPTIONS"])
def transcribe_options():
    return _options_204()

if __name__ == "__main__":
    try:
        build_phrases(DEFAULT_AZIENDA_ID, force=True)
    except Exception as e:
        print("[STT] WARNING build iniziale fallito:", e)

    print("Server STT avviato su porta", APP_PORT)
    app.run(host=APP_HOST, port=APP_PORT)

cd /home/pi

mount | grep SARDO1 || true
ls -la /home/pi/stt_vosk | grep model
ls -la /media/pi/SARDO1/vosk-model-it-0.22 | head

curl -v https://giuseppemontisci.sytes.net/stt/health

cd /home/pi

ffmpeg -version
journalctl -u stt_vosk --since "10 minutes ago"