backend/api.py (32720 bytes)
1 """HTTP API. 2 3 Flow: drop files -> POST /api/uploads (stages, pairs, detects language) -> 4 POST /api/jobs/{id}/start (entitlement check, enqueue) -> poll GET /api/jobs/{id} 5 -> download from /api/jobs/{id}/files/{kind}. 6 7 Upload and start are separate so a wrong language guess costs a click rather 8 than a re-upload and a wasted multi-hour run. 9 10 Around that: /api/auth/* (email sign-in links) and /api/billing/* (Stripe 11 checkout, its return trip and its webhook). 12 """ 13 14 from __future__ import annotations 15 16 import re 17 from pathlib import Path 18 from typing import Annotated, Literal 19 20 from fastapi import ( 21 APIRouter, Cookie, Depends, File, HTTPException, Request, Response, 22 UploadFile, 23 ) 24 from fastapi.responses import FileResponse, RedirectResponse 25 from pydantic import BaseModel, Field 26 from sqlalchemy.orm import Session 27 28 from . import ( 29 accounts, auth, billing, convert, detect, languages, mailer, matching, 30 payments, pricing, 31 ) 32 from .aligner import aligner 33 from .db import Account, Artifact, Job, JobStatus, SessionLocal, new_id, utcnow 34 from .queue import queue 35 from .runner import ( 36 Paths, input_prefix, probe_duration, staged_audio_path, staged_cover_path, 37 staged_part_path, staged_text_path, 38 ) 39 from .settings import settings 40 from .storage import LocalStorage, storage 41 42 router = APIRouter(prefix="/api") 43 44 DEVICE_COOKIE = "subplz_device" 45 _CHUNK = 4 * 1024 * 1024 46 47 48 # -------------------------------------------------------------------------- 49 # session / account 50 # -------------------------------------------------------------------------- 51 52 def get_session(): 53 with SessionLocal() as s: 54 yield s 55 56 57 def get_account( 58 response: Response, 59 session: Annotated[Session, Depends(get_session)], 60 subplz_device: Annotated[str | None, Cookie()] = None, 61 ) -> Account: 62 """Identify the caller. 63 64 A cookie is the whole identity check for the free tier, on purpose. Anyone 65 who clears it gets another free book, and that is an accepted cost: a 66 signup wall does not belong in front of a tool whose pitch is "drop two 67 files in". The real protection against abuse is capacity - the queue and 68 per-worker limits - not identity. 69 70 Signing in does not replace the cookie, it re-points it: the cookie then 71 names the signed-in account, on every device that has signed in. 72 """ 73 account = None 74 if subplz_device: 75 account = ( 76 session.query(Account) 77 .filter(Account.device_token == subplz_device) 78 .first() 79 ) 80 81 if account is None: 82 account = accounts.new_account(session) 83 session.commit() 84 _set_identity(response, account) 85 return account 86 87 # This device's anonymous row was folded into a real account while no 88 # browser was attached (a payment landing by webhook). Follow it. 89 survivor = accounts.resolve(session, account) 90 if survivor.id != account.id: 91 _set_identity(response, survivor) 92 return survivor 93 94 95 def _set_identity(response: Response, account: Account) -> None: 96 response.set_cookie( 97 DEVICE_COOKIE, account.device_token, 98 max_age=60 * 60 * 24 * 365, httponly=True, samesite="lax", 99 secure=settings.cookie_secure, 100 ) 101 102 103 # -------------------------------------------------------------------------- 104 # schemas 105 # -------------------------------------------------------------------------- 106 107 class LanguageOut(BaseModel): 108 code: str 109 name: str 110 splitter: str 111 # Anything the active backend wants the user to know about this language. 112 note: str | None = None 113 114 115 class DetectionOut(BaseModel): 116 code: str | None 117 name: str | None 118 confidence: float 119 supported: bool 120 121 122 class ArtifactOut(BaseModel): 123 kind: str 124 filename: str 125 size_bytes: int 126 url: str 127 128 129 class JobOut(BaseModel): 130 id: str 131 status: str 132 stage: str 133 progress: float 134 language: str 135 language_name: str 136 splitter: str 137 model: str 138 audio_filename: str 139 audio_parts: int 140 cover_filename: str | None 141 text_filename: str 142 audio_bytes: int 143 audio_duration_seconds: float | None 144 error: str | None 145 created_at: str 146 local: bool = False 147 tier: str = billing.FREE 148 artifacts: list[ArtifactOut] = Field(default_factory=list) 149 150 151 class UploadOut(BaseModel): 152 job: JobOut 153 detected: DetectionOut 154 # Set when an image was dropped but the visitor is not signed in. 155 cover_requires_sign_in: bool = False 156 # How well the book scores against a sample of the audio, using the 157 # backend's own rule. None when the check is disabled. 158 match: dict | None = None 159 160 161 class StartIn(BaseModel): 162 language: str | None = None 163 model: str | None = None 164 165 166 class AccountOut(BaseModel): 167 id: str 168 signed_in: bool 169 email: str | None 170 # True once the owner opened a sign-in link sent to `email`. 171 email_verified: bool 172 billing_enabled: bool 173 # Whether the server can actually take money / send sign-in email yet. 174 payments_available: bool 175 email_sign_in_available: bool 176 free_tier_summary: str 177 # The cloud tier: conversions on this server's hardware. 178 cloud_available: bool 179 # Bought and free credits together. 180 credits: int 181 # The part of `credits` that is free. 182 free_credits: int 183 # The free credits (not the daily one) that this visitor has after the 184 # email is verified. Zero when it is verified. 185 free_credits_with_account: int 186 # Whether a verified account gets one free credit each day. 187 daily_free_credit: bool 188 # The part of `credits` that is left of this month's plan. 189 plan_credits: int 190 # A monthly plan with no limit: no server job spends a credit. 191 unlimited: bool 192 subscribed: bool 193 subscription_ends: str | None 194 cloud_allowed: bool 195 queue_depth: int 196 # During the beta, the page sends a copy of the files of a job in the 197 # browser to the server (settings.browser_copy). 198 browser_copy: bool = False 199 200 201 class LocalJobIn(BaseModel): 202 """A conversion about to run in the visitor's browser.""" 203 audio_filename: str = Field(max_length=512) 204 audio_parts: int = Field(default=1, ge=1, le=2000) 205 audio_bytes: int = Field(default=0, ge=0) 206 audio_duration_seconds: float | None = None 207 text_filename: str = Field(max_length=512) 208 language: str = Field(max_length=16) 209 210 211 class LocalFinishIn(BaseModel): 212 # A twenty-hour book is a few megabytes of subtitles. 213 srt: str = Field(max_length=8_000_000) 214 filename: str = Field(max_length=512) 215 metadata: dict = Field(default_factory=dict) 216 217 218 class LocalFailIn(BaseModel): 219 error: str = Field(default="", max_length=4000) 220 221 222 class EmailIn(BaseModel): 223 email: str 224 225 226 class TokenIn(BaseModel): 227 token: str 228 229 230 class CheckoutIn(BaseModel): 231 plan_id: str 232 233 234 def _job_out(job: Job, arts: list[Artifact]) -> JobOut: 235 lang = languages.get(job.language) 236 return JobOut( 237 id=job.id, 238 status=job.status.value, 239 stage=job.stage, 240 progress=round(job.progress, 4), 241 language=job.language, 242 language_name=lang.name if lang else job.language, 243 splitter=job.splitter, 244 model=job.model, 245 audio_filename=job.audio_filename, 246 audio_parts=job.audio_parts or 1, 247 cover_filename=job.cover_filename, 248 text_filename=job.text_filename, 249 audio_bytes=job.audio_bytes, 250 audio_duration_seconds=job.audio_duration_seconds, 251 error=job.error, 252 created_at=job.created_at.isoformat(), 253 local=bool(job.local), 254 tier=job.tier or billing.FREE, 255 artifacts=[ 256 ArtifactOut( 257 kind=a.kind, filename=a.filename, size_bytes=a.size_bytes, 258 url=f"/api/jobs/{job.id}/files/{a.kind}", 259 ) 260 for a in arts 261 ], 262 ) 263 264 265 def _load(session: Session, account: Account, job_id: str) -> Job: 266 job = session.get(Job, job_id) 267 if job is None or job.account_id != account.id: 268 raise HTTPException(404, "Job not found") 269 return job 270 271 272 def _artifacts(session: Session, job_id: str) -> list[Artifact]: 273 return ( 274 session.query(Artifact) 275 .filter(Artifact.job_id == job_id) 276 .order_by(Artifact.kind) 277 .all() 278 ) 279 280 281 # -------------------------------------------------------------------------- 282 # routes 283 # -------------------------------------------------------------------------- 284 285 @router.get("/languages", response_model=list[LanguageOut]) 286 def list_languages(): 287 return [ 288 LanguageOut( 289 code=l.code, name=l.name, splitter=l.splitter, 290 note=aligner.language_note(l.code), 291 ) 292 for l in languages.all_languages() 293 ] 294 295 296 @router.get("/account", response_model=AccountOut) 297 def get_account_info( 298 account: Annotated[Account, Depends(get_account)], 299 session: Annotated[Session, Depends(get_session)], 300 ): 301 return _account_out(session, account) 302 303 304 def _account_out(session: Session, account: Account) -> AccountOut: 305 ent = billing.check(session, account) 306 return AccountOut( 307 id=account.id, 308 signed_in=account.signed_in, 309 email=account.email, 310 email_verified=bool(account.email_verified), 311 billing_enabled=settings.billing_enabled, 312 payments_available=settings.payments_configured, 313 email_sign_in_available=settings.sign_in_available, 314 free_tier_summary=pricing.free_tier_summary(), 315 credits=ent.credits, 316 free_credits=ent.free_credits, 317 free_credits_with_account=( 318 0 if account.email_verified 319 else billing.free_credits_left(session, account, verified=True) 320 ), 321 daily_free_credit=settings.cloud_enabled and settings.daily_free_credit, 322 plan_credits=ent.plan_credits, 323 unlimited=ent.unlimited, 324 subscribed=ent.subscribed, 325 subscription_ends=( 326 ent.subscription_ends.isoformat() if ent.subscription_ends else None 327 ), 328 cloud_available=settings.cloud_enabled, 329 cloud_allowed=ent.cloud_allowed or not settings.billing_enabled, 330 queue_depth=queue.depth(), 331 browser_copy=settings.browser_copy, 332 ) 333 334 335 @router.get("/pricing") 336 def get_pricing(): 337 """The catalogue, for the paywall. Static - no account needed.""" 338 return { 339 "free_tier": pricing.free_tier_summary(), 340 "billing_enabled": settings.billing_enabled, 341 "payments_available": settings.payments_configured, 342 "tiers": pricing.TIER_OUTPUTS, 343 "contact_email": settings.contact_email, 344 "plans": pricing.as_dicts(), 345 } 346 347 348 # -------------------------------------------------------------------------- 349 # sign-in 350 # -------------------------------------------------------------------------- 351 352 @router.post("/auth/request") 353 def request_sign_in( 354 body: EmailIn, 355 account: Annotated[Account, Depends(get_account)], 356 session: Annotated[Session, Depends(get_session)], 357 ): 358 """Mail a one-time sign-in link.""" 359 try: 360 email = accounts.normalize_email(body.email) 361 except accounts.InvalidEmail as exc: 362 raise HTTPException(400, str(exc)) from exc 363 364 # A server that cannot send mail has no safe way to do this: the only 365 # fallback is showing the link, which would sign anyone in as anyone. 366 if not settings.sign_in_available: 367 raise HTTPException( 368 503, "Email sign-in is not set up on this server yet." 369 ) 370 371 try: 372 link = auth.issue(session, account, email) 373 except auth.TooManyRequests as exc: 374 raise HTTPException(429, str(exc)) from exc 375 376 try: 377 sent = mailer.send_login_link(email, link) 378 except mailer.MailError as exc: 379 raise HTTPException(502, str(exc)) from exc 380 381 out = {"sent": sent, "email": email} 382 if not sent and settings.dev_login_links: 383 out["dev_link"] = link 384 return out 385 386 387 @router.post("/auth/verify", response_model=AccountOut) 388 def verify_sign_in( 389 body: TokenIn, 390 response: Response, 391 account: Annotated[Account, Depends(get_account)], 392 session: Annotated[Session, Depends(get_session)], 393 ): 394 owner = auth.redeem(session, body.token, account) 395 if owner is None: 396 raise HTTPException( 397 400, "That sign-in link has expired or was already used. " 398 "Request a new one." 399 ) 400 _set_identity(response, owner) 401 return _account_out(session, owner) 402 403 404 @router.post("/auth/signout") 405 def sign_out(response: Response): 406 """Forget this browser. The account and everything in it stay put.""" 407 response.delete_cookie(DEVICE_COOKIE) 408 return {"signed_out": True} 409 410 411 # -------------------------------------------------------------------------- 412 # billing 413 # -------------------------------------------------------------------------- 414 415 @router.post("/billing/checkout") 416 def create_checkout( 417 body: CheckoutIn, 418 account: Annotated[Account, Depends(get_account)], 419 session: Annotated[Session, Depends(get_session)], 420 ): 421 plan = pricing.get(body.plan_id) 422 if plan is None: 423 raise HTTPException(404, "No such plan.") 424 try: 425 url = payments.start_checkout(session, account, plan) 426 except payments.PaymentsUnavailable as exc: 427 raise HTTPException(503, str(exc)) from exc 428 except payments.PaymentError as exc: 429 raise HTTPException(400, str(exc)) from exc 430 return {"url": url} 431 432 433 # The pages that a checkout can come back to. Any other page goes to "/", so 434 # the return trip cannot send the browser off this site. 435 RETURN_PAGES = {"/", "/subrep.html"} 436 437 438 @router.get("/billing/return") 439 def checkout_return( 440 session_id: str, 441 session: Annotated[Session, Depends(get_session)], 442 page: str = "/", 443 ): 444 """Where Stripe sends the browser after paying. 445 446 Fulfils from here as well as from the webhook, so credits are there by the 447 time the page loads rather than whenever the webhook gets round to it. 448 Identity needs no handling: if paying folded this device into an existing 449 account, get_account re-points the cookie on the very next request. 450 """ 451 try: 452 paid = payments.fulfil_by_id(session, session_id) is not None 453 except payments.PaymentsUnavailable: 454 paid = False 455 state = "paid" if paid else "pending" 456 page = page if page in RETURN_PAGES else "/" 457 return RedirectResponse(f"{page}?checkout={state}", status_code=303) 458 459 460 @router.post("/billing/webhook") 461 async def stripe_webhook( 462 request: Request, 463 session: Annotated[Session, Depends(get_session)], 464 ): 465 payload = await request.body() 466 try: 467 event = payments.verify_webhook( 468 payload, request.headers.get("stripe-signature") 469 ) 470 except payments.PaymentsUnavailable as exc: 471 raise HTTPException(503, str(exc)) from exc 472 except payments.PaymentError as exc: 473 raise HTTPException(400, str(exc)) from exc 474 payments.handle_event(session, event) 475 return {"received": True} 476 477 478 @router.post("/billing/portal") 479 def billing_portal( 480 account: Annotated[Account, Depends(get_account)], 481 ): 482 try: 483 return {"url": payments.portal_url(account)} 484 except payments.PaymentsUnavailable as exc: 485 raise HTTPException(503, str(exc)) from exc 486 except payments.PaymentError as exc: 487 raise HTTPException(400, str(exc)) from exc 488 489 490 @router.post("/uploads", response_model=UploadOut) 491 async def create_upload( 492 account: Annotated[Account, Depends(get_account)], 493 session: Annotated[Session, Depends(get_session)], 494 files: Annotated[list[UploadFile], File()], 495 ): 496 """Stage a dropped pair, work out which is which, and guess the language.""" 497 names = [f.filename or "" for f in files] 498 try: 499 pairing = detect.classify(names) 500 except detect.DetectionError as exc: 501 raise HTTPException(400, str(exc)) from exc 502 503 job_id = new_id("job") 504 paths = Paths.for_job(job_id) 505 paths.create() 506 507 try: 508 order = {name: i for i, name in enumerate(pairing.audio_names)} 509 by_name = {(f.filename or ""): f for f in files} 510 511 text_path = staged_text_path(job_id, pairing.text_name) 512 await _save(by_name[pairing.text_name], text_path) 513 514 # fb2/mobi/azw3 become something the aligner can read. Do it now, not at 515 # run time, so a book we cannot open fails while the user is watching. 516 if convert.needs_conversion(pairing.text_name): 517 try: 518 converted = convert.to_readable( 519 text_path, text_path.with_suffix("") 520 ) 521 except convert.ConversionError as exc: 522 raise HTTPException(400, str(exc)) from exc 523 if converted != text_path: 524 text_path.unlink(missing_ok=True) 525 text_path = converted 526 527 audio_paths: list[Path] = [] 528 for name in pairing.audio_names: 529 upload = by_name.get(name) 530 if upload is None: 531 raise HTTPException(400, f"Upload did not include {name}.") 532 dest = ( 533 staged_part_path(job_id, order[name] + 1, name) 534 if pairing.is_multipart 535 else staged_audio_path(job_id, name) 536 ) 537 # The staged stem is shared between audio and text, so a single 538 # audio file must not land on the text file's path. 539 if dest == text_path: 540 raise HTTPException( 541 400, "The audiobook and the book must be different formats." 542 ) 543 await _save(upload, dest) 544 audio_paths.append(dest) 545 546 if not audio_paths: 547 raise HTTPException(400, "Upload did not include any audio.") 548 549 # Custom cover art is a signed-in feature. Say so plainly rather than 550 # accepting the file and quietly ignoring it: the UI labels the drop 551 # zone before anyone picks an image, and this is the backstop. 552 cover_name = None 553 cover_blocked = False 554 if pairing.cover_name: 555 if account.signed_in: 556 upload = by_name.get(pairing.cover_name) 557 if upload is not None: 558 await _save(upload, staged_cover_path(job_id, pairing.cover_name)) 559 cover_name = pairing.cover_name 560 else: 561 cover_blocked = True 562 563 detection = detect.detect_language(detect.extract_text_sample(text_path)) 564 # Parts are merged at run time, so total the durations here. 565 durations = [probe_duration(p) for p in audio_paths] 566 duration = sum(d for d in durations if d) if any(durations) else None 567 568 # Does this text actually belong to this audio? Answering now costs a 569 # few seconds; finding out during the run costs the whole run. 570 match = matching.check( 571 audio_paths[0], 572 text_path, 573 detection.code if detection.supported else "en", 574 aligner, 575 ) 576 577 language = detection.code if detection.supported else "en" 578 lang = languages.require(language) 579 580 job = Job( 581 id=job_id, 582 account_id=account.id, 583 status=JobStatus.draft, 584 language=lang.code, 585 splitter=lang.splitter, 586 model=settings.model, 587 audio_filename=pairing.display_name, 588 text_filename=pairing.text_name, 589 audio_parts=len(audio_paths), 590 cover_filename=cover_name, 591 audio_bytes=sum(p.stat().st_size for p in audio_paths), 592 audio_duration_seconds=duration, 593 stage="Ready to start", 594 ) 595 session.add(job) 596 session.commit() 597 598 # A refused upload keeps what arrived, for debugging. The operator deletes 599 # it by hand. 600 except HTTPException: 601 raise 602 except detect.DetectionError as exc: 603 raise HTTPException(400, str(exc)) from exc 604 except Exception as exc: # noqa: BLE001 605 raise HTTPException(500, f"Upload failed: {exc}") from exc 606 607 return UploadOut( 608 job=_job_out(job, []), 609 detected=DetectionOut( 610 code=detection.code, name=detection.name, 611 confidence=round(detection.confidence, 4), 612 supported=detection.supported, 613 ), 614 match=match.as_dict(), 615 cover_requires_sign_in=cover_blocked, 616 ) 617 618 619 async def _save(upload: UploadFile, dest: Path) -> None: 620 """Stream to disk in chunks - these files run to hundreds of megabytes.""" 621 dest.parent.mkdir(parents=True, exist_ok=True) 622 written = 0 623 with dest.open("wb") as out: 624 while chunk := await upload.read(_CHUNK): 625 written += len(chunk) 626 if written > settings.max_upload_bytes: 627 raise HTTPException( 628 413, 629 f"{upload.filename} exceeds the " 630 f"{settings.max_upload_bytes // (1024**3)} GiB upload limit.", 631 ) 632 out.write(chunk) 633 634 635 @router.post("/jobs/{job_id}/start", response_model=JobOut) 636 def start_job( 637 job_id: str, 638 body: StartIn, 639 account: Annotated[Account, Depends(get_account)], 640 session: Annotated[Session, Depends(get_session)], 641 ): 642 job = _load(session, account, job_id) 643 # A failed job keeps its staged inputs, so it can be retried in place - 644 # usually after correcting the language. 645 if job.status not in (JobStatus.draft, JobStatus.failed): 646 raise HTTPException(409, f"Job is already {job.status.value}.") 647 if job.status == JobStatus.failed and not Paths.for_job(job.id).inp.exists(): 648 raise HTTPException( 649 409, "The uploaded files for this job are gone. Upload them again." 650 ) 651 652 if body.language: 653 try: 654 lang = languages.require(body.language) 655 except languages.UnsupportedLanguage as exc: 656 raise HTTPException(400, str(exc)) from exc 657 job.language, job.splitter = lang.code, lang.splitter 658 if body.model: 659 job.model = body.model 660 661 try: 662 billing.authorize_start(session, account, job) 663 except billing.PaymentRequired as exc: 664 session.rollback() 665 raise HTTPException(402, str(exc)) from exc 666 667 # With an external queue the worker is probably not this machine, so the 668 # staged inputs have to go somewhere both sides can reach before enqueuing. 669 if settings.queue_backend != "memory": 670 paths = Paths.for_job(job.id) 671 prefix = input_prefix(job.id) 672 for local in sorted(paths.inp.rglob("*")): 673 if local.is_file(): 674 rel = local.relative_to(paths.inp).as_posix() 675 storage.put_file(f"{prefix}/{rel}", local) 676 677 job.status = JobStatus.queued 678 job.stage = "Queued" 679 job.progress = 0.0 680 job.error = None 681 session.commit() 682 683 queue.enqueue(job.id, paid=billing.is_paid(job)) 684 return _job_out(job, []) 685 686 687 # -------------------------------------------------------------------------- 688 # jobs that run in the browser 689 # -------------------------------------------------------------------------- 690 # 691 # The audio never reaches us. The server's part is to say whether the job may 692 # start (the free window, or a credit), and to keep the finished subtitles so 693 # they are still there on another device. Nothing here can be enforced against 694 # someone who edits the page's JavaScript, and that is accepted: see billing.py. 695 696 _LOCAL_STALE_HOURS = 48 697 698 699 @router.post("/local/jobs", response_model=JobOut) 700 def start_local_job( 701 body: LocalJobIn, 702 account: Annotated[Account, Depends(get_account)], 703 session: Annotated[Session, Depends(get_session)], 704 ): 705 # The same book started again - a closed tab, a reload - carries on under 706 # the job it already paid for rather than being charged a second time. 707 running = ( 708 session.query(Job) 709 .filter( 710 Job.account_id == account.id, Job.local == 1, 711 Job.status == JobStatus.running, 712 Job.audio_filename == body.audio_filename, 713 Job.audio_bytes == body.audio_bytes, 714 ) 715 .order_by(Job.created_at.desc()) 716 .first() 717 ) 718 if running is not None: 719 running.language = body.language 720 session.commit() 721 return _job_out(running, []) 722 723 job = Job( 724 account_id=account.id, status=JobStatus.running, local=1, 725 language=body.language, splitter="browser", model="whisper-tiny", 726 audio_filename=body.audio_filename, text_filename=body.text_filename, 727 audio_parts=body.audio_parts, audio_bytes=body.audio_bytes, 728 audio_duration_seconds=body.audio_duration_seconds, 729 stage="Running in your browser", started_at=utcnow(), 730 ) 731 session.add(job) 732 try: 733 billing.authorize_start(session, account, job) 734 except billing.PaymentRequired as exc: 735 session.rollback() 736 raise HTTPException(402, str(exc)) from exc 737 session.commit() 738 return _job_out(job, []) 739 740 741 @router.post("/local/jobs/{job_id}/finish", response_model=JobOut) 742 def finish_local_job( 743 job_id: str, 744 body: LocalFinishIn, 745 account: Annotated[Account, Depends(get_account)], 746 session: Annotated[Session, Depends(get_session)], 747 ): 748 job = _load(session, account, job_id) 749 if not job.local or job.status != JobStatus.running: 750 raise HTTPException(409, f"Job is already {job.status.value}.") 751 752 import json 753 import tempfile 754 755 name = Path(body.filename).name or "subtitles.srt" 756 with tempfile.TemporaryDirectory() as tmp: 757 for kind, filename, content in ( 758 ("srt", name, body.srt), 759 ("metadata", "metadata.json", 760 json.dumps({"job_id": job.id, **body.metadata}, ensure_ascii=False, indent=2)), 761 ): 762 src = Path(tmp) / filename 763 # Bytes, not text: on Windows write_text would turn every line ending into CRLF. 764 src.write_bytes(content.encode("utf-8")) 765 key = f"{job.id}/{filename}" 766 size = storage.put_file(key, src) 767 session.add(Artifact(job_id=job.id, kind=kind, filename=filename, 768 storage_key=key, size_bytes=size)) 769 770 if settings.browser_copy: 771 # The subtitles go next to the copy of the audio and the book. 772 root = Paths.for_job(job.id).root 773 root.mkdir(parents=True, exist_ok=True) 774 (root / (_plain_name(body.filename) or "subtitles.srt")).write_bytes( 775 body.srt.encode("utf-8")) 776 777 job.status = JobStatus.succeeded 778 job.stage, job.progress, job.finished_at = "Done", 1.0, utcnow() 779 session.commit() 780 return _job_out(job, _artifacts(session, job.id)) 781 782 783 def _plain_name(raw: str) -> str: 784 """The last part of a file name from a browser, with no folder in it.""" 785 name = re.split(r"[\\/]", raw or "")[-1].strip() 786 return "" if name in ("", ".", "..") else name 787 788 789 @router.post("/local/jobs/{job_id}/files") 790 async def copy_local_job_files( 791 job_id: str, 792 account: Annotated[Account, Depends(get_account)], 793 session: Annotated[Session, Depends(get_session)], 794 files: Annotated[list[UploadFile], File()], 795 ): 796 """Keep a copy of the audio and the book of a job in the browser. 797 798 Only during the beta (settings.browser_copy), for debugging, as the 799 privacy and user data policy says. The copy goes with the other files of 800 the job, and it costs no credit. 801 """ 802 if not settings.browser_copy: 803 raise HTTPException(404, "Not found") 804 job = _load(session, account, job_id) 805 if not job.local: 806 raise HTTPException(409, "A server job has its files already.") 807 inp = Paths.for_job(job.id).inp 808 saved = 0 809 for upload in files: 810 name = _plain_name(upload.filename or "") 811 if name: 812 await _save(upload, inp / name) 813 saved += 1 814 return {"saved": saved} 815 816 817 @router.post("/local/jobs/{job_id}/fail", response_model=JobOut) 818 def fail_local_job( 819 job_id: str, 820 body: LocalFailIn, 821 account: Annotated[Account, Depends(get_account)], 822 session: Annotated[Session, Depends(get_session)], 823 ): 824 job = _load(session, account, job_id) 825 if not job.local or job.status != JobStatus.running: 826 raise HTTPException(409, f"Job is already {job.status.value}.") 827 job.status = JobStatus.failed 828 job.stage, job.error, job.finished_at = "Failed", body.error or "Failed in the browser.", utcnow() 829 # Whatever went wrong, they got nothing: the free slot or the credit goes back. 830 billing.refund(session, job) 831 session.commit() 832 return _job_out(job, []) 833 834 835 def expire_stale_local_jobs(session: Session) -> int: 836 """A tab that was closed for good never reports back. Release what it held.""" 837 from datetime import timedelta 838 839 stale = ( 840 session.query(Job) 841 .filter(Job.local == 1, Job.status == JobStatus.running, 842 Job.created_at < utcnow() - timedelta(hours=_LOCAL_STALE_HOURS)) 843 .all() 844 ) 845 for job in stale: 846 job.status, job.stage, job.finished_at = JobStatus.canceled, "Abandoned", utcnow() 847 billing.refund(session, job) 848 session.commit() 849 return len(stale) 850 851 852 @router.post("/convert") 853 async def convert_book(file: Annotated[UploadFile, File()]): 854 """mobi / azw3 to epub. The one thing the browser cannot do for itself: 855 those formats need a real parser, and a book is small enough to send.""" 856 name = Path(file.filename or "book").name 857 if not convert.needs_conversion(name): 858 raise HTTPException(400, "That format does not need converting.") 859 import tempfile 860 861 with tempfile.TemporaryDirectory() as tmp: 862 src = Path(tmp) / name 863 await _save(file, src) 864 try: 865 out = convert.to_readable(src, Path(tmp) / "converted") 866 except convert.ConversionError as exc: 867 raise HTTPException(400, str(exc)) from exc 868 data = out.read_bytes() 869 return Response(data, media_type="application/epub+zip", 870 headers={"X-Filename": Path(name).stem + ".epub"}) 871 872 873 @router.get("/jobs", response_model=list[JobOut]) 874 def list_jobs( 875 account: Annotated[Account, Depends(get_account)], 876 session: Annotated[Session, Depends(get_session)], 877 ): 878 jobs = ( 879 session.query(Job) 880 .filter(Job.account_id == account.id) 881 .order_by(Job.created_at.desc()) 882 .limit(50) 883 .all() 884 ) 885 return [_job_out(j, _artifacts(session, j.id)) for j in jobs] 886 887 888 @router.get("/jobs/{job_id}", response_model=JobOut) 889 def get_job( 890 job_id: str, 891 account: Annotated[Account, Depends(get_account)], 892 session: Annotated[Session, Depends(get_session)], 893 ): 894 job = _load(session, account, job_id) 895 return _job_out(job, _artifacts(session, job.id)) 896 897 898 @router.post("/jobs/{job_id}/cancel", response_model=JobOut) 899 def cancel_job( 900 job_id: str, 901 account: Annotated[Account, Depends(get_account)], 902 session: Annotated[Session, Depends(get_session)], 903 ): 904 job = _load(session, account, job_id) 905 if job.status in (JobStatus.succeeded, JobStatus.failed, JobStatus.canceled): 906 raise HTTPException(409, f"Job is already {job.status.value}.") 907 job.status = JobStatus.canceled 908 job.stage = "Canceled" 909 job.finished_at = utcnow() 910 # A canceled job must not eat the free conversion. 911 billing.refund(session, job) 912 session.commit() 913 return _job_out(job, []) 914 915 916 @router.delete("/jobs/{job_id}") 917 def delete_job( 918 job_id: str, 919 account: Annotated[Account, Depends(get_account)], 920 session: Annotated[Session, Depends(get_session)], 921 ): 922 """Take the job off the visitor's list. Its files stay on the server for 923 debugging, and the operator deletes them by hand.""" 924 job = _load(session, account, job_id) 925 if job.status in (JobStatus.queued, JobStatus.running): 926 raise HTTPException(409, "Cancel the job before deleting it.") 927 session.delete(job) 928 session.commit() 929 return {"deleted": job_id} 930 931 932 @router.get("/jobs/{job_id}/files/{kind}") 933 def download( 934 job_id: str, 935 kind: Literal["srt", "video", "video_embedded", "metadata", "log"], 936 account: Annotated[Account, Depends(get_account)], 937 session: Annotated[Session, Depends(get_session)], 938 ): 939 job = _load(session, account, job_id) 940 art = ( 941 session.query(Artifact) 942 .filter(Artifact.job_id == job.id, Artifact.kind == kind) 943 .first() 944 ) 945 if art is None: 946 raise HTTPException(404, f"No {kind} for this job.") 947 948 # S3 hands the browser a presigned URL; local storage serves the file. 949 url = storage.presigned_url(art.storage_key, art.filename) 950 if url: 951 return RedirectResponse(url, status_code=307) 952 953 assert isinstance(storage, LocalStorage) 954 return FileResponse( 955 storage.path_for(art.storage_key), 956 filename=art.filename, 957 media_type="application/octet-stream", 958 )