commit 52c5cbbaa04c4b3c150cfa4dceb9a45e83ea9920 equwal <truex@equwal.com> 2026-09-22 16:00:31 -0700 Sell hours of cloud captions for Subrep Subrep makes live captions on the phone for free. The Android app can now send each piece of speech to /api/captions/transcribe. The server gives it to Groq (whisper-large-v3-turbo) and takes the length of the piece from the hour balance of the account. - backend/captions.py: the hour balance (table caption_balances), three packs (20 h $4.99, 80 h $16.99, 200 h $39), GET /api/captions, POST /api/captions/checkout and POST /api/captions/transcribe. - The server takes the seconds before it calls Groq, and gives them back when Groq fails. It keeps no sound and no text. - payments.fulfil sells a pack like a book pack and adds its hours. accounts.merge moves the hours to the account that stays. - The packs are not in pricing.plans(), so the book page does not show them. - Without SUBPLZ_WEB_GROQ_API_KEY the transcribe route answers 503. Tests: 17 in tests/test_captions.py, with property tests for the balance, the WAV wrapper and the billed seconds, and a check of the form that Groq receives. Without the hooks in payments and accounts, 2 of them fail.
backend/accounts.py | 3 + backend/captions.py | 253 +++++++++++++++++++++++++++++++++++++++++++++++++ backend/main.py | 2 + backend/payments.py | 6 +- backend/settings.py | 9 ++ tests/test_captions.py | 223 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 495 insertions(+), 1 deletion(-)
diff --git a/backend/accounts.py b/backend/accounts.py index e3e389c..48b8acb 100644 --- a/backend/accounts.py +++ b/backend/accounts.py @@ -64,6 +64,9 @@ def merge(session: Session, src: Account, dst: Account) -> None: session.query(Purchase).filter(Purchase.account_id == src.id).update( {Purchase.account_id: dst.id}, synchronize_session=False ) + from . import captions # captions imports api, which imports this module + + captions.merge(session, src, dst) dst.purchased_credits += src.purchased_credits src.purchased_credits = 0 # Add the free credits used, not the free credits left: a second device diff --git a/backend/captions.py b/backend/captions.py new file mode 100644 index 0000000..697d5b3 --- /dev/null +++ b/backend/captions.py @@ -0,0 +1,253 @@ +"""Cloud captions for Subrep: hours of speech recognition, bought in packs. + +Subrep makes live captions on the phone for free. With cloud captions, the app +sends each piece of speech here, and this server gives it to Groq +(whisper-large-v3-turbo), a larger model than a phone can run. The account +pays with hours from a pack. + + 20 hours $4.99 + 80 hours $16.99 + 200 hours $39.00 + +Groq bills about $0.04 for each audio hour, and at least 10 seconds for each +request. Pieces of speech are 1 to 11 seconds long, so one hour costs about +$0.08. The packs keep more than half of the price after that cost and the +Stripe fee (owner decision, 2026-09-22). + +The server takes the seconds of a piece before it calls Groq, and gives them +back if Groq fails. So two requests at the same time cannot spend one second +twice. The server does not keep the sound or the text. + +A pack is a pricing.Plan, so payments.py sells it the same way as a book pack. +The packs are not in pricing.plans(): the book page does not show them. +""" +from __future__ import annotations + +import io +import json +import logging +import math +import secrets +import urllib.error +import urllib.request +import wave +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi.concurrency import run_in_threadpool +from pydantic import BaseModel +from sqlalchemy import ForeignKey, Integer, String, update +from sqlalchemy.orm import Mapped, Session, mapped_column + +from .api import get_account, get_session +from .db import Account, Base +from .pricing import Plan +from .settings import settings + +log = logging.getLogger(__name__) + +SAMPLE_RATE = 16_000 +BYTES_PER_SECOND = SAMPLE_RATE * 2 # 16-bit mono + +PACKS: list[Plan] = [ + Plan(id="captions20", name="20 hours of cloud captions", credits=0, price_cents=499), + Plan(id="captions80", name="80 hours of cloud captions", credits=0, price_cents=1699), + Plan(id="captions200", name="200 hours of cloud captions", credits=0, price_cents=3900), +] +HOURS = {"captions20": 20, "captions80": 80, "captions200": 200} + + +class CaptionBalance(Base): + """The seconds of cloud captions that an account has.""" + + __tablename__ = "caption_balances" + + account_id: Mapped[str] = mapped_column( + String(64), ForeignKey("accounts.id"), primary_key=True + ) + seconds_left: Mapped[int] = mapped_column(Integer, default=0) + seconds_used: Mapped[int] = mapped_column(Integer, default=0) + + +def pack(plan_id: str) -> Plan | None: + return next((p for p in PACKS if p.id == plan_id), None) + + +def seconds_left(session: Session, account_id: str) -> int: + row = session.get(CaptionBalance, account_id) + return row.seconds_left if row else 0 + + +def add(session: Session, account_id: str, seconds: int) -> None: + """Give `seconds` to the account. The caller commits.""" + row = session.get(CaptionBalance, account_id) + if row is None: + row = CaptionBalance(account_id=account_id, seconds_left=0, seconds_used=0) + session.add(row) + row.seconds_left += seconds + + +def spend(session: Session, account_id: str, seconds: int) -> bool: + """Take `seconds` from the account, and commit. False if it has fewer.""" + done = session.execute( + update(CaptionBalance) + .where(CaptionBalance.account_id == account_id, + CaptionBalance.seconds_left >= seconds) + .values(seconds_left=CaptionBalance.seconds_left - seconds, + seconds_used=CaptionBalance.seconds_used + seconds) + ).rowcount + session.commit() + return done == 1 + + +def refund(session: Session, account_id: str, seconds: int) -> None: + session.execute( + update(CaptionBalance) + .where(CaptionBalance.account_id == account_id) + .values(seconds_left=CaptionBalance.seconds_left + seconds, + seconds_used=CaptionBalance.seconds_used - seconds) + ) + session.commit() + + +def fulfil(session: Session, account: Account, plan: Plan) -> None: + """Called by payments.fulfil for a paid pack. The caller commits.""" + if plan.id in HOURS: + add(session, account.id, HOURS[plan.id] * 3600) + + +def merge(session: Session, src: Account, dst: Account) -> None: + """Called by accounts.merge: the hours of `src` go to `dst`.""" + row = session.get(CaptionBalance, src.id) + if row is None or row.seconds_left == 0: + return + add(session, dst.id, row.seconds_left) + row.seconds_left = 0 + + +def billed_seconds(pcm_bytes: int) -> int: + """The seconds that a piece of `pcm_bytes` costs: its length, rounded up.""" + return max(1, math.ceil(pcm_bytes / BYTES_PER_SECOND)) + + +def wav(pcm: bytes) -> bytes: + out = io.BytesIO() + with wave.open(out, "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(SAMPLE_RATE) + w.writeframes(pcm) + return out.getvalue() + + +class TranscribeError(RuntimeError): + pass + + +def transcribe(pcm: bytes, language: str) -> str: + """Send one piece of speech to Groq. Returns its text.""" + boundary = secrets.token_hex(16) + fields = [("model", settings.caption_model), ("response_format", "json"), + ("temperature", "0")] + if language and language != "auto": + fields.append(("language", language)) + body = io.BytesIO() + for name, value in fields: + body.write(f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"' + f"\r\n\r\n{value}\r\n".encode()) + body.write(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; ' + f'filename="piece.wav"\r\nContent-Type: audio/wav\r\n\r\n'.encode()) + body.write(wav(pcm)) + body.write(f"\r\n--{boundary}--\r\n".encode()) + request = urllib.request.Request( + settings.caption_api_url, data=body.getvalue(), method="POST", + headers={"Authorization": f"Bearer {settings.groq_api_key}", + "Content-Type": f"multipart/form-data; boundary={boundary}"}, + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + return json.load(response).get("text", "").strip() + except urllib.error.HTTPError as exc: + raise TranscribeError(f"speech service answered {exc.code}") from exc + except (urllib.error.URLError, TimeoutError, ValueError) as exc: + raise TranscribeError(f"speech service unreachable: {exc}") from exc + + +# --------------------------------------------------------------------------- +# routes +# --------------------------------------------------------------------------- + +router = APIRouter(prefix="/api/captions") + + +class PackIn(BaseModel): + pack_id: str + + +def _state(session: Session, account: Account) -> dict: + return { + "account_id": account.id, + "seconds_left": seconds_left(session, account.id), + "available": bool(settings.groq_api_key) and settings.payments_configured, + "packs": [ + {"id": p.id, "name": p.name, "hours": HOURS[p.id], + "price_display": p.price_display} + for p in PACKS + ], + } + + +@router.get("") +def caption_state( + account: Annotated[Account, Depends(get_account)], + session: Annotated[Session, Depends(get_session)], +): + return _state(session, account) + + +@router.post("/checkout") +def caption_checkout( + body: PackIn, + account: Annotated[Account, Depends(get_account)], + session: Annotated[Session, Depends(get_session)], +): + from . import payments + + plan = pack(body.pack_id) + if plan is None: + raise HTTPException(404, "No such pack.") + try: + return {"url": payments.start_checkout(session, account, plan)} + except payments.PaymentsUnavailable as exc: + raise HTTPException(503, str(exc)) from exc + except payments.PaymentError as exc: + raise HTTPException(400, str(exc)) from exc + + +@router.post("/transcribe") +async def caption_transcribe( + request: Request, + account: Annotated[Account, Depends(get_account)], + session: Annotated[Session, Depends(get_session)], + lang: Annotated[str, Query(max_length=8)] = "auto", +): + """The body is one piece of speech: 16 kHz mono 16-bit little-endian PCM.""" + if not settings.groq_api_key: + raise HTTPException(503, "Cloud captions are not set up on this server.") + pcm = await request.body() + if not pcm or len(pcm) % 2: + raise HTTPException(400, "The body must be 16-bit PCM.") + if len(pcm) > settings.caption_max_seconds * BYTES_PER_SECOND: + raise HTTPException(413, f"A piece is at most {settings.caption_max_seconds} s.") + + cost = billed_seconds(len(pcm)) + if not spend(session, account.id, cost): + raise HTTPException(402, "No cloud caption hours left.") + try: + text = await run_in_threadpool(transcribe, pcm, lang) + except TranscribeError as exc: + refund(session, account.id, cost) + log.warning("caption piece for %s failed: %s", account.id, exc) + raise HTTPException(502, str(exc)) from exc + return {"text": text, "seconds": cost, + "seconds_left": seconds_left(session, account.id)} diff --git a/backend/main.py b/backend/main.py index c08cf4b..d50ca52 100644 --- a/backend/main.py +++ b/backend/main.py @@ -19,6 +19,7 @@ from fastapi import FastAPI from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles +from . import captions from .api import router from .db import Job, JobStatus, SessionLocal, init_db from .languages import all_languages @@ -126,6 +127,7 @@ app = FastAPI( lifespan=lifespan, ) app.include_router(router) +app.include_router(captions.router) @app.middleware("http") diff --git a/backend/payments.py b/backend/payments.py index 166d270..6653225 100644 --- a/backend/payments.py +++ b/backend/payments.py @@ -189,7 +189,10 @@ def fulfil(session: Session, checkout: dict) -> Account | None: ) # A retired plan too: the buyer may have opened the payment page before # the plan was retired. - plan = pricing.get(meta.get("plan_id") or "", retired=True) + from . import captions # captions imports api, which imports this module + + plan = (pricing.get(meta.get("plan_id") or "", retired=True) + or captions.pack(meta.get("plan_id") or "")) if account is None or plan is None: log.error("checkout %s: unknown account or plan %r", checkout.get("id"), meta) return None @@ -221,6 +224,7 @@ def fulfil(session: Session, checkout: dict) -> Account | None: account.stripe_customer_id = checkout["customer"] if plan.credits: account.purchased_credits += plan.credits + captions.fulfil(session, account, plan) session.commit() if checkout.get("subscription"): diff --git a/backend/settings.py b/backend/settings.py index 9b5de23..f78cf3e 100644 --- a/backend/settings.py +++ b/backend/settings.py @@ -124,6 +124,15 @@ class Settings(BaseSettings): video_encoder: str = "auto" video_crf: int = 28 + # --- cloud captions (Subrep) ------------------------------------------ + # Without a key, /api/captions/transcribe answers 503. Set it with + # tools/set-secret.sh SUBPLZ_WEB_GROQ_API_KEY on the server. + groq_api_key: str = "" + caption_api_url: str = "https://api.groq.com/openai/v1/audio/transcriptions" + caption_model: str = "whisper-large-v3-turbo" + # The longest piece of speech in one request. The app sends 11 s at most. + caption_max_seconds: int = 30 + # --- uploads ----------------------------------------------------------- max_upload_bytes: int = 2 * 1024 * 1024 * 1024 # 2 GiB diff --git a/tests/test_captions.py b/tests/test_captions.py new file mode 100644 index 0000000..4ef29d8 --- /dev/null +++ b/tests/test_captions.py @@ -0,0 +1,223 @@ +"""Cloud captions for Subrep: hour packs, the balance, and the speech proxy.""" + +from __future__ import annotations + +import io +import wave + +import pytest +import stripe +from hypothesis import given +from hypothesis import settings as hsettings +from hypothesis import strategies as st + +from backend import accounts, captions +from backend.db import Account, SessionLocal +from backend.settings import settings + +from .conftest import FakeStripeObject, account_id, checkout_event, post_webhook + +HOUR = 3600 +SECOND = captions.BYTES_PER_SECOND + + +@pytest.fixture +def groq(monkeypatch): + """A server with a Groq key, and Groq faked: it answers with a fixed text.""" + monkeypatch.setattr(settings, "groq_api_key", "gsk_test") + calls = [] + + def fake(pcm, language): + calls.append((len(pcm), language)) + return "こんにちは" + + monkeypatch.setattr(captions, "transcribe", fake) + return calls + + +def give(client, seconds): + with SessionLocal() as s: + captions.add(s, account_id(client), seconds) + s.commit() + + +def left(client): + return client.get("/api/captions").json()["seconds_left"] + + +def post_piece(client, n_bytes, lang="ja"): + return client.post(f"/api/captions/transcribe?lang={lang}", content=b"\0" * n_bytes) + + +# --- packs ------------------------------------------------------------------- + +def test_the_book_page_does_not_show_the_caption_packs(client): + ids = {p["id"] for p in client.get("/api/pricing").json()["plans"]} + assert not ids & set(captions.HOURS) + + +def test_the_app_sees_three_packs_and_an_empty_balance(client): + state = client.get("/api/captions").json() + assert [(p["id"], p["hours"], p["price_display"]) for p in state["packs"]] == [ + ("captions20", 20, "$4.99"), ("captions80", 80, "$16.99"), + ("captions200", 200, "$39.00"), + ] + assert state["seconds_left"] == 0 + + +def test_checkout_sells_the_pack_to_this_account(client, monkeypatch): + seen = {} + monkeypatch.setattr(stripe.checkout.Session, "create", + lambda **p: seen.update(p) or FakeStripeObject({"url": "https://stripe.test/pay"})) + r = client.post("/api/captions/checkout", json={"pack_id": "captions80"}) + assert r.status_code == 200 and r.json()["url"] == "https://stripe.test/pay" + assert seen["metadata"]["plan_id"] == "captions80" + assert seen["client_reference_id"] == account_id(client) + assert seen["line_items"][0]["price_data"]["unit_amount"] == 1699 + assert client.post("/api/captions/checkout", json={"pack_id": "pack10"}).status_code == 404 + + +def test_a_paid_pack_gives_its_hours_once(client): + acct = account_id(client) + event = checkout_event("cs_captions", acct, "captions20") + assert post_webhook(client, event).status_code == 200 + assert post_webhook(client, event).status_code == 200 # Stripe retries + assert left(client) == 20 * HOUR + # A caption pack is not a book credit. + assert client.get("/api/account").json()["credits"] == 0 + + +def test_a_merge_moves_the_hours(client, second_client): + give(client, 5 * HOUR) + give(second_client, 2 * HOUR) + with SessionLocal() as s: + src = s.get(Account, account_id(client)) + dst = s.get(Account, account_id(second_client)) + accounts.merge(s, src, dst) + s.commit() + assert captions.seconds_left(s, dst.id) == 7 * HOUR + assert captions.seconds_left(s, src.id) == 0 + + +# --- transcribe -------------------------------------------------------------- + +def test_no_key_on_the_server_is_503(client, monkeypatch): + monkeypatch.setattr(settings, "groq_api_key", "") + give(client, HOUR) + assert post_piece(client, 3 * SECOND).status_code == 503 + assert left(client) == HOUR + + +def test_no_hours_is_402_and_groq_is_not_called(client, groq): + assert post_piece(client, 3 * SECOND).status_code == 402 + assert groq == [] + + +def test_a_piece_costs_its_length_rounded_up(client, groq): + give(client, 60) + r = post_piece(client, int(2.5 * SECOND)) + assert r.status_code == 200 + assert r.json() == {"text": "こんにちは", "seconds": 3, "seconds_left": 57} + assert groq == [(int(2.5 * SECOND), "ja")] + + +def test_the_last_seconds_do_not_pay_for_a_longer_piece(client, groq): + give(client, 2) + assert post_piece(client, 3 * SECOND).status_code == 402 + assert left(client) == 2 + + +def test_a_failure_of_groq_gives_the_seconds_back(client, monkeypatch): + monkeypatch.setattr(settings, "groq_api_key", "gsk_test") + + def down(pcm, language): + raise captions.TranscribeError("speech service answered 500") + + monkeypatch.setattr(captions, "transcribe", down) + give(client, 60) + assert post_piece(client, 4 * SECOND).status_code == 502 + assert left(client) == 60 + + +@pytest.mark.parametrize("n_bytes, status", [(0, 400), (3, 400), (31 * SECOND, 413)]) +def test_a_bad_piece_is_refused_without_cost(client, groq, n_bytes, status): + give(client, HOUR) + assert post_piece(client, n_bytes).status_code == status + assert left(client) == HOUR and groq == [] + + +# --- properties -------------------------------------------------------------- + +@hsettings(max_examples=50, deadline=None) +@given(given_seconds=st.integers(0, 100), + pieces=st.lists(st.integers(1, 40 * SECOND), max_size=12), + failures=st.lists(st.booleans(), max_size=12)) +def test_the_balance_never_goes_below_zero_and_no_second_is_lost(given_seconds, pieces, failures): + with SessionLocal() as s: + acct = accounts.new_account(s) + s.commit() + captions.add(s, acct.id, given_seconds) + s.commit() + for i, n in enumerate(pieces): + cost = captions.billed_seconds(n) + if captions.spend(s, acct.id, cost) and i < len(failures) and failures[i]: + captions.refund(s, acct.id, cost) + row = s.get(captions.CaptionBalance, acct.id) + s.refresh(row) + assert row.seconds_left >= 0 + assert row.seconds_left + row.seconds_used == given_seconds + + +@given(st.binary(max_size=4 * SECOND).filter(lambda b: len(b) % 2 == 0)) +def test_wav_holds_the_pcm_unchanged(pcm): + with wave.open(io.BytesIO(captions.wav(pcm))) as w: + assert (w.getnchannels(), w.getsampwidth(), w.getframerate()) == (1, 2, 16_000) + assert w.readframes(w.getnframes()) == pcm + + +@given(st.integers(1, 60 * SECOND)) +def test_the_billed_seconds_cover_the_piece(n_bytes): + billed = captions.billed_seconds(n_bytes) + assert billed * SECOND >= n_bytes > (billed - 1) * SECOND + + + +def test_the_request_to_groq_is_the_openai_form(monkeypatch): + """transcribe() against a local server that reads the form as Groq does.""" + import json + import threading + from email.parser import BytesParser + from email.policy import HTTP + from http.server import BaseHTTPRequestHandler, HTTPServer + + seen = {} + + class Groq(BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])) + head = f"Content-Type: {self.headers['Content-Type']}\r\n\r\n".encode() + form = BytesParser(policy=HTTP).parsebytes(head + body) + for part in form.iter_parts(): + seen[part.get_param("name", header="content-disposition")] = part.get_payload(decode=True) + seen["auth"] = self.headers["Authorization"] + out = json.dumps({"text": " 今日は "}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(out))) + self.end_headers() + self.wfile.write(out) + + def log_message(self, *args): + pass + + server = HTTPServer(("127.0.0.1", 0), Groq) + threading.Thread(target=server.handle_request, daemon=True).start() + monkeypatch.setattr(settings, "caption_api_url", f"http://127.0.0.1:{server.server_port}/") + monkeypatch.setattr(settings, "groq_api_key", "gsk_test") + pcm = bytes(range(256)) * 10 + + assert captions.transcribe(pcm, "ja") == "今日は" + server.server_close() + assert seen["auth"] == "Bearer gsk_test" + assert seen["model"] == b"whisper-large-v3-turbo" + assert seen["language"] == b"ja" + assert seen["file"] == captions.wav(pcm)