skills/captcha-third-party-services/scripts/captcha_api_cli.py

#!/usr/bin/env python3
"""Unified CLI for 2Captcha, YesCaptcha, and Anti-Captcha API methods."""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Dict


PROVIDERS = {
    "2captcha": {
        "endpoint": "https://api.2captcha.com",
        "env_var": "CAPTCHA_2CAPTCHA_KEY",
    },
    "yescaptcha": {
        "endpoint": "https://api.yescaptcha.com",
        "env_var": "CAPTCHA_YESCAPTCHA_KEY",
    },
    "anti-captcha": {
        "endpoint": "https://api.anti-captcha.com",
        "env_var": "CAPTCHA_ANTI_CAPTCHA_KEY",
    },
}

PROVIDER_ALIASES = {
    "2captcha": "2captcha",
    "2-captcha": "2captcha",
    "yescaptcha": "yescaptcha",
    "yes-captcha": "yescaptcha",
    "anticaptcha": "anti-captcha",
    "anti-captcha": "anti-captcha",
}


class CliError(Exception):
    """Raised for expected CLI errors."""


def normalize_provider(provider: str) -> str:
    key = provider.strip().lower()
    normalized = PROVIDER_ALIASES.get(key)
    if not normalized:
        supported = ", ".join(sorted(PROVIDERS))
        raise CliError(f"Unsupported provider '{provider}'. Supported: {supported}")
    return normalized


def resolve_api_key(provider: str, explicit_key: str | None) -> str:
    if explicit_key:
        return explicit_key
    env_var = PROVIDERS[provider]["env_var"]
    value = os.getenv(env_var, "").strip()
    if value:
        return value
    raise CliError(
        f"Missing API key for provider '{provider}'. "
        f"Set --api-key or environment variable {env_var}."
    )


def call_api(provider: str, method: str, payload: Dict[str, Any], timeout: int = 60) -> Dict[str, Any]:
    base = PROVIDERS[provider]["endpoint"]
    url = f"{base}/{method}"
    body = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        url=url,
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            raw = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise CliError(f"HTTP {exc.code} from {url}: {detail}") from exc
    except urllib.error.URLError as exc:
        raise CliError(f"Network error calling {url}: {exc}") from exc

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise CliError(f"Invalid JSON response from {url}: {raw}") from exc

    return data


def ensure_api_success(response: Dict[str, Any]) -> None:
    error_id = response.get("errorId")
    if error_id in (0, "0", None):
        return

    code = response.get("errorCode", "UNKNOWN")
    desc = response.get("errorDescription", "")
    message = f"Provider API error (errorId={error_id}, errorCode={code})"
    if desc:
        message = f"{message}: {desc}"
    raise CliError(message)


def load_task_payload(task_json: str | None, task_file: str | None) -> Dict[str, Any]:
    if not task_json and not task_file:
        raise CliError("Provide either --task-json or --task-file.")

    if task_json and task_file:
        raise CliError("Use only one of --task-json or --task-file.")

    if task_json:
        source = task_json
    else:
        path = Path(task_file or "")
        if not path.exists():
            raise CliError(f"Task file does not exist: {path}")
        source = path.read_text()

    try:
        task = json.loads(source)
    except json.JSONDecodeError as exc:
        raise CliError(f"Invalid task JSON: {exc}") from exc

    if not isinstance(task, dict):
        raise CliError("Task JSON must be an object.")
    if "type" not in task:
        raise CliError("Task JSON must include a 'type' field.")
    return task


def print_json(data: Dict[str, Any]) -> None:
    print(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True))


def cmd_balance(args: argparse.Namespace) -> int:
    provider = normalize_provider(args.provider)
    key = resolve_api_key(provider, args.api_key)
    payload = {"clientKey": key}
    result = call_api(provider, "getBalance", payload)
    ensure_api_success(result)
    print_json({"provider": provider, "result": result})
    return 0


def cmd_create_task(args: argparse.Namespace) -> int:
    provider = normalize_provider(args.provider)
    key = resolve_api_key(provider, args.api_key)
    task = load_task_payload(args.task_json, args.task_file)

    payload: Dict[str, Any] = {"clientKey": key, "task": task}
    if args.soft_id is not None:
        payload["softId"] = args.soft_id

    result = call_api(provider, "createTask", payload)
    ensure_api_success(result)
    print_json({"provider": provider, "result": result})
    return 0


def cmd_get_task_result(args: argparse.Namespace) -> int:
    provider = normalize_provider(args.provider)
    key = resolve_api_key(provider, args.api_key)
    payload = {"clientKey": key, "taskId": args.task_id}

    result = call_api(provider, "getTaskResult", payload)
    ensure_api_success(result)
    print_json({"provider": provider, "result": result})
    return 0


def cmd_solve(args: argparse.Namespace) -> int:
    provider = normalize_provider(args.provider)
    key = resolve_api_key(provider, args.api_key)
    task = load_task_payload(args.task_json, args.task_file)

    create_payload: Dict[str, Any] = {"clientKey": key, "task": task}
    if args.soft_id is not None:
        create_payload["softId"] = args.soft_id

    create_result = call_api(provider, "createTask", create_payload)
    ensure_api_success(create_result)

    task_id = create_result.get("taskId")
    if not task_id:
        raise CliError(f"createTask succeeded but taskId missing: {create_result}")

    deadline = time.time() + args.timeout

    while time.time() < deadline:
        poll_payload = {"clientKey": key, "taskId": task_id}
        poll_result = call_api(provider, "getTaskResult", poll_payload)
        ensure_api_success(poll_result)

        status = poll_result.get("status")
        if status == "ready":
            print_json(
                {
                    "provider": provider,
                    "taskId": task_id,
                    "createTask": create_result,
                    "getTaskResult": poll_result,
                }
            )
            return 0

        if status != "processing":
            raise CliError(f"Unexpected getTaskResult status '{status}': {poll_result}")

        time.sleep(args.poll_interval)

    raise CliError(
        f"Timeout waiting for task {task_id} after {args.timeout} seconds "
        f"(poll interval {args.poll_interval}s)."
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Unified captcha API CLI for 2Captcha, YesCaptcha, Anti-Captcha."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    p_balance = subparsers.add_parser("balance", help="Call getBalance")
    p_balance.add_argument("--provider", required=True, help="2captcha|yescaptcha|anti-captcha")
    p_balance.add_argument("--api-key", help="Provider API key (optional if env var is set)")
    p_balance.set_defaults(func=cmd_balance)

    p_create = subparsers.add_parser("create-task", help="Call createTask")
    p_create.add_argument("--provider", required=True, help="2captcha|yescaptcha|anti-captcha")
    p_create.add_argument("--api-key", help="Provider API key (optional if env var is set)")
    p_create.add_argument("--task-json", help="Inline JSON object for 'task'")
    p_create.add_argument("--task-file", help="Path to JSON file containing 'task' object")
    p_create.add_argument("--soft-id", type=int, help="Optional softId value")
    p_create.set_defaults(func=cmd_create_task)

    p_result = subparsers.add_parser("get-task-result", help="Call getTaskResult")
    p_result.add_argument("--provider", required=True, help="2captcha|yescaptcha|anti-captcha")
    p_result.add_argument("--api-key", help="Provider API key (optional if env var is set)")
    p_result.add_argument("--task-id", required=True, type=int, help="Task ID from createTask")
    p_result.set_defaults(func=cmd_get_task_result)

    p_solve = subparsers.add_parser("solve", help="Create task and poll until ready")
    p_solve.add_argument("--provider", required=True, help="2captcha|yescaptcha|anti-captcha")
    p_solve.add_argument("--api-key", help="Provider API key (optional if env var is set)")
    p_solve.add_argument("--task-json", help="Inline JSON object for 'task'")
    p_solve.add_argument("--task-file", help="Path to JSON file containing 'task' object")
    p_solve.add_argument("--soft-id", type=int, help="Optional softId value")
    p_solve.add_argument("--poll-interval", type=int, default=3, help="Polling interval in seconds")
    p_solve.add_argument("--timeout", type=int, default=180, help="Solve timeout in seconds")
    p_solve.set_defaults(func=cmd_solve)

    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    try:
        return args.func(args)
    except CliError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())