#!/usr/bin/env python3 """Free single-file API example. Python 3 + ffprobe; no key, no paid processing. Usage: python3 free-api.py remove-noise interview.wav results python3 free-api.py cut interview.wav results '{"mode":"keep","ranges":[[1,4]]}' Inspect this source before running. It never executes a returned shell command. """ import json import mimetypes import re import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from pathlib import Path BASE = "https://audo.ai" # Audo's hosts: the site, and files.audo.ai, where file links may be served (only file links). FILE_HOST = "files.audo.ai" # 100 MB, the most one request to Audo takes. MAX_BYTES = 100_000_000 class ClientError(Exception): pass class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ClientError("Redirect refused; review the destination separately.") OPENER = urllib.request.build_opener(NoRedirect) def checked_url(url, prefix): if not isinstance(url, str) or any(c.isspace() for c in url) or "\\" in url: raise ClientError("Malformed transfer URL.") parsed = urllib.parse.urlsplit(url) hosts = ("audo.ai", FILE_HOST) if prefix == "/api/files/" else ("audo.ai",) if (parsed.scheme != "https" or parsed.hostname not in hosts or parsed.port not in (None, 443) or parsed.username or parsed.password or not parsed.path.startswith(prefix) or parsed.fragment): raise ClientError("Unexpected transfer origin or path.") return url def request(url, data=None, headers=None): req = urllib.request.Request(url, data=data, headers={"User-Agent": "Audo-free-example/1", **(headers or {})}) try: with OPENER.open(req, timeout=50) as response: body = response.read(MAX_BYTES + 1) if len(body) > MAX_BYTES: raise ClientError("Response exceeds the example's 100 MB limit.") return response.status, response.headers.get_content_type(), body, response.headers except urllib.error.HTTPError as exc: # Never print URLs, provider messages, or bodies: they may contain tokens. retry = exc.headers.get("Retry-After", "") retry = retry if retry.isdigit() else "unspecified" raise ClientError(f"HTTP {exc.code}; request failed. Retry-After seconds: {retry}") from None except (urllib.error.URLError, TimeoutError): raise ClientError("Network request failed; no completed media was saved.") from None def parse_json(body): try: return json.loads(body) except (ValueError, UnicodeError): raise ClientError("Invalid JSON response.") from None def validate(body, kind): if kind == "application/json": parse_json(body) return ".json" if kind in ("application/x-subrip", "text/vtt", "text/plain"): try: text = body.decode("utf-8") except UnicodeError: raise ClientError("Invalid text output.") from None if kind == "text/vtt" and not text.startswith("WEBVTT"): raise ClientError("Invalid VTT output.") # No-speech transcripts have no SRT cues and serialize as an empty string. if kind == "application/x-subrip" and text != "" and "-->" not in text: raise ClientError("Invalid SRT output.") return {"text/vtt": ".vtt", "application/x-subrip": ".srt", "text/plain": ".txt"}[kind] if not kind.startswith(("audio/", "video/")): raise ClientError("Unexpected output content type.") # Content-Type alone is not enough: ffprobe must recognize a media stream. with tempfile.NamedTemporaryFile() as tmp: tmp.write(body) tmp.flush() try: probe = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "stream=codec_type", "-of", "json", tmp.name], capture_output=True, timeout=30, check=True) streams = parse_json(probe.stdout).get("streams", []) except (OSError, subprocess.SubprocessError): raise ClientError("Media validation failed; install ffprobe and check the output.") from None if not any(s.get("codec_type") in ("audio", "video") for s in streams): raise ClientError("No media stream found.") return mimetypes.guess_extension(kind) or ".media" def finish(response, folder): status, kind, body, headers = response if status not in (200, 202): raise ClientError(f"Unexpected HTTP {status}.") job = parse_json(body) if kind == "application/json" else None if status == 202 and not isinstance(job, dict): raise ClientError("Expected a JSON job for HTTP 202.") deadline = time.monotonic() + 1800 while isinstance(job, dict) and "status" in job: state = job["status"] if state in ("failed", "cancelled"): raise ClientError(f"Job {state}; no completed media was saved.") if state == "succeeded": break if state not in ("queued", "running") or time.monotonic() >= deadline: raise ClientError("Unknown job state or polling deadline exceeded.") url = checked_url(job.get("statusUrl", ""), "/api/try/jobs/") # Poll only the returned same-origin link, never log its token. status, kind, body, headers = request(url) if status not in (200, 202) or kind != "application/json": raise ClientError("Unexpected polling response.") job = parse_json(body) if not isinstance(job, dict) or "status" not in job: raise ClientError("Invalid polling job.") if job["status"] not in ("succeeded", "failed", "cancelled"): time.sleep(2) results = [] if isinstance(job, dict) and "status" in job: if job["status"] != "succeeded": raise ClientError("Job has not succeeded.") for output in job.get("outputs", []): url = checked_url(output.get("url", ""), "/api/files/") code, output_kind, output_body, _ = request(url) if code != 200 or output_kind != output.get("contentType"): raise ClientError("Unexpected download status or content type.") results.append((output_body, validate(output_body, output_kind))) # Keep the useful result, but never persist capability links/tokens. summary = {k: job.get(k) for k in ("free", "processedSeconds", "result")} results.append((json.dumps(summary).encode(), ".json")) processed = job.get("processedSeconds") else: if status != 200: raise ClientError("Incomplete response.") results.append((body, validate(body, kind))) processed = headers.get("Audo-Processed-Seconds", "unknown") # Create a fresh directory only after every output has validated; never overwrite originals. folder.mkdir(parents=True, exist_ok=False) for index, (contents, extension) in enumerate(results, 1): (folder / f"result-{index}{extension}").write_bytes(contents) return processed def main(): if len(sys.argv) not in (4, 5): raise ClientError("Usage: free-api.py TOOL FILE NEW_OUTPUT_FOLDER [JSON_OPTIONS]") tool, filename, folder = sys.argv[1:4] if not re.fullmatch(r"[a-z]+(?:-[a-z]+)*", tool): raise ClientError("Invalid tool slug.") options = parse_json(sys.argv[4]) if len(sys.argv) == 5 else {} if not isinstance(options, dict) or any(not re.fullmatch(r"[a-z_]+", k) for k in options): raise ClientError("Options must be an object with field names.") if tool == "cut" and "mode" not in options: raise ClientError("Specify cut mode: keep or remove.") media = Path(filename).read_bytes() if len(media) > MAX_BYTES - 8192: raise ClientError("File too large for this in-memory example.") boundary = "audo-example-boundary" parts = [] for key, value in options.items(): value = value if isinstance(value, str) else json.dumps(value) parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n'.encode()) parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="input"\r\nContent-Type: application/octet-stream\r\n\r\n'.encode() + media + b"\r\n") parts.append(f"--{boundary}--\r\n".encode()) response = request(f"{BASE}/api/try/{tool}", b"".join(parts), {"Content-Type": f"multipart/form-data; boundary={boundary}"}) processed = finish(response, Path(folder)) print(f"Validated results saved. Processed seconds: {processed}. This no-key request uses the free portion.") if __name__ == "__main__": try: main() except ClientError as exc: print(str(exc), file=sys.stderr) sys.exit(1) except (OSError, ValueError): # Errors intentionally omit the exception text, which may include a local name or URL. print("Request or validation failed; no successful result is claimed. Check inputs, quota, and ffprobe availability.", file=sys.stderr) sys.exit(1)