#!/usr/bin/env python3 """VideoClick v2 server-to-server upload. Python 3.8+, standard library only.""" import argparse import json import os import time import uuid from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import quote from urllib.parse import urlencode from urllib.request import Request, urlopen BASE = os.environ.get("VIDEOCLICK_API_BASE", "https://videoclick.ir/api/v2").rstrip("/") SECRET = os.environ.get("VIDEOCLICK_CLIENT_SECRET", "") CLIENT_ID = os.environ.get("VIDEOCLICK_CLIENT_ID", SECRET[:16]) ACCESS_TOKEN = None TOKEN_DEADLINE = 0.0 class ApiError(Exception): def __init__(self, status, payload, retry_after=None): self.status = status self.payload = payload self.retry_after = retry_after super().__init__(f"HTTP {status}: {payload}") def get_access_token(): global ACCESS_TOKEN, TOKEN_DEADLINE if ACCESS_TOKEN and time.monotonic() < TOKEN_DEADLINE - 60: return ACCESS_TOKEN payload = urlencode({"grant_type": "client_credentials", "client_id": CLIENT_ID, "client_secret": SECRET, "scope": "groups.read video.read video.upload"}).encode("ascii") request = Request(BASE + "/oauth/token", data=payload, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") try: with urlopen(request, timeout=30) as response: result = json.loads(response.read().decode("utf-8")) except HTTPError as error: raise ApiError(error.code, error.read().decode("utf-8", errors="replace")[:500]) from error ACCESS_TOKEN = result["access_token"] TOKEN_DEADLINE = time.monotonic() + int(result["expires_in"]) return ACCESS_TOKEN def call(method, path, data=None, content_type=None): headers = {"Authorization": "Bearer " + get_access_token(), "Accept": "application/json"} if content_type: headers["Content-Type"] = content_type request = Request(BASE + path, data=data, headers=headers, method=method) try: with urlopen(request, timeout=120) as response: return json.loads(response.read().decode("utf-8")) except HTTPError as error: raw = error.read().decode("utf-8", errors="replace") try: payload = json.loads(raw) except ValueError: payload = raw[:500] # An unexpected 401 is surfaced to the caller. Completion retries use # the same uploadId and the destination's stable video identity. raise ApiError(error.code, payload, error.headers.get("Retry-After")) from error def get_upload(upload_id): return call("GET", "/upload-sessions/" + quote(upload_id, safe=""))["upload"] def complete_with_recovery(upload_id): for attempt in range(6): try: return call("POST", f"/upload-sessions/{upload_id}/complete", b"") except (ApiError, URLError) as exc: current = get_upload(upload_id) if current["State"] == "completed": return {"videoId": current["VideoId"]} if isinstance(exc, ApiError) and exc.status not in (409, 502, 503): raise if attempt == 5: raise RuntimeError("Complete is still pending; rerun with the same uploadId") from exc time.sleep(min(5 * (attempt + 1), 30)) def main(): parser = argparse.ArgumentParser(description="Upload one MP4/WebM video to VideoClick") parser.add_argument("file", type=Path) parser.add_argument("--group-id", type=int, required=True) parser.add_argument("--duration", type=int, required=True, help="Duration in seconds") parser.add_argument("--title", required=True) parser.add_argument("--upload-id", help="Resume an existing upload instead of creating a new one") parser.add_argument("--request-id", help="Stable idempotency key when creating/retrying the same session") args = parser.parse_args() if not SECRET.startswith("vck_") or len(SECRET) != 47 or CLIENT_ID != SECRET[:16]: parser.error("Set VIDEOCLICK_CLIENT_SECRET to your vck_ key and VIDEOCLICK_CLIENT_ID to its first 16 characters") if not args.file.is_file(): parser.error("File does not exist") size = args.file.stat().st_size capabilities = call("GET", "/capabilities") groups = call("GET", "/groups")["groups"] if not any(group["groupId"] == args.group_id for group in groups): parser.error("group-id does not belong to this account") chunk_size = capabilities["chunkSize"] if size > capabilities["maxFileBytes"]: parser.error("File exceeds the account API size limit") if args.upload_id: upload_id = args.upload_id upload = get_upload(upload_id) if upload["FileSize"] != size: parser.error("The local file size differs from the existing session") else: request_id = args.request_id or uuid.uuid4().hex print("Request ID:", request_id) upload = call("POST", "/upload-sessions", json.dumps({ "groupId": args.group_id, "title": args.title, "fileName": args.file.name, "fileSize": size, "durationSeconds": args.duration, "idempotencyKey": request_id, }, ensure_ascii=False).encode("utf-8"), "application/json")["upload"] upload_id = upload["uploadId"] print("Upload ID:", upload_id, "(save it to resume with --upload-id)") if upload["State"] == "uploading": offset = upload["NextOffset"] with args.file.open("rb") as stream: while offset < size: stream.seek(offset) piece = stream.read(min(chunk_size, size - offset)) failures = 0 while True: try: result = call("PUT", f"/upload-sessions/{upload_id}/chunks?offset={offset}", piece, "application/octet-stream") offset = result["nextOffset"] print(f"Transferred: {offset}/{size} bytes") break except (ApiError, URLError) as error: failures += 1 if failures > 5: raise current = get_upload(upload_id) new_offset = current["NextOffset"] if current["State"] != "uploading": raise RuntimeError("Upload state changed; inspect the session") from error if new_offset != offset: offset = new_offset break delay = int(error.retry_after or 2) if isinstance(error, ApiError) else 2 time.sleep(max(1, min(delay, 30))) result = complete_with_recovery(upload_id) video_id = result.get("videoId") or result.get("upload", {}).get("VideoId") elif upload["State"] == "completed": video_id = upload["VideoId"] elif upload["State"] in ("uncertain", "transferring"): result = complete_with_recovery(upload_id) video_id = result.get("videoId") or result.get("upload", {}).get("VideoId") else: raise RuntimeError(f"Session state: {upload['State']}; inspect before retrying") print("Video ID:", video_id) if not video_id: raise RuntimeError("Video ID is missing; inspect the upload status") for _ in range(180): state = call("GET", "/videos/" + quote(str(video_id), safe=""))["state"] print("Processing:", state) if state == "ready": print("Video is ready.") return if state == "failed": raise RuntimeError("Processing failed") time.sleep(10) print("Processing is still in progress; check the video status later.") if __name__ == "__main__": main()