coordinate/toolkit/vs_cloud_api.py

# toolkit/vs_cloud_api.py
import requests
import json
import time
import urllib.parse
from typing import Dict, Any, Optional
from vs_types import NotFoundError, PermissionDeniedError, RateLimiteddError, SessionExpiredOrInvalidError, BizLogicError 
from vs_log_macros import VSC_ERROR, VSC_INFO, VSC_WARN, VSC_DEBUG

class VSCloudApi:
    """
    @brief VSCloudApi 的 Python 实现
    用于对接云端服务 (打码、邮件、Session存储、任务调度)
    """
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(VSCloudApi, cls).__new__(cls)
            # 初始化默认配置
            cls._instance.base_url = "http://45.137.220.138:8888"
            cls._instance.api_token = "Bearer tok_8cb26cf337cb405784eb346dfafb7f54"
            cls._instance.session = requests.Session()
        return cls._instance

    @staticmethod
    def Instance():
        return VSCloudApi()

    def _get_headers(self, content_type: str = "application/json") -> Dict[str, str]:
        return {
            "Authorization": self.api_token,
            "Content-Type": content_type,
            "Accept": "application/json, text/plain, */*"
        }
        
    def _perform_request(self, method, url, headers=None, data=None, json_data=None, params=None):
        """
        统一 HTTP 请求封装，严格复刻 C++ 逻辑：
        1. 发送 OPTIONS 请求
        2. 发送实际请求
        """
        print(f'[perform request] {method} {url} {data} {json_data} {params}')
        resp = self.session.request(method, url, headers=headers, data=data, json=json_data, params=params)
        VSC_INFO('vs_cloud', resp.text)
        if resp.status_code == 200:
            return resp
        elif resp.status_code == 401:
            raise SessionExpiredOrInvalidError()
        elif resp.status_code == 403:
            raise PermissionDeniedError()
        elif resp.status_code == 429:
            raise RateLimiteddError()
        else:
            raise BizLogicError(message=f"HTTP Error {resp.status_code}: {resp.text[:100]}")

    # =========================================================================
    #  VAS Task Management (新增 API)
    # =========================================================================

    def get_vas_task_pop(self, routing_key: str) -> Optional[Dict]:
        """
        获取任务信息
        API: GET /api/vas/task/pop
        """
        url = f"{self.base_url}/api/vas/task/pop"
        params = {
            "queue_name": routing_key,
        }

        headers = self._get_headers()
        
        resp = self._perform_request('GET', url, params=params, headers=headers)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Get vas task pop biz error: {result.get('message')}")

    def update_vas_task(self, 
                        task_id: int, 
                        update_data: Dict[str, Any]) -> Optional[Dict]:
        """
        更新任务
        API: POST /api/vas/task/update?id=1
        Body: update_data (json)
        """
        url = f"{self.base_url}/api/vas/task/update"
        params = {"id": task_id}
        headers = self._get_headers()
        resp = self._perform_request('POST', url, params=params, json_data=update_data, headers=headers)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Update vas task biz error: {result.get('message')}")
    
    def return_vas_task_to_queue(self, task_id: int):
        """
        重入队列
        API: POST /api/vas/task/return_to_queue?task_id=1
        """
        url = f"{self.base_url}/api/vas/task/return_to_queue"
        params = {"task_id": task_id}
        headers = self._get_headers()
        resp = self._perform_request('POST', url, params=params, headers=headers)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Return vas task to queue biz error: {result.get('message')}")

    def submit_anti_turnstile_task(self, proxy: str, website_url: str) -> Optional[Dict]:
        """
        提交反 Turnstile 任务
        """
        url = f"{self.base_url}/api/tasks"
        headers = self._get_headers()
        
        args = {
            "proxy": proxy,
            "websiteUrl": website_url
        }
        
        payload = {
            "command": "AntiCloudflareTurnstileTask",
            "args": json.dumps(args), 
            "status": 0
        }

        resp = self._perform_request('POST', url, headers=headers, json_data=payload)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Submit anti turnstile task biz error: {result.get('message')}")


    def get_anti_turnstile_result(self, task_id: str) -> Optional[Dict]:
        """获取反 Turnstile 结果"""
        url = f"{self.base_url}/api/tasks/{task_id}"
        headers = self._get_headers()
        resp = self._perform_request('GET', url, headers=headers)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Get anti turnstile result biz error: {result.get('message')}")
 
    
    def submit_anticloudflare_task(self, proxy: str, website_url: str) -> Optional[Dict]:
        """
        提交 AntiCloudflareTask (用于 TLSContact 5s 盾)
        """
        url = f"{self.base_url}/api/tasks" 
        
        args = {
            'proxy': proxy,
            'websiteUrl': website_url
        }
        data = {
            "command": "AntiCloudflareTask", 
            "args": json.dumps(args),
            "status": 0
        }
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, json_data=data)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Submit anticloudflare task biz error: {result.get('message')}")
    
    def get_anticloudflare_result(self, task_id, retry_interval=5, max_retries=20) -> Optional[Dict]:
        """
        获取 AntiCloudflareTask 结果 (带轮询)
        """
        url = f"{self.base_url}/api/tasks/{task_id}"
        headers = self._get_headers()
        
        for attempt in range(1, max_retries + 1):
            try:
                resp = self._perform_request('GET', url, headers=headers)
                result = resp.json()
                if result.get("code") == 0:
                    data = result.get("data", {})
                    # status 2 表示成功
                    if data.get("status") == 2:
                        return data
                    elif data.get("status") == 3:
                        VSC_ERROR("vs_cloud", f"AntiCloudflareTask failed: {data.get('result')}")
                        return None
                    else:
                        time.sleep(retry_interval)
            except Exception as e:
                VSC_WARN(
                    "vs_cloud",
                    "Get anticloudflare result exception, attempt %d/%d",
                    attempt, max_retries
                )

                if attempt >= max_retries:
                    break

                time.sleep(retry_interval)
                continue
        return None

    def create_http_session(
        self, 
        session_id: str, 
        cookies: str, 
        local_storage: str, 
        user_agent: str, 
        proxy: str, 
        page: str
    ) -> Optional[Dict]:
        """创建 http session"""
        url = f"{self.base_url}/api/http-session"
        headers = self._get_headers()
        
        payload = {
            "local_storage": local_storage,
            "cookies": cookies,
            "user_agent": user_agent,
            "proxy": proxy,
            "page": page,
            "session_id": session_id
        }
        resp = self._perform_request('POST', url, headers=headers, json_data=payload)

        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Create http session biz error: {result.get('message')}")

    def fetch_mail_content(
        self,
        email: str,
        sender: str,
        recipient: str,
        subject_keywords: str,
        body_keywords: str,
        sent_date: str,
        expiry: int
    ) -> Optional[str]:
        """
        获取邮件内容
        """
        params = {
            "email": email,
            "sender": sender,
            "recipient": recipient,
            "subjectKeywords": subject_keywords,
            "bodyKeywords": body_keywords,
            "sentDate": sent_date,
            "expiry": str(expiry)
        }
        
        url = f"{self.base_url}/api/email-authorizations/fetch"
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, params=params, data="")
        result = resp.json()
        if result.get('code') == 0:
            data = result.get('data', {})
            return data.get('body', '')
        else:
            raise BizLogicError(message=f"Fetch mail content biz error: {result.get('message')}")

    def fetch_mail_content_from_top(
        self,
        email: str,
        sender: str,
        recipient: str,
        subject_keywords: str,
        body_keywords: str,
        top: int
    ) -> Optional[str]:
        """从顶部获取邮件内容"""
        params = {
            "email": email,
            "sender": sender,
            "recipient": recipient,
            "subjectKeywords": subject_keywords,
            "bodyKeywords": body_keywords,
            "top": str(top)
        }
        
        url = f"{self.base_url}/api/email-authorizations/fetch-top"
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, params=params, data="")
        result = resp.json()
        if result.get('code') == 0:
            data = result.get('data', {})
            return data.get('body', "")
        else:
            raise BizLogicError(message=f"Fetch mail content from top biz error: {result.get('message')}")