coordinate/toolkit/vs_cloud_api.py

# toolkit/vs_cloud_api.py
import requests
import json
import time
from urllib.parse import quote
from datetime import datetime
from typing import Dict, Any, List, Optional

import configure
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 = "https://api.text.skin"
            cls._instance.api_token = "Bearer tok_e946329a60ff45ba807f3f41b0e8b7fc"
            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, timeout=5*60):
        """
        统一 HTTP 请求封装
        """
        resp = self.session.request(method, url, headers=headers, data=data, json=json_data, params=params, timeout=timeout)
        VSC_DEBUG('vs_cloud', f'[perform request] {method} {url} {data} {json_data} {params} {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]}")
        
    def get_dynamic_config(self, config_name: str) -> Dict:
        url = f'https://api.text.skin/api/dynamic-configurations/key/{quote(config_name)}'
        headers = self._get_headers()
        resp = self._perform_request('GET', url, headers=headers, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            data = result.get("data") or {}
            return data.get("config_value")
        else:
            raise BizLogicError(message=f"Get dynamic config biz error: {result.get('message')}")

    def get_vas_task_pop(self, routing_key: str) -> Dict:
        if configure.TEST_TASK:
            return configure.TEST_TASK
        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, timeout=10)
        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 get_vas_task(self, task_id: str) -> Dict:
        url = f"{self.base_url}/api/vas/task/detail"
        params = {"task_id": task_id}
        headers = self._get_headers()
        resp = self._perform_request('GET', url, params=params, headers=headers, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Get Task={task_id} error: {result.get('message')}")

    def update_vas_task(self, task_id: str, update_data: Dict[str, Any]) -> Dict:
        """
        更新任务
        """
        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, timeout=10)
        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):
        """
        重入队列
        """
        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, timeout=10)
        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 push_weixin_text(self, text:str):
        """
        推送微信文本消息
        """
        url = f"{self.base_url}/api/wechat/send_no_token"
        payload = {"message": text}
        headers = self._get_headers()
        resp = self._perform_request('POST', url, json_data=payload, headers=headers, timeout=10)
        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 create_task(self, command: str, args: Dict) -> str:
        """
        创建任务
        """
        url = f"{self.base_url}/api/tasks"
        headers = self._get_headers()
        payload = {
            "command": command,
            "args": args,
            "status": 0
        }
        resp = self._perform_request('POST', url, headers=headers, json_data=payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            data = result.get("data", {})
            task_id = data.get("id")
            if not task_id:
                raise BizLogicError(message=f"Task created but no ID returned. Resp: {data}")
            return str(task_id)
        else:
            raise BizLogicError(message=f"Create task failed ({command}): {result.get('message')}")

    def get_task_result(self, task_id: str, timeout: int = 120, interval: int = 3) -> Dict:
        """
        轮询获取任务结果
        """
        url = f"{self.base_url}/api/tasks/{task_id}"
        headers = self._get_headers()
        start_time = time.time()
        while True:
            if time.time() - start_time > timeout:
                raise BizLogicError(message=f"Wait for task result timeout ({timeout}s). TaskID: {task_id}")
            try:
                resp = self._perform_request('GET', url, headers=headers, timeout=10)
                result = resp.json()
                if result.get("code") != 0:
                    raise BizLogicError(message=f"API Error fetching task: {result.get('message')}")
                data = result.get("data", {})
                status = data.get("status")
                if status == 2:
                    return data
                elif status == 3:
                    error_msg = data.get("result", "Unknown error")
                    raise BizLogicError(message=f"Task execution failed: {error_msg}")
            except Exception as e:
                if isinstance(e, BizLogicError):
                    raise e
                VSC_WARN("vs_cloud", f"Polling exception: {str(e)}")
            time.sleep(interval)

    def create_http_session(
        self, 
        session_id: str, 
        cookies: Optional[str] = None, 
        local_storage: Optional[str] = None, 
        user_agent: Optional[str] = None, 
        proxy: Optional[str] = None, 
        page: Optional[str] = None, 
        session_storage: Optional[str] = None
    ) -> Dict:
        """
        创建 http session
        """
        url = f"{self.base_url}/api/http-session"
        headers = self._get_headers()
        payload = {
            "local_storage": local_storage,
            "session_storage": session_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, timeout=10)
        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 slot_refresh_start(
        self,
        routing_key: str,
        country: str = "",
        city: str = "",
        visa_type: str = "Tourist",
        snapshot_source: str = "worker",
    ):
        url = f"{self.base_url}/api/slot_refresh/start"
        payload = {
            "routing_key": routing_key,
            "country": country,
            "city": city,
            "visa_type": visa_type,
            "snapshot_source": snapshot_source,
        }
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, json_data=payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Slot refresh start biz error: {result.get('message')}")

    def slot_refresh_success(self, routing_key: str, snapshot_source: str = "worker"):
        url = f'{self.base_url}/api/slot_refresh/success'
        payload = {
            "routing_key": routing_key,
            "snapshot_source": snapshot_source,
        }
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, json_data=payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Slot refresh success biz error: {result.get('message')}")

    def slot_refresh_fail(self, routing_key: str, error: str, snapshot_source: str = "worker"):
        url = f'{self.base_url}/api/slot_refresh/fail'
        payload = {
            "routing_key": routing_key,
            "snapshot_source": snapshot_source,
            "error": error,
        }
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, json_data=payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Slot refresh fail biz error: {result.get('message')}")
        
    def get_next_account(self, pool_name: str, account_cd: int = 60):
        if configure.TEST_ACCOUNT:
            return configure.TEST_ACCOUNT
        url = f'{self.base_url}/api/account/next'
        payload = {
            "pool_name": pool_name,
            "account_cd": account_cd
        }
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, json_data=payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Get next account biz error: {result.get('message')}")
        
    def get_next_proxy(self, pools: List[str], proxy_cd: int = 60):
        if configure.TEST_PROXY:
            return configure.TEST_PROXY 
        url = f'{self.base_url}/api/proxy/next-ip'
        payload = {
            "pools": pools,
            "proxy_cd": proxy_cd
        }
        headers = self._get_headers()
        resp = self._perform_request('POST', url, headers=headers, json_data=payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Get next account biz error: {result.get('message')}")

    def slot_snapshot_report(self, query_payload: Dict[str, Any] = {}):
        url = f"{self.base_url}/api/slots/report"
        headers = self._get_headers()
        resp = self._perform_request("POST", url, headers=headers, json_data=query_payload, timeout=10)
        result = resp.json()
        if result.get("code") == 0:
            return result.get("data", {})
        else:
            raise BizLogicError(message=f"Slot refresh fail 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="", timeout=5*60)
        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="", timeout=5*60)
        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')}")