Alibaba Cloud Captcha

import "dotenv/config";
import fs from "fs";
import { gotScraping } from "got-scraping";

function parse(text, start, end, isJson = true) {
  const startIndex = text.indexOf(start);
  if (startIndex === -1) return null;

  const contentStart = startIndex + start.length;
  const endIndex = text.indexOf(end, contentStart);
  if (endIndex === -1) return null;

  let extracted = text.substring(contentStart, endIndex).trim();
  extracted = extracted.replace(/\n/g, "").trim();

  let jsonStr = extracted
    .replace(/(['"])?([a-zA-Z0-9_]+)(['"])?:/g, '"$2":')
    .replace(/'/g, '"');

  try {
    return isJson ? JSON.parse(jsonStr) : jsonStr;
  } catch (err) {
    console.error("Failed to parse JSON:", err.message);
    return null;
  }
}

const delay = (ms) => new Promise((res) => setTimeout(res, ms));

class Worker {
  constructor() {
    this.providerVendorSolverUrl = "https://captcha69.com";
    this.API_KEY = process.env.apiKey || "max1_YOUR_API_KEY";

    this.currentDate = new Date();
    this.AliyunGeneratedDynamicJS =
      `https://o.example.com/captcha-frontend/aliyunCaptcha/AliyunCaptcha.js?t=${
        this.currentDate.getFullYear() +
        (this.currentDate.getMonth() + 1) +
        this.currentDate.getDate() +
        this.currentDate.getHours()
      }`;

    this.websiteUrl = "https://example.com/auth";
    this.userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36";
  }

  async requireAlibabaSolverResponse(captchaMetadataParams) {
    let cmReqData = {
      type: "CustomTask",
      class: "alibaba",
      websiteURL: this.websiteUrl,
      websiteKey: "customTask",
      userAgent: this.userAgent,
    };

    if (captchaMetadataParams) {
      cmReqData.metadata = captchaMetadataParams;
    }

    const response = await gotScraping.post(
      `${this.providerVendorSolverUrl}/createTask`,
      {
        body: JSON.stringify({ clientKey: this.API_KEY, task: cmReqData }),
        headers: { "Content-Type": "application/json" },
      }
    );

    let JSON_responseData = JSON.parse(response.body);
    if (JSON_responseData.errorId) throw new Error("JSON.TaskId.error");

    let taskId = JSON_responseData.taskId;
    let responseData;

    while (true) {
      let cmTaskRes = { clientKey: this.API_KEY, taskId: taskId };
      let task_response = await gotScraping.post(
        `${this.providerVendorSolverUrl}/getTaskResult`,
        {
          body: JSON.stringify(cmTaskRes),
          headers: { "Content-Type": "application/json" },
        }
      );

      let JSON_responseDataTaskResponse = JSON.parse(task_response.body);
      if (JSON_responseDataTaskResponse.status !== "processing") {
        responseData = JSON_responseDataTaskResponse;
        break;
      }

      await delay(5000);
    }

    return responseData;
  }
}

new Worker().executor();

import os
import json
import time
import re
import requests
from datetime import datetime

def parse(text, start, end, is_json=True):
    start_index = text.find(start)
    if start_index == -1:
        return None
    content_start = start_index + len(start)
    end_index = text.find(end, content_start)
    if end_index == -1:
        return None
    extracted = text[content_start:end_index].strip()
    extracted = extracted.replace("\n", "").strip()
    json_str = re.sub(r"(['\"])?([a-zA-Z0-9_]+)(['\"])?:", r'"\2":', extracted)
    json_str = json_str.replace("'", '"')
    try:
        return json.loads(json_str) if is_json else json_str
    except Exception as e:
        print("Failed to parse JSON:", str(e))
        return None

class Worker:
    def __init__(self):
        self.provider_vendor_solver_url = "https://captcha69.com"
        self.api_key = os.getenv("API_KEY", "max1_YOUR_API_KEY")

        now = datetime.now()
        self.aliyun_generated_dynamic_js = (
            "https://o.example.com/captcha-frontend/aliyunCaptcha/AliyunCaptcha.js?"
            f"t={now.year}{now.month}{now.day}{now.hour}"
        )

        self.website_url = "https://example.com/auth"
        self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"

    def require_alibaba_solver_response(self, metadata):
        task_payload = {
            "clientKey": self.api_key,
            "task": {
                "type": "CustomTask",
                "class": "alibaba",
                "websiteURL": self.website_url,
                "websiteKey": "customTask",
                "userAgent": self.user_agent,
                "metadata": metadata,
            },
        }

        response = requests.post(
            f"{self.provider_vendor_solver_url}/createTask",
            json=task_payload,
        )
        data = response.json()
        if data.get("errorId"):
            raise Exception("Task creation error")

        task_id = data["taskId"]
        while True:
            result = requests.post(
                f"{self.provider_vendor_solver_url}/getTaskResult",
                json={"clientKey": self.api_key, "taskId": task_id},
            ).json()
            if result["status"] != "processing":
                return result
            time.sleep(5)

if __name__ == "__main__":
    Worker().executor()