#!/usr/bin/env python3 """transkriptor-transcribe — Adapter für transkriptor.com Pro-API. 3-Step lokaler File-Upload: 1. POST get_upload_url → {upload_url, public_url} 2. PUT raw binary → upload_url (S3 presigned) 3. POST initiate_transcription mit public_url → {order_id} 4. GET /files/{order_id}/content (polling bis status=Completed) URL-Mode (YouTube/Drive/Dropbox/OneDrive): 1. POST /transcription/url mit {url, service, language} → {order_id} 2. GET /files/{order_id}/content (polling) Auth: API-Key in `/root/.aria-secrets/transkriptor.env` TRANSKRIPTOR_API_KEY= CLI: python3 transkriptor-transcribe.py file [--language en-US] [--service Standard] python3 transkriptor-transcribe.py url [--language en-US] python3 transkriptor-transcribe.py result # nur ergebnis abholen """ from __future__ import annotations import argparse import json import os import sys import time import urllib.parse import urllib.request from pathlib import Path API_BASE = "https://api.tor.app/developer" USER_AGENT = "aria-transkriptor/0.1" TIMEOUT = 60 POLL_MAX_SECONDS = 900 # 15 min cap POLL_INTERVAL_INITIAL = 5 POLL_INTERVAL_MAX = 30 SECRETS_PATH = "/root/.aria-secrets/transkriptor.env" def _load_api_key() -> str: key = os.environ.get("TRANSKRIPTOR_API_KEY") if key: return key if os.path.exists(SECRETS_PATH): for line in Path(SECRETS_PATH).read_text().splitlines(): line = line.strip() if line.startswith("TRANSKRIPTOR_API_KEY="): return line.split("=", 1)[1].strip().strip('"').strip("'") raise SystemExit( f"ERROR: TRANSKRIPTOR_API_KEY missing.\n" f" Place in {SECRETS_PATH} as TRANSKRIPTOR_API_KEY=\n" f" (chmod 600 — siehe feedback_no_secrets_via_telegram standing order)" ) def _auth_headers(api_key: str) -> dict: return { "Authorization": f"Bearer {api_key}", "Accept": "application/json", "User-Agent": USER_AGENT, } def _post_json(path: str, body: dict, api_key: str) -> dict: url = f"{API_BASE}{path}" data = json.dumps(body).encode() headers = {**_auth_headers(api_key), "Content-Type": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return json.loads(resp.read()) def _put_binary(url: str, file_path: Path) -> None: data = file_path.read_bytes() req = urllib.request.Request(url, data=data, method="PUT") with urllib.request.urlopen(req, timeout=TIMEOUT * 5) as resp: if resp.status not in (200, 201): raise RuntimeError(f"upload failed: {resp.status}") def _get_json(path: str, api_key: str) -> dict: url = f"{API_BASE}{path}" req = urllib.request.Request(url, headers=_auth_headers(api_key), method="GET") with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return json.loads(resp.read()) def transcribe_file(file_path: Path, *, language: str, service: str, api_key: str) -> dict: """3-Step lokaler Upload + Transcribe-Init. Returns dict mit `order_id`.""" if not file_path.exists(): raise FileNotFoundError(file_path) sys.stderr.write(f"[1/3] get_upload_url for {file_path.name}\n") step1 = _post_json("/transcription/local_file/get_upload_url", {"file_name": file_path.name}, api_key) upload_url = step1["upload_url"] public_url = step1["public_url"] sys.stderr.write(f"[2/3] PUT {file_path.stat().st_size} bytes\n") _put_binary(upload_url, file_path) sys.stderr.write(f"[3/3] initiate_transcription (lang={language}, service={service})\n") step3 = _post_json("/transcription/local_file/initiate_transcription", {"url": public_url, "language": language, "service": service}, api_key) return {"order_id": step3.get("order_id"), "message": step3.get("message")} def transcribe_url(url: str, *, language: str, service: str, api_key: str) -> dict: """URL-Mode für YouTube/Drive/Dropbox/OneDrive.""" sys.stderr.write(f"[1/1] transcription/url ({url[:80]}...)\n") return _post_json("/transcription/url", {"url": url, "language": language, "service": service}, api_key) def poll_result(order_id: str, *, api_key: str) -> dict: """Pollt /files/{order_id}/content bis status=Completed.""" start = time.monotonic() interval = POLL_INTERVAL_INITIAL attempt = 0 while True: attempt += 1 elapsed = time.monotonic() - start if elapsed > POLL_MAX_SECONDS: raise TimeoutError(f"poll timeout after {POLL_MAX_SECONDS}s, order_id={order_id}") result = _get_json(f"/files/{order_id}/content", api_key) body = result.get("body") or result status = body.get("status") or (body.get("data", {}) if isinstance(body.get("data"), dict) else {}).get("status") sys.stderr.write(f" [poll {attempt}] elapsed={int(elapsed)}s status={status}\n") if status == "Completed": return result if status in ("Failed", "Error"): raise RuntimeError(f"transcription failed: {json.dumps(result)[:300]}") time.sleep(interval) interval = min(int(interval * 1.5), POLL_INTERVAL_MAX) def _extract_plain_text(result: dict) -> str: body = result.get("body") or result content = body.get("content") if isinstance(content, list): chunks = [] for seg in content: if isinstance(seg, dict): txt = seg.get("text") or seg.get("content") or "" speaker = seg.get("speaker") or seg.get("speaker_name") or "" if speaker: chunks.append(f"{speaker}: {txt}") else: chunks.append(txt) elif isinstance(seg, str): chunks.append(seg) return "\n".join(chunks) if isinstance(content, str): return content if isinstance(body.get("text"), str): return body["text"] return json.dumps(result, indent=2, ensure_ascii=False) def main() -> int: parser = argparse.ArgumentParser(description="Transkriptor.com adapter") sub = parser.add_subparsers(dest="cmd", required=True) p_file = sub.add_parser("file", help="Transcribe a local audio/video file") p_file.add_argument("path", help="path to local file") p_file.add_argument("--language", default="en-US") p_file.add_argument("--service", default="Standard", choices=["Standard", "Subtitle"]) p_file.add_argument("--wait", action="store_true", help="poll for result + print transcript") p_file.add_argument("--json", action="store_true", help="output raw JSON") p_url = sub.add_parser("url", help="Transcribe a URL (YouTube/Drive/Dropbox)") p_url.add_argument("source_url") p_url.add_argument("--language", default="en-US") p_url.add_argument("--service", default="Standard", choices=["Standard", "Subtitle"]) p_url.add_argument("--wait", action="store_true") p_url.add_argument("--json", action="store_true") p_res = sub.add_parser("result", help="Fetch result for an order_id") p_res.add_argument("order_id") p_res.add_argument("--wait", action="store_true", help="poll until Completed") p_res.add_argument("--json", action="store_true") args = parser.parse_args() api_key = _load_api_key() if args.cmd == "file": init = transcribe_file(Path(args.path), language=args.language, service=args.service, api_key=api_key) order_id = init.get("order_id") if not order_id: print(json.dumps(init, indent=2)) return 1 sys.stderr.write(f"order_id={order_id}\n") if not args.wait: print(order_id) return 0 result = poll_result(order_id, api_key=api_key) print(json.dumps(result, indent=2) if args.json else _extract_plain_text(result)) return 0 if args.cmd == "url": init = transcribe_url(args.source_url, language=args.language, service=args.service, api_key=api_key) order_id = init.get("order_id") if not order_id: print(json.dumps(init, indent=2)) return 1 sys.stderr.write(f"order_id={order_id}\n") if not args.wait: print(order_id) return 0 result = poll_result(order_id, api_key=api_key) print(json.dumps(result, indent=2) if args.json else _extract_plain_text(result)) return 0 if args.cmd == "result": if args.wait: result = poll_result(args.order_id, api_key=api_key) else: result = _get_json(f"/files/{args.order_id}/content", api_key=api_key) print(json.dumps(result, indent=2) if args.json else _extract_plain_text(result)) return 0 return 2 if __name__ == "__main__": sys.exit(main())