Claude (homelab agent)andClaude Opus 5.5
4f31784a49
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
236 lines
9.8 KiB
Python
236 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Transcription queue for SDR voice recordings (runs as user sdr, loop).
|
||
|
||
Every voice recording (per-day .cls.json, class "voice"; also archived ones from .archived.json) since
|
||
SDR_TR_START that has no transcript yet is in the queue. Whisper runs on the owner's desktop (RTX 5070,
|
||
whisper_server.py) and is reached through the home Caddy: https://whisper.home.example.com (connected to the
|
||
server's LAN address directly). While the desktop is off the queue just grows; when /health answers, the
|
||
backlog is sent newest first (so live traffic appears quickly), then the rest.
|
||
|
||
Per recording: ffmpeg -> 16 kHz mono WAV -> POST /v1/audio/transcriptions. Language: auto; anything other
|
||
than ru/en/uk is retried as ru (short noisy clips fool the detector). Known Whisper hallucinations on
|
||
silence ("Продолжение следует...", subtitle credits) are stored as empty text with "h": 1.
|
||
|
||
Results: /srv/sdr/transcripts/YYYY-MM-DD.json {file: {"t": text, "l": lang, "at": epoch}} (atomic rewrite);
|
||
failures {"err": msg, "n": tries} (given up after 3). Status for the web page and the dashboard:
|
||
/srv/sdr/transcripts/status.json.
|
||
"""
|
||
import http.client
|
||
import json
|
||
import os
|
||
import socket
|
||
import ssl
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from urllib.parse import quote, urlparse
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from radio.channels import channels # noqa: E402
|
||
from radio.config import ARCHIVE_URL, REC_DIR # noqa: E402
|
||
from radio.recordings import DAY_RE, FILE_RE, arch_index, cls_index # noqa: E402
|
||
|
||
HOST = os.environ.get("SDR_TR_HOST", "whisper.home.example.com")
|
||
CONNECT = os.environ.get("SDR_TR_CONNECT", "192.168.88.221") # Caddy on the server, SNI = HOST
|
||
START = os.environ.get("SDR_TR_START", "2026-10-28")
|
||
OUT = os.environ.get("SDR_TR_DIR", "/srv/sdr/transcripts")
|
||
IDLE_S = 60 # health re-check while the desktop is down / queue empty
|
||
BATCH = 50 # re-scan for new (newer) recordings after this many
|
||
MAX_TRIES = 3
|
||
VERSION = 2 # results with a lower "v" are redone (v2: 0.6 s silence padding, language policy below)
|
||
AUTO_GROUPS = {"avia"} # only aviation can be English; marine/PMR/LPD/rail/hams are Russian -> forced "ru"
|
||
LOW_CONF = -1.0 # avg token log-prob below this -> "lc": 1 (shown dimmed on the page)
|
||
# Whisper pads the clip itself, but speech that starts at the very first millisecond loses its first word
|
||
PAD = "adelay=600:all=1,apad=pad_dur=0.6"
|
||
HALLU = ("продолжение следует", "субтитр", "dimatorzok", "редактор субтитров", "спасибо за просмотр",
|
||
"подписывайтесь", "thank you for watching", "thanks for watching", "amara.org", "ставьте лайк",
|
||
"terima kasih", "редактор а.")
|
||
HALLU_EXACT = {"you", "thank you", "bye", "thanks", "okay"}
|
||
|
||
|
||
class Conn(http.client.HTTPSConnection):
|
||
"""HTTPS to the server's LAN address with SNI/Host = HOST (new home names resolve to the WAN IP here)."""
|
||
def connect(self):
|
||
sock = socket.create_connection((CONNECT, 443), self.timeout)
|
||
self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
|
||
|
||
|
||
def request(method, path, body=None, headers=None, timeout=10):
|
||
c = Conn(HOST, 443, timeout=timeout, context=ssl.create_default_context())
|
||
try:
|
||
c.request(method, path, body=body, headers=headers or {})
|
||
r = c.getresponse()
|
||
return r.status, r.read()
|
||
finally:
|
||
c.close()
|
||
|
||
|
||
def whisper_api():
|
||
"""0 = down; 1 = first server version (no confidence, no ru/en choice); 2 = picks ru vs en by confidence."""
|
||
try:
|
||
st, b = request("GET", "/health", timeout=8)
|
||
j = json.loads(b)
|
||
return (int(j.get("api", 1)) if j.get("ok") is True else 0) if st == 200 else 0
|
||
except (OSError, ValueError, http.client.HTTPException):
|
||
return 0
|
||
|
||
|
||
def whisper_up():
|
||
return whisper_api() > 0
|
||
|
||
|
||
def load(day):
|
||
try:
|
||
with open(os.path.join(OUT, day + ".json"), encoding="utf-8") as fh:
|
||
return json.load(fh)
|
||
except (OSError, ValueError):
|
||
return {}
|
||
|
||
|
||
def save(day, d):
|
||
p = os.path.join(OUT, day + ".json")
|
||
with open(p + ".tmp", "w", encoding="utf-8") as fh:
|
||
json.dump(d, fh, ensure_ascii=False, separators=(",", ":"))
|
||
os.replace(p + ".tmp", p)
|
||
|
||
|
||
def write_status(**kw):
|
||
kw["ts"] = int(time.time())
|
||
p = os.path.join(OUT, "status.json")
|
||
with open(p + ".tmp", "w", encoding="utf-8") as fh:
|
||
json.dump(kw, fh, ensure_ascii=False)
|
||
os.replace(p + ".tmp", p)
|
||
|
||
|
||
def redo(r):
|
||
if not r:
|
||
return True
|
||
if "err" in r:
|
||
return r.get("n", 0) < MAX_TRIES
|
||
return r.get("v", 1) < VERSION
|
||
|
||
|
||
def queue(done):
|
||
"""[(day, name, archived)] voice recordings without a final result, newest first."""
|
||
items = []
|
||
days = sorted(d for d in os.listdir(REC_DIR) if DAY_RE.match(d) and d >= START)
|
||
for day in days:
|
||
dd = done.setdefault(day, load(day))
|
||
local = set(n for n in os.listdir(os.path.join(REC_DIR, day)) if FILE_RE.match(n))
|
||
for n, ci in cls_index(day).items():
|
||
if n in local and (ci or {}).get("c") == "voice":
|
||
if redo(dd.get(n)):
|
||
items.append((day, n, False))
|
||
for n, ai in arch_index(day).items():
|
||
cls = (ai or {}).get("c") or (ai or {}).get("class")
|
||
if n not in local and FILE_RE.match(n) and cls == "voice":
|
||
if redo(dd.get(n)):
|
||
items.append((day, n, True))
|
||
items.sort(key=lambda x: x[1].split("_", 1)[1], reverse=True)
|
||
return items, days
|
||
|
||
|
||
def wav(day, name, archived):
|
||
if archived:
|
||
u = urlparse(ARCHIVE_URL)
|
||
c = http.client.HTTPConnection(u.hostname, u.port or 80, timeout=30)
|
||
c.request("GET", f"/{day}/{quote(name)}")
|
||
r = c.getresponse()
|
||
if r.status != 200:
|
||
raise OSError(f"archive HTTP {r.status}")
|
||
src, inp = "pipe:0", r.read()
|
||
c.close()
|
||
else:
|
||
src, inp = os.path.join(REC_DIR, day, name), None
|
||
p = subprocess.run(["ffmpeg", "-v", "error", "-i", src,
|
||
"-af", PAD, "-ac", "1", "-ar", "16000", "-f", "wav", "pipe:1"],
|
||
input=inp, capture_output=True, timeout=60)
|
||
if p.returncode != 0 or len(p.stdout) < 1000:
|
||
raise OSError("ffmpeg: " + p.stderr.decode(errors="replace")[:200])
|
||
return p.stdout
|
||
|
||
|
||
def transcribe(audio, language=""):
|
||
b = uuid.uuid4().hex
|
||
parts = [f'--{b}\r\nContent-Disposition: form-data; name="file"; filename="a.wav"\r\n'
|
||
f"Content-Type: audio/wav\r\n\r\n".encode() + audio + b"\r\n"]
|
||
if language:
|
||
parts.append(f'--{b}\r\nContent-Disposition: form-data; name="language"\r\n\r\n{language}\r\n'.encode())
|
||
body = b"".join(parts) + f"--{b}--\r\n".encode()
|
||
st, resp = request("POST", "/v1/audio/transcriptions", body,
|
||
{"Content-Type": f"multipart/form-data; boundary={b}"}, timeout=120)
|
||
if st != 200:
|
||
raise OSError(f"whisper HTTP {st}: {resp[:200]!r}")
|
||
return json.loads(resp)
|
||
|
||
|
||
def one(day, name, archived, api):
|
||
audio = wav(day, name, archived)
|
||
grp = channels().get("channels", {}).get(FILE_RE.match(name)["f"], {}).get("group", "other")
|
||
# api 2 chooses ru vs en itself by confidence; api 1 has no such choice -> Russian for everything
|
||
r = transcribe(audio, "" if (api >= 2 and grp in AUTO_GROUPS) else "ru")
|
||
text = (r.get("text") or "").strip()
|
||
res = {"t": text, "l": r.get("language") or "ru", "at": int(time.time()), "v": VERSION}
|
||
p = r.get("avg_logprob")
|
||
if isinstance(p, (int, float)):
|
||
res["p"] = round(p, 3)
|
||
if p < LOW_CONF:
|
||
res["lc"] = 1
|
||
low = text.lower()
|
||
if not any(ch.isalnum() for ch in text) or any(h in low for h in HALLU) or low.strip(" .!?,") in HALLU_EXACT:
|
||
res["t"], res["h"] = "", 1
|
||
return res
|
||
|
||
|
||
def main():
|
||
os.makedirs(OUT, exist_ok=True)
|
||
stats = {"done_run": 0, "last_done": None}
|
||
while True:
|
||
done = {}
|
||
items, days = queue(done)
|
||
total = sum(1 for d in days for r in done[d].values() if "t" in r)
|
||
api = whisper_api()
|
||
up = api > 0
|
||
write_status(endpoint_up=up, api=api, pending=len(items), done=total, since=START, **stats)
|
||
if not up or not items:
|
||
time.sleep(IDLE_S)
|
||
continue
|
||
dirty, last_save = set(), time.time()
|
||
fails = 0
|
||
for k, (day, name, archived) in enumerate(items[:BATCH], 1):
|
||
try:
|
||
res = one(day, name, archived, api)
|
||
fails = 0
|
||
stats["done_run"] += 1
|
||
stats["last_done"] = int(time.time())
|
||
total += 1
|
||
except (OSError, ValueError, http.client.HTTPException, subprocess.SubprocessError) as e:
|
||
prev = done[day].get(name) or {}
|
||
res = {"err": str(e)[:200], "n": prev.get("n", 0) + 1}
|
||
fails += 1
|
||
if not whisper_up(): # desktop went away mid-batch: keep the item, stop
|
||
break
|
||
if fails >= 4: # server answers but fails (bug / model loading): do not burn tries
|
||
log_err = res["err"]
|
||
print("transcribe: серия ошибок, пауза 5 мин:", log_err, flush=True)
|
||
res = None
|
||
time.sleep(300)
|
||
break
|
||
if res is None:
|
||
continue
|
||
done[day][name] = res
|
||
dirty.add(day)
|
||
if time.time() - last_save > 10:
|
||
for d in dirty:
|
||
save(d, done[d])
|
||
dirty.clear()
|
||
last_save = time.time()
|
||
write_status(endpoint_up=True, pending=len(items) - k, done=total, since=START, **stats)
|
||
for d in dirty:
|
||
save(d, done[d])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|