| """Spend limits for the Deepgram speech-to-text proxy. |
| |
| Deliberately generous — a ceiling on a runaway or a bot, not a quota aimed at |
| real use. Deepgram is metered (see PRICE_PER_MIN) and the rate-limit |
| middleware cannot see the WebSocket route: Starlette's BaseHTTPMiddleware skips |
| non-http scopes, so `_RATE_LIMITS["/api/stt"]` has never applied. Limits are |
| enforced at the socket or nowhere. |
| |
| The daily budget is per client. That needs a stable per-client key in the usage |
| log, but the raw key is a captcha session token or an IP address and neither is |
| worth storing — so the log holds `client_id()`, a keyed hash under the captcha |
| secret. It is stable enough to count against and cannot be turned back into an |
| address by anyone reading the file. |
| """ |
|
|
| import hashlib |
| import json |
| import secrets |
| import time |
|
|
| MAX_CONCURRENT = 4 |
| |
| |
| MAX_SESSION_S = 8 * 3600 |
| |
| |
| DAILY_BUDGET_S = 10 * 3600 |
| DIRECT_SLOT_S = 11 * 60 |
| |
|
|
| |
| |
| |
| PRICE_PER_MIN = {"nova-3": 0.0048, "nova-3-multi": 0.0058, "nova-2": 0.005833} |
| DIARIZE_PER_MIN = 0.002 |
| KEYTERM_PER_MIN = 0.0013 |
|
|
|
|
| def claim_slot(pages: dict[str, float], sid: str, now: float) -> str | None: |
| """Keep `sid`'s slot or open a new one; None when MAX_CONCURRENT are live. |
| |
| `pages` maps slot id → expiry for one client's sockets straight to |
| Deepgram EU. The server never sees those sockets, only the token requests |
| and usage reports of the pages that hold them. |
| """ |
| for old in [s for s, until in pages.items() if until < now]: |
| del pages[old] |
| if sid not in pages: |
| if len(pages) >= MAX_CONCURRENT: |
| return None |
| sid = secrets.token_urlsafe(9) |
| pages[sid] = now + DIRECT_SLOT_S |
| return sid |
|
|
|
|
| def client_id(raw_key: str, secret: bytes) -> str: |
| """Pseudonymous, stable id for a client. The raw key is never stored.""" |
| return hashlib.blake2b( |
| raw_key.encode("utf-8", "replace"), key=secret[:64], digest_size=8 |
| ).hexdigest() |
|
|
|
|
| def cost_per_min(model: str, lang: str, diarize: bool, keyterm: bool) -> float: |
| """What Deepgram charges per streamed minute for this request.""" |
| is_multi = model == "nova-3" and lang == "multi" |
| rate = PRICE_PER_MIN["nova-3-multi" if is_multi else model] |
| if diarize: |
| rate += DIARIZE_PER_MIN |
| if keyterm and model == "nova-3": |
| rate += KEYTERM_PER_MIN |
| return rate |
|
|
|
|
| def seconds_today(usage_file: str, client: str | None = None, |
| now: float | None = None) -> float: |
| """Sum today's `duration_s`, for one client or for everyone. |
| |
| Tolerates a missing file and half-written lines: the log is appended to |
| without locking, so a truncated final line is normal and must not be |
| allowed to abort the count — under-counting here would silently disable |
| the daily cap. |
| """ |
| today = time.strftime("%Y-%m-%d", time.localtime(now)) |
| total = 0.0 |
| try: |
| with open(usage_file, encoding="utf-8") as log: |
| for line in log: |
| try: |
| entry = json.loads(line) |
| except ValueError: |
| continue |
| if not isinstance(entry, dict): |
| continue |
| if not str(entry.get("ts", "")).startswith(today): |
| continue |
| if client is not None and entry.get("client") != client: |
| continue |
| duration = entry.get("duration_s") or 0 |
| if isinstance(duration, (int, float)): |
| total += duration |
| except FileNotFoundError: |
| return 0.0 |
| return total |
|
|
|
|
| def is_exhausted(usage_file: str, client: str, now: float | None = None) -> bool: |
| """True when this client has reached its daily budget.""" |
| return seconds_today(usage_file, client, now) >= DAILY_BUDGET_S |
|
|
