795 lines
40 KiB
Python
795 lines
40 KiB
Python
"""Multiple-stream, loopback-only transcription service. No audio persistence."""
|
||
from contextlib import asynccontextmanager
|
||
import argparse
|
||
import ipaddress
|
||
import logging
|
||
import os
|
||
import asyncio
|
||
from pathlib import Path
|
||
import threading
|
||
import time
|
||
import hashlib
|
||
import re
|
||
import sysconfig
|
||
import secrets
|
||
import json
|
||
from collections import OrderedDict
|
||
|
||
import numpy as np
|
||
from fastapi import FastAPI, HTTPException, Request
|
||
from fastapi.responses import FileResponse, JSONResponse
|
||
from starlette.middleware.trustedhost import TrustedHostMiddleware
|
||
from starlette.middleware.cors import CORSMiddleware
|
||
from urllib.parse import urlsplit
|
||
from starlette.staticfiles import StaticFiles
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
RATE = 16000
|
||
MAX_BYTES = RATE * 4 * 12
|
||
VERSION = "1.1.0"
|
||
DEFAULT_MODEL_REVISION = "536b0662742c02347bc0e980a01041f333bce120"
|
||
MODEL_REVISIONS = {"small": DEFAULT_MODEL_REVISION, "base": "ebe41f70d5b6dfa9166e2c581c45c9c0cfc57b66",
|
||
"medium": "08e178d48790749d25932bbc082711ddcfdfbc4f",
|
||
"large-v3": "edaa852ec7e145841d8ffdb056a99866b5f0a478"}
|
||
_CUDA_DLL_HANDLES = []
|
||
|
||
|
||
def configure_windows_cuda():
|
||
"""Make optional NVIDIA wheels in this environment visible to native loaders.
|
||
|
||
Only this process is changed; no driver, registry or system PATH edits.
|
||
Keep add_dll_directory handles alive for delayed cuDNN dependency loading.
|
||
"""
|
||
if os.name != 'nt' or _CUDA_DLL_HANDLES:
|
||
return
|
||
base = Path(sysconfig.get_path('purelib')) / 'nvidia'
|
||
directories = [base / name / 'bin' for name in ('cublas', 'cuda_nvrtc', 'cudnn')]
|
||
directories = [directory for directory in directories if directory.is_dir()]
|
||
for directory in directories:
|
||
_CUDA_DLL_HANDLES.append(os.add_dll_directory(str(directory)))
|
||
if directories:
|
||
# CTranslate2/cuDNN also use LoadLibrary, which consults process PATH.
|
||
os.environ['PATH'] = os.pathsep.join(map(str, directories)) + os.pathsep + os.environ.get('PATH', '')
|
||
|
||
|
||
class Engine:
|
||
def __init__(self, model, threads=4, offline=False, device="cpu", compute_type="auto", workers=1, beam_size=5):
|
||
if device in ('cuda', 'auto'):
|
||
configure_windows_cuda()
|
||
import ctranslate2
|
||
from faster_whisper import WhisperModel
|
||
requested_device = device
|
||
if device == "auto":
|
||
try:
|
||
device = "cuda" if ctranslate2.get_cuda_device_count() > 0 else "cpu"
|
||
except RuntimeError:
|
||
device = "cpu"
|
||
def load(selected):
|
||
precision = compute_type if compute_type != "auto" else ("float16" if selected == "cuda" else "int8")
|
||
instance = WhisperModel(model, device=selected, compute_type=precision,
|
||
cpu_threads=threads, num_workers=workers, local_files_only=offline,
|
||
revision=MODEL_REVISIONS.get(model))
|
||
return instance, precision
|
||
try:
|
||
self.model, self.compute_type = load(device)
|
||
except RuntimeError:
|
||
if requested_device != "auto" or device != "cuda" or compute_type != "auto":
|
||
raise
|
||
logging.warning("CUDA model loading failed; auto mode is falling back to CPU int8")
|
||
device = "cpu"
|
||
self.model, self.compute_type = load(device)
|
||
self.beam_size = beam_size
|
||
self.workers = workers
|
||
self.threads = threads
|
||
self.device = device
|
||
self.model_name = Path(model).name
|
||
self.multilingual = self.model.model.is_multilingual
|
||
self.actual_compute_type = getattr(self.model.model, 'compute_type', self.compute_type)
|
||
logging.getLogger("uvicorn.error").info("Inference device: %s; compute type: %s", device, self.compute_type)
|
||
|
||
def transcribe(self, audio, language, task="transcribe"):
|
||
segments, info = self.model.transcribe(
|
||
audio, language=language, task=task, beam_size=getattr(self, "beam_size", 5), temperature=0, vad_filter=True,
|
||
condition_on_previous_text=False,
|
||
vad_parameters={"min_silence_duration_ms": 500},
|
||
)
|
||
return " ".join(s.text.strip() for s in segments).strip(), info.language
|
||
|
||
|
||
def window(self, audio, language, context_seconds=0, final=True):
|
||
"""Commit complete words; retain the last second for the next window."""
|
||
duration = len(audio) / RATE
|
||
segments, info = self.model.transcribe(
|
||
audio, language=language, beam_size=getattr(self, "beam_size", 5), temperature=0, vad_filter=True,
|
||
word_timestamps=True, condition_on_previous_text=False,
|
||
hallucination_silence_threshold=1.0,
|
||
vad_parameters={"min_silence_duration_ms": 500},
|
||
)
|
||
cutoff = duration if final else max(context_seconds, duration - 1.0)
|
||
selected = []
|
||
for segment in segments:
|
||
for word in segment.words or []:
|
||
# Midpoint excludes previously committed context, even if a
|
||
# subsequent decode adjusts its timestamps slightly.
|
||
repeated_boundary = (context_seconds > 0 and word.end - word.start <= 0.001
|
||
and word.end <= context_seconds + 0.08)
|
||
if not repeated_boundary and (word.start + word.end) / 2 > context_seconds and word.end <= cutoff + 0.001:
|
||
selected.append(word)
|
||
text = "".join(word.word for word in selected).strip()
|
||
if not any(char.isalnum() for char in text):
|
||
text = ""
|
||
committed = duration if final else (selected[-1].end if selected else cutoff)
|
||
return text, info.language, max(context_seconds, min(committed, duration))
|
||
|
||
def warmup(self):
|
||
# Exercise the actual decoder before readiness, including CUDA libraries.
|
||
segments, _ = self.model.transcribe(np.zeros(RATE, dtype=np.float32),
|
||
language="en", beam_size=1, temperature=0,
|
||
vad_filter=False, max_new_tokens=1)
|
||
list(segments)
|
||
|
||
|
||
def _loopback_host(host):
|
||
value = (host or '').strip('[]').lower()
|
||
if value in ('localhost', '127.0.0.1', '::1'):
|
||
return True
|
||
try:
|
||
return ipaddress.ip_address(value).is_loopback
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def _private_lan_host(host):
|
||
value = (host or '').strip('[]').lower()
|
||
if not value:
|
||
return False
|
||
try:
|
||
address = ipaddress.ip_address(value)
|
||
if address.version == 4:
|
||
return any(address in ipaddress.ip_network(network) for network in
|
||
('10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16'))
|
||
return address in ipaddress.ip_network('fc00::/7')
|
||
except ValueError:
|
||
return value.endswith(('.lan', '.local', '.internal'))
|
||
|
||
|
||
def relay_target_from_env(env):
|
||
"""Validated relay address and caption site, or (None, '').
|
||
|
||
Requires wss:// except for loopback and explicitly allowed private LAN use.
|
||
"""
|
||
allow_insecure_lan = env.get('CAPTION_ALLOW_INSECURE_LAN', '').strip().lower() in ('1', 'true', 'yes', 'on')
|
||
address = env.get('RELAY_ADDRESS', '').strip()
|
||
parsed = urlsplit(address)
|
||
local_ws = parsed.scheme == 'ws' and _loopback_host(parsed.hostname)
|
||
private_ws = parsed.scheme == 'ws' and allow_insecure_lan and _private_lan_host(parsed.hostname)
|
||
if not (parsed.scheme == 'wss' or local_ws or private_ws) or not parsed.hostname:
|
||
if address:
|
||
print('RELAY_ADDRESS must be wss://, ws://localhost, or an allowed private LAN address; '
|
||
'relay auto-configuration disabled.', flush=True)
|
||
return None, ''
|
||
site = env.get('RELAY_SITE', '').strip()
|
||
if site:
|
||
parsed_site = urlsplit(site)
|
||
local_http = parsed_site.scheme == 'http' and _loopback_host(parsed_site.hostname)
|
||
private_http = parsed_site.scheme == 'http' and allow_insecure_lan and _private_lan_host(parsed_site.hostname)
|
||
if (parsed_site.scheme != 'https' and not local_http and not private_http) or not parsed_site.hostname \
|
||
or parsed_site.username or parsed_site.password or parsed_site.query or parsed_site.fragment:
|
||
print('RELAY_SITE must be https://, http://localhost, or an allowed private LAN address; ignoring it.', flush=True)
|
||
site = ''
|
||
elif not site.endswith('/'):
|
||
site += '/'
|
||
return address, site
|
||
|
||
|
||
def create_app(engine, max_streams=12, queue_timeout=20, registry=None, relay_address=None, relay_site='',
|
||
log_requests=False, allowed_origins=(), admin_key=None, allow_creation=True,
|
||
cookie_secure=False, create_limit=5, create_window=3600, clock=time.time):
|
||
registry = registry or None
|
||
admin_key = admin_key or None
|
||
if admin_key is not None and (len(admin_key) < 24 or not admin_key.isascii() or any(c.isspace() for c in admin_key)):
|
||
raise ValueError('CAPTION_ADMIN_KEY must contain at least 24 ASCII characters without whitespace')
|
||
allowed_origins = tuple(allowed_origins)
|
||
for origin in allowed_origins:
|
||
parsed = urlsplit(origin)
|
||
if (parsed.scheme not in ('https', 'http') or not parsed.hostname or
|
||
parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment or
|
||
origin != f'{parsed.scheme}://{parsed.netloc}' or '*' in origin or
|
||
any(c.isspace() or ord(c) < 32 for c in origin) or
|
||
(parsed.scheme == 'http' and parsed.hostname not in ('localhost', '127.0.0.1', '::1'))):
|
||
raise ValueError('Allowed origins must be exact HTTPS origins (HTTP only for localhost)')
|
||
parsed.port # Validate malformed ports before starting.
|
||
pending_tasks = set()
|
||
@asynccontextmanager
|
||
async def lifespan(app):
|
||
yield
|
||
# Finish shielded requests before releasing model resources at shutdown.
|
||
if pending_tasks:
|
||
await asyncio.gather(*list(pending_tasks), return_exceptions=True)
|
||
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None, lifespan=lifespan)
|
||
app.state.allowed_origins = allowed_origins
|
||
app.state.registry = registry
|
||
app.state.relay_address = relay_address
|
||
app.state.relay_site = relay_site
|
||
app.add_middleware(CORSMiddleware, allow_origins=list(allowed_origins),
|
||
allow_methods=['GET', 'POST', 'PATCH', 'DELETE'],
|
||
allow_headers=['Authorization', 'Content-Type', 'X-Caption-Local', 'X-Stream-ID', 'X-Request-ID'],
|
||
expose_headers=['Retry-After'], max_age=600)
|
||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["localhost", "127.0.0.1", "[::1]"])
|
||
# One outstanding request per stream bounds both body memory and queue size.
|
||
# asyncio's semaphore queues waiters FIFO; the model owns matching CT2 workers.
|
||
workers = getattr(engine, "workers", 1)
|
||
slots = asyncio.Semaphore(workers)
|
||
streams = OrderedDict()
|
||
running = {}
|
||
def release(stream_id):
|
||
streams[stream_id]["busy"] = False
|
||
streams[stream_id]["seen"] = time.monotonic()
|
||
if streams[stream_id].get("close_when_idle"):
|
||
del streams[stream_id]
|
||
running.pop(stream_id, None)
|
||
state["busy_since"] = min(running.values(), default=None)
|
||
|
||
def admit(stream_id):
|
||
now = time.monotonic()
|
||
for key in list(streams):
|
||
if not streams[key]["busy"] and now - streams[key]["seen"] > 120:
|
||
del streams[key]
|
||
if stream_id not in streams:
|
||
if len(streams) >= max_streams:
|
||
state["rejected"] += 1
|
||
raise HTTPException(429, "Stream capacity reached; idle sessions expire after 120 seconds", headers={"Retry-After": "2"})
|
||
streams[stream_id] = {"busy": False, "seen": now, "cache": OrderedDict()}
|
||
session = streams[stream_id]
|
||
if session["busy"]:
|
||
state["rejected"] += 1
|
||
raise HTTPException(429, "Stream busy; retry shortly", headers={"Retry-After": "1"})
|
||
session["busy"] = True
|
||
session["seen"] = now
|
||
return session["cache"]
|
||
state = {"busy_since": None, "completed": 0, "failed": 0, "rejected": 0, "queue_timeouts": 0}
|
||
app.state.inference_status = state
|
||
|
||
public_pages = ('/', '/index.html', '/capture-local.html', '/admin.html', '/health')
|
||
session_path = re.compile(r'^/api/rooms/[A-Za-z0-9_-]+/session$')
|
||
cookie_name = 'caption_room'
|
||
|
||
def bearer(request):
|
||
value = request.headers.get('authorization', '')
|
||
return value[7:].strip() if value[:7].lower() == 'bearer ' else ''
|
||
|
||
def client_ip(request):
|
||
return (request.headers.get('x-real-ip', '').strip() or
|
||
request.headers.get('x-forwarded-for', '').split(',')[0].strip() or
|
||
(request.client.host if request.client else 'unknown'))
|
||
|
||
def origin_allowed(request):
|
||
origin = request.headers.get('origin')
|
||
if not origin:
|
||
return True
|
||
expected = (relay_site or str(request.base_url)).rstrip('/')
|
||
return origin == expected or origin in allowed_origins
|
||
|
||
def is_public(request):
|
||
path = request.url.path
|
||
if path in public_pages or path.startswith('/static/') or path.startswith('/api/public/'):
|
||
return True
|
||
if request.method == 'POST' and path == '/api/rooms':
|
||
return True
|
||
return request.method == 'POST' and bool(session_path.match(path))
|
||
|
||
@app.middleware("http")
|
||
async def response_headers(request, call_next):
|
||
started = time.perf_counter()
|
||
path = request.url.path
|
||
preflight = request.method == 'OPTIONS' and request.headers.get('origin') and request.headers.get('access-control-request-method')
|
||
authorized = True
|
||
protected = (path.startswith(('/api/', '/transcribe', '/streams/', '/v1/', '/relay-config')) or
|
||
path in ('/api', '/docs', '/redoc', '/openapi.json'))
|
||
if path.startswith('/api/admin'):
|
||
token = bearer(request)
|
||
authorized = bool(admin_key) and secrets.compare_digest(token.encode(), admin_key.encode())
|
||
elif registry is not None and protected and not is_public(request) and not preflight:
|
||
room = registry.room_from_session(request.cookies.get(cookie_name), now=clock())
|
||
if room:
|
||
request.state.room = room
|
||
request.state.room_public_id = room['public_id']
|
||
registry.touch(room['public_id'])
|
||
else:
|
||
authorized = False
|
||
if not authorized:
|
||
response = JSONResponse({'detail': 'Open your room link or provide a valid service key'}, status_code=401,
|
||
headers={'WWW-Authenticate': 'Bearer'})
|
||
else:
|
||
response = await call_next(request)
|
||
if log_requests and not is_public(request):
|
||
# Fixed route labels only: no URL queries, tokens, IDs, IPs, audio or text.
|
||
route = request.scope.get('route')
|
||
print(json.dumps({'event':'http_request','method':request.method,
|
||
'route':getattr(route, 'path', 'unmatched'), 'status':response.status_code,
|
||
'duration_seconds':round(time.perf_counter()-started,4)}),flush=True)
|
||
if request.headers.get('origin') in allowed_origins:
|
||
response.headers['Access-Control-Allow-Origin'] = request.headers['origin']
|
||
response.headers.add_vary_header('Origin')
|
||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||
response.headers["Referrer-Policy"] = "no-referrer"
|
||
response.headers["Cache-Control"] = "no-store"
|
||
response.headers["Content-Security-Policy"] = (
|
||
"default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; "
|
||
"connect-src 'self' wss: ws://localhost:* ws://127.0.0.1:*; object-src 'none'; "
|
||
"base-uri 'none'; frame-ancestors 'self'"
|
||
)
|
||
return response
|
||
|
||
def require_registry():
|
||
if registry is None:
|
||
raise HTTPException(404, 'Not found')
|
||
|
||
def require_master():
|
||
if registry is None or not registry.master or not relay_address:
|
||
raise HTTPException(404, 'Room links are not configured')
|
||
|
||
async def json_body(request, limit=1024):
|
||
raw = await request.body()
|
||
if len(raw) > limit:
|
||
raise HTTPException(413, 'Request body is too large')
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
payload = json.loads(raw)
|
||
except ValueError:
|
||
raise HTTPException(400, 'Invalid JSON body')
|
||
if not isinstance(payload, dict):
|
||
raise HTTPException(400, 'Expected a JSON object')
|
||
return payload
|
||
|
||
def portal_base(request):
|
||
base = relay_site or str(request.base_url)
|
||
return base if base.endswith('/') else base + '/'
|
||
|
||
def room_links(request, public_id, secret=None):
|
||
base = portal_base(request)
|
||
admin = base + 'room/' + public_id
|
||
if secret:
|
||
admin += '#key=' + secret
|
||
return {'adminUrl': admin, 'audienceUrl': base + 'watch/' + public_id}
|
||
|
||
def room_json(room):
|
||
return {'id': room['public_id'], 'label': room['label'], 'created': room['created'],
|
||
'revoked': bool(room['revoked']), 'lastSeen': room['last_seen']}
|
||
|
||
def set_session_cookie(response, room):
|
||
token, _ = registry.issue_session(room, now=clock())
|
||
if not token:
|
||
raise HTTPException(404, 'Room links are not configured')
|
||
response.set_cookie(cookie_name, token, max_age=registry.session_ttl, httponly=True,
|
||
secure=bool(cookie_secure), samesite='lax', path='/')
|
||
return response
|
||
|
||
def room_for_path(request, public_id):
|
||
room = getattr(request.state, 'room', None)
|
||
if not room or room['public_id'] != public_id:
|
||
raise HTTPException(401, 'Open this room with its admin link')
|
||
return room
|
||
|
||
create_attempts = {}
|
||
|
||
def allow_create(ip):
|
||
if create_limit <= 0:
|
||
return True
|
||
now = time.monotonic()
|
||
recent = [t for t in create_attempts.get(ip, ()) if now - t < create_window]
|
||
if len(recent) >= create_limit:
|
||
create_attempts[ip] = recent
|
||
return False
|
||
recent.append(now)
|
||
create_attempts[ip] = recent
|
||
return True
|
||
|
||
public_attempts = {}
|
||
|
||
def allow_public(ip):
|
||
now = time.monotonic()
|
||
recent = [t for t in public_attempts.get(ip, ()) if now - t < 60]
|
||
if len(recent) >= 240:
|
||
public_attempts[ip] = recent
|
||
return False
|
||
recent.append(now)
|
||
public_attempts[ip] = recent
|
||
return True
|
||
|
||
@app.get("/")
|
||
async def index():
|
||
return FileResponse(ROOT / "static/index.html")
|
||
|
||
@app.get('/capture-local.html')
|
||
async def capture_local():
|
||
return FileResponse(ROOT / 'static/capture-local.html')
|
||
|
||
@app.get('/admin.html')
|
||
async def admin_page():
|
||
return FileResponse(ROOT / 'static/admin.html')
|
||
|
||
@app.post('/api/rooms')
|
||
async def create_room(request: Request):
|
||
require_registry()
|
||
require_master()
|
||
if not allow_creation:
|
||
raise HTTPException(403, 'Room creation is disabled on this deployment')
|
||
if not allow_create(client_ip(request)):
|
||
raise HTTPException(429, 'Too many rooms created; try again later', headers={'Retry-After': '3600'})
|
||
payload = await json_body(request)
|
||
label = payload.get('label', '')
|
||
if not isinstance(label, str) or len(label) > 128:
|
||
raise HTTPException(400, 'Label must be at most 128 characters')
|
||
try:
|
||
_, public_id, admin_secret = registry.create_room(label)
|
||
except RuntimeError:
|
||
raise HTTPException(429, 'Room capacity reached')
|
||
room = registry.get_room(public_id)
|
||
response = JSONResponse({'room': room_json(room), **room_links(request, public_id, admin_secret)}, status_code=201)
|
||
set_session_cookie(response, room)
|
||
return response
|
||
|
||
@app.post('/api/rooms/{public_id}/session')
|
||
async def room_session(public_id: str, request: Request):
|
||
require_registry()
|
||
require_master()
|
||
secret = bearer(request)
|
||
if not secret:
|
||
payload = await json_body(request)
|
||
secret = payload.get('key') if isinstance(payload.get('key'), str) else ''
|
||
room = registry.verify_admin(public_id, secret)
|
||
if not room:
|
||
raise HTTPException(401, 'This room link is invalid or has been closed')
|
||
response = JSONResponse({'room': room_json(room)})
|
||
return set_session_cookie(response, room)
|
||
|
||
@app.get('/api/me')
|
||
async def current_room(request: Request):
|
||
require_registry()
|
||
room = getattr(request.state, 'room', None)
|
||
if not room:
|
||
raise HTTPException(401, 'Open your room link first')
|
||
return {'room': room_json(room), 'audienceUrl': room_links(request, room['public_id'])['audienceUrl']}
|
||
|
||
@app.get('/api/rooms/{public_id}')
|
||
async def room_details(public_id: str, request: Request):
|
||
require_registry()
|
||
room = room_for_path(request, public_id)
|
||
return {'room': room_json(room),
|
||
'audienceUrl': room_links(request, public_id)['audienceUrl']}
|
||
|
||
@app.patch('/api/rooms/{public_id}')
|
||
async def room_update(public_id: str, request: Request):
|
||
require_registry()
|
||
room_for_path(request, public_id)
|
||
payload = await json_body(request)
|
||
label = payload.get('label')
|
||
if not isinstance(label, str) or len(label) > 128:
|
||
raise HTTPException(400, 'Label must be at most 128 characters')
|
||
if not registry.set_label(public_id, label):
|
||
raise HTTPException(404, 'Room not found')
|
||
return {'room': room_json(registry.get_room(public_id))}
|
||
|
||
@app.post('/api/rooms/{public_id}/rotate')
|
||
async def room_rotate(public_id: str, request: Request):
|
||
require_registry()
|
||
room_for_path(request, public_id)
|
||
secret = registry.rotate_admin(public_id)
|
||
if not secret:
|
||
raise HTTPException(404, 'Room not found')
|
||
room = registry.get_room(public_id)
|
||
response = JSONResponse({'room': room_json(room), **room_links(request, public_id, secret)})
|
||
return set_session_cookie(response, room)
|
||
|
||
@app.delete('/api/rooms/{public_id}')
|
||
async def room_close(public_id: str, request: Request):
|
||
require_registry()
|
||
room_for_path(request, public_id)
|
||
if not registry.revoke_room(public_id):
|
||
raise HTTPException(404, 'Room not found')
|
||
response = JSONResponse({'closed': public_id})
|
||
response.delete_cookie(cookie_name, path='/')
|
||
return response
|
||
|
||
@app.get('/api/rooms/{public_id}/config')
|
||
async def room_config(public_id: str, request: Request):
|
||
require_registry()
|
||
require_master()
|
||
room_for_path(request, public_id)
|
||
config = registry.relay_config(public_id, relay_address, relay_site, now=clock())
|
||
if config is None:
|
||
raise HTTPException(404, 'Room links are not configured')
|
||
return config
|
||
|
||
@app.get('/api/public/rooms/{public_id}')
|
||
async def public_room_config(public_id: str, request: Request):
|
||
require_registry()
|
||
require_master()
|
||
if not allow_public(client_ip(request)):
|
||
raise HTTPException(429, 'Too many requests', headers={'Retry-After': '60'})
|
||
config = registry.public_config(public_id, relay_address, relay_site, now=clock())
|
||
if config is None:
|
||
raise HTTPException(404, 'Room not found')
|
||
return config
|
||
|
||
@app.get('/api/admin/rooms')
|
||
async def admin_list_rooms():
|
||
require_registry()
|
||
return {'rooms': registry.list_rooms()}
|
||
|
||
@app.delete('/api/admin/rooms/{public_id}')
|
||
async def admin_close_room(public_id: str):
|
||
require_registry()
|
||
if not registry.revoke_room(public_id):
|
||
raise HTTPException(404, 'Room not found or already closed')
|
||
return {'closed': public_id}
|
||
|
||
@app.post('/api/admin/rooms/{public_id}/rotate')
|
||
async def admin_rotate_room(public_id: str, request: Request):
|
||
require_registry()
|
||
secret = registry.rotate_admin(public_id)
|
||
if not secret:
|
||
raise HTTPException(404, 'Room not found or already closed')
|
||
return {'room': room_json(registry.get_room(public_id)),
|
||
**room_links(request, public_id, secret)}
|
||
|
||
@app.get("/relay-config")
|
||
async def relay_config_route(request: Request):
|
||
require_registry()
|
||
require_master()
|
||
room = getattr(request.state, 'room', None)
|
||
if not room:
|
||
raise HTTPException(401, 'Open your room link first')
|
||
config = registry.relay_config(room['public_id'], relay_address, relay_site, now=clock())
|
||
if config is None:
|
||
raise HTTPException(404, "Relay not configured")
|
||
return config
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
from faster_whisper.tokenizer import _LANGUAGE_CODES
|
||
age = time.monotonic() - state["busy_since"] if state["busy_since"] is not None else 0
|
||
return {"ready": age < 90, "version": VERSION, "sample_rate": RATE, "max_seconds": 12,
|
||
"rooms_enabled": registry is not None,
|
||
"max_streams": max_streams, "workers": workers, "beam_size": getattr(engine, "beam_size", 5),
|
||
"threads_per_worker": getattr(engine, "threads", None),
|
||
"sessions": len(streams), "running": len(running),
|
||
"pending": sum(v["busy"] for v in streams.values()) - len(running),
|
||
"queue_timeout": queue_timeout, "rejected": state["rejected"],
|
||
"queue_timeouts": state["queue_timeouts"],
|
||
"multilingual": engine.multilingual,
|
||
"model": getattr(engine, "model_name", "unknown"),
|
||
"languages": sorted(_LANGUAGE_CODES) if engine.multilingual else ["en"],
|
||
"device": engine.device, "compute_type": engine.compute_type,
|
||
"actual_compute_type": getattr(engine, 'actual_compute_type', engine.compute_type),
|
||
"busy": state["busy_since"] is not None, "completed": state["completed"], "failed": state["failed"],
|
||
"modes": ["transcribe", "translate", "both"] if engine.multilingual else ["transcribe"],
|
||
"translation_target": "en" if engine.multilingual else None}
|
||
|
||
@app.delete("/streams/{stream_id}")
|
||
async def close_stream(stream_id: str, request: Request):
|
||
if request.headers.get("x-caption-local") != "1" or not origin_allowed(request):
|
||
raise HTTPException(403, "Use the local capture page")
|
||
session = streams.get(stream_id)
|
||
if session and session["busy"]:
|
||
raise HTTPException(409, "Finish pending audio before closing a stream")
|
||
streams.pop(stream_id, None)
|
||
return {"closed": stream_id}
|
||
|
||
@app.post("/transcribe")
|
||
async def transcribe(request: Request):
|
||
if request.headers.get("x-caption-local") != "1" or not origin_allowed(request):
|
||
raise HTTPException(403, "Use the local capture page")
|
||
if request.headers.get("content-type") != "application/octet-stream":
|
||
raise HTTPException(415, "Expected little-endian float32 mono PCM at 16000 Hz")
|
||
language = request.query_params.get("language", "en")
|
||
from faster_whisper.tokenizer import _LANGUAGE_CODES
|
||
if language != "auto" and language not in _LANGUAGE_CODES:
|
||
raise HTTPException(400, "Unknown language code")
|
||
mode = request.query_params.get("mode", "transcribe")
|
||
if mode not in ("transcribe", "translate", "both"):
|
||
raise HTTPException(400, "Mode must be transcribe, translate, or both")
|
||
if not engine.multilingual and (mode != "transcribe" or language not in ("auto", "en")):
|
||
raise HTTPException(400, "Restart with a multilingual model such as --model small for this language or translation")
|
||
window = request.query_params.get("window", "0") == "1"
|
||
final = request.query_params.get("final", "1") == "1"
|
||
try:
|
||
context = float(request.query_params.get("context_seconds", "0"))
|
||
if not np.isfinite(context) or not 0 <= context <= 1:
|
||
raise ValueError()
|
||
except ValueError:
|
||
raise HTTPException(400, "Context must be between 0 and 1 seconds")
|
||
request_id = request.headers.get("x-request-id", "")
|
||
if request_id and not re.fullmatch(r"[A-Za-z0-9_-]{8,80}", request_id):
|
||
raise HTTPException(400, "Invalid request ID")
|
||
stream_id = request.headers.get("x-stream-id", "legacy")
|
||
if stream_id != "legacy" and not re.fullmatch(r"[A-Za-z0-9_-]{8,80}", stream_id):
|
||
raise HTTPException(400, "Invalid stream ID")
|
||
cache = admit(stream_id)
|
||
handed_to_worker = False
|
||
try:
|
||
body = bytearray()
|
||
try:
|
||
async with asyncio.timeout(10):
|
||
async for part in request.stream():
|
||
if len(body) + len(part) > MAX_BYTES:
|
||
raise HTTPException(413, "Audio chunk exceeds 12 seconds")
|
||
body.extend(part)
|
||
except TimeoutError:
|
||
raise HTTPException(408, "Audio upload timed out")
|
||
if len(body) < 4 * 160 or len(body) % 4:
|
||
raise HTTPException(400, "Invalid PCM length")
|
||
audio = np.frombuffer(body, dtype="<f4").copy()
|
||
if not np.isfinite(audio).all() or np.max(np.abs(audio)) > 1.01:
|
||
raise HTTPException(400, "Invalid PCM samples")
|
||
duration = len(audio) / RATE
|
||
if context >= duration or (window and not final and duration - context < 2):
|
||
raise HTTPException(400, "Window must contain new audio; unfinished windows need at least two seconds")
|
||
signature = hashlib.sha256(body + str((language, mode, window, final, context)).encode()).hexdigest()
|
||
now = time.monotonic()
|
||
while cache and (len(cache) > 64 or now - next(iter(cache.values()))[0] > 120):
|
||
cache.popitem(last=False)
|
||
if request_id and request_id in cache:
|
||
_, previous_signature, result = cache[request_id]
|
||
if signature != previous_signature:
|
||
raise HTTPException(409, "Request ID was already used for different audio or options")
|
||
return result
|
||
|
||
def infer():
|
||
try:
|
||
start = time.perf_counter()
|
||
transcript = translation = ""
|
||
detected = None if language == "auto" else language
|
||
committed = duration if final else duration - 1
|
||
if np.max(np.abs(audio)) >= 0.0001:
|
||
if window:
|
||
transcript, detected, committed = engine.window(audio, detected, context, final)
|
||
elif mode in ("transcribe", "both"):
|
||
transcript, detected = engine.transcribe(audio, detected)
|
||
if mode in ("translate", "both"):
|
||
if detected == "en" and (window or mode == "both"):
|
||
translation = transcript
|
||
elif (not window and mode == "translate") or transcript:
|
||
selected_audio = audio[round(context * RATE):round(committed * RATE)] if window else audio
|
||
translation, detected = engine.transcribe(selected_audio, detected, task="translate")
|
||
if mode == "translate":
|
||
transcript = ""
|
||
result = {"text": translation if mode == "translate" else transcript,
|
||
"transcript": transcript, "translation": translation,
|
||
"language": detected, "output_language": "en" if mode == "translate" else detected,
|
||
"translation_language": "en" if mode in ("translate", "both") else None,
|
||
"mode": mode, "audio_seconds": duration, "committed_seconds": committed,
|
||
"request_id": request_id, "stream_id": stream_id,
|
||
"inference_seconds": round(time.perf_counter() - start, 3)}
|
||
return result
|
||
except Exception:
|
||
logging.exception("Transcription worker failed")
|
||
raise
|
||
|
||
async def scheduled():
|
||
queued = time.monotonic()
|
||
try:
|
||
try:
|
||
await asyncio.wait_for(slots.acquire(), timeout=queue_timeout)
|
||
except TimeoutError:
|
||
state["queue_timeouts"] += 1
|
||
raise HTTPException(429, "Inference queue deadline reached; retry pending audio", headers={"Retry-After": "1"})
|
||
try:
|
||
running[stream_id] = time.monotonic()
|
||
state["busy_since"] = min(running.values())
|
||
wait = running[stream_id] - queued
|
||
try:
|
||
result = await asyncio.to_thread(infer)
|
||
except Exception:
|
||
state["failed"] += 1
|
||
raise
|
||
result["queue_seconds"] = round(wait, 3)
|
||
if request_id:
|
||
cache[request_id] = (time.monotonic(), signature, result)
|
||
while len(cache) > 64:
|
||
cache.popitem(last=False)
|
||
state["completed"] += 1
|
||
return result
|
||
finally:
|
||
slots.release()
|
||
finally:
|
||
release(stream_id)
|
||
|
||
task = asyncio.create_task(scheduled())
|
||
pending_tasks.add(task)
|
||
task.add_done_callback(pending_tasks.discard)
|
||
# Consume late errors if the HTTP client disconnects while inference
|
||
# finishes. The result remains cached for its request ID.
|
||
task.add_done_callback(lambda done: done.exception() if not done.cancelled() else None)
|
||
handed_to_worker = True
|
||
try:
|
||
return await asyncio.shield(task)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc:
|
||
raise HTTPException(500, "Transcription failed; pending audio can be retried") from exc
|
||
finally:
|
||
if not handed_to_worker:
|
||
release(stream_id)
|
||
|
||
from api_compat import register_audio_api
|
||
def close_ephemeral(stream_id):
|
||
session = streams.get(stream_id)
|
||
if session:
|
||
if session['busy']:
|
||
# A disconnected WAV client has no stream ID to reuse. Keep
|
||
# admission locked until its shielded worker finishes, then
|
||
# release the otherwise abandoned session immediately.
|
||
session['close_when_idle'] = True
|
||
else:
|
||
streams.pop(stream_id, None)
|
||
register_audio_api(app, transcribe, close_ephemeral, max_streams, getattr(engine, 'model_name', 'local'))
|
||
app.mount("/static", StaticFiles(directory=ROOT / "static"), name="static")
|
||
return app
|
||
|
||
|
||
def start_watchdog(state, timeout=90, interval=5):
|
||
def watch():
|
||
while True:
|
||
time.sleep(interval)
|
||
since = state["busy_since"]
|
||
if since is not None and time.monotonic() - since > timeout:
|
||
logging.critical("Inference exceeded its deadline; exiting for the service supervisor to restart")
|
||
os._exit(70)
|
||
thread = threading.Thread(target=watch, name="inference-watchdog", daemon=True)
|
||
thread.start()
|
||
return thread
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--model", default=os.environ.get("CAPTION_MODEL", "small"), help="Model name or downloaded directory")
|
||
parser.add_argument("--beam-size", type=int, choices=range(1,6), default=int(os.environ.get("CAPTION_BEAM_SIZE", "5")))
|
||
parser.add_argument("--workers", type=int, default=int(os.environ.get("CAPTION_WORKERS", "1")))
|
||
parser.add_argument("--max-streams", type=int, default=int(os.environ.get("CAPTION_MAX_STREAMS", "12")))
|
||
parser.add_argument("--threads", type=int, default=int(os.environ.get("CAPTION_THREADS", "4")))
|
||
parser.add_argument("--port", type=int, default=8765)
|
||
parser.add_argument("--offline", action="store_true", default=os.environ.get("CAPTION_OFFLINE") == "1")
|
||
parser.add_argument("--log-requests", action="store_true", default=os.environ.get("CAPTION_LOG_REQUESTS") == "1",
|
||
help="Write request status/duration JSON to stdout; exclude captions and credentials")
|
||
parser.add_argument("--host", choices=["127.0.0.1", "0.0.0.0"], default="127.0.0.1",
|
||
help="Use 0.0.0.0 only inside a container with a loopback host port mapping")
|
||
parser.add_argument("--device", choices=["cpu", "cuda", "auto"], default=os.environ.get("CAPTION_DEVICE", "cpu"))
|
||
parser.add_argument("--compute-type", choices=["auto", "int8", "float32", "float16", "int8_float16"],
|
||
default=os.environ.get("CAPTION_COMPUTE_TYPE", "auto"))
|
||
args = parser.parse_args()
|
||
if args.threads < 1 or not 1 <= args.beam_size <= 5 or not 1 <= args.workers <= 8 or not 1 <= args.max_streams <= 64 or not 1 <= args.port <= 65535:
|
||
parser.error("Threads must be positive; beam size 1–5; workers 1–8; streams 1–64; port 1–65535")
|
||
print("Loading speech model; first run downloads weights. Audio stays on this host.", flush=True)
|
||
engine = Engine(args.model, args.threads, args.offline, args.device, args.compute_type, args.workers, args.beam_size)
|
||
engine.warmup()
|
||
print(f"Inference ready: {engine.device} / {engine.compute_type}", flush=True)
|
||
import uvicorn
|
||
from rooms import RoomRegistry, decode_master
|
||
relay_address, relay_site = relay_target_from_env(os.environ)
|
||
master = decode_master(os.environ.get('CAPTION_RELAY_MASTER', ''))
|
||
admin_key = os.environ.get('CAPTION_ADMIN_KEY', '').strip() or None
|
||
rooms_db = os.environ.get('CAPTION_ROOMS_DB', '').strip() or os.environ.get('CAPTION_KEYS_DB', '').strip()
|
||
registry = None
|
||
if rooms_db or master or admin_key:
|
||
registry = RoomRegistry(
|
||
rooms_db or str(ROOT / 'rooms.db'), master=master,
|
||
token_ttl=int(os.environ.get('CAPTION_TOKEN_TTL', str(60 * 60))),
|
||
session_ttl=int(os.environ.get('CAPTION_SESSION_TTL', str(30 * 24 * 3600))),
|
||
max_rooms=int(os.environ.get('CAPTION_MAX_ROOMS', '400')))
|
||
allow_creation = os.environ.get('CAPTION_ROOM_CREATION', 'open').strip().lower() not in ('off', '0', 'false', 'no')
|
||
cookie_setting = os.environ.get('CAPTION_COOKIE_SECURE', 'auto').strip().lower()
|
||
cookie_secure = cookie_setting == '1' if cookie_setting in ('0', '1') else relay_site.startswith('https://')
|
||
app = create_app(engine, args.max_streams, registry=registry, relay_address=relay_address, relay_site=relay_site,
|
||
log_requests=args.log_requests, allow_creation=allow_creation, cookie_secure=cookie_secure,
|
||
create_limit=int(os.environ.get('CAPTION_CREATE_LIMIT', '5')),
|
||
create_window=int(os.environ.get('CAPTION_CREATE_WINDOW', str(3600))),
|
||
allowed_origins=tuple(s.strip() for s in os.environ.get('CAPTION_ALLOWED_ORIGINS', '').split(',') if s.strip()),
|
||
admin_key=admin_key)
|
||
start_watchdog(app.state.inference_status)
|
||
# Browsers can retain six HTTP/1.1 connections per origin while loading assets.
|
||
# This transport bound includes idle sockets; inference admission stays separate.
|
||
uvicorn.run(app, host=args.host, port=args.port, access_log=False, limit_concurrency=max(64, args.max_streams * 6 + 16), timeout_keep_alive=5, timeout_graceful_shutdown=90)
|