Update serializer.py

This commit is contained in:
Andrei 2024-10-05 21:11:45 +03:00
parent 2d8c91095c
commit df0c8798f5
1 changed files with 43 additions and 55 deletions

View File

@ -1,75 +1,63 @@
from typing import Dict, Literal
from typing import Dict, Literal, Optional
from pydantic import Field, BaseModel, constr
from msgspec import Struct
from python3_anticaptcha.core.enum import ProxyTypeEnm, CaptchaTypeEnm, ResponseStatusEnm
from python3_anticaptcha.core.config import APP_KEY
from .enum import ResponseStatusEnm
from .config import APP_KEY
class MyBaseModel(BaseModel):
class Config:
use_enum_values = True
validate_assignment = True
class MyBaseModel(Struct):
def to_dict(self):
return {f: getattr(self, f) for f in self.__struct_fields__}
"""
HTTP API Serializers
"""
class BaseAPIRequestSer(MyBaseModel):
clientKey: constr(min_length=32, max_length=32)
clientKey: str = None
class BaseAPIResponseSer(MyBaseModel):
errorId: int = Field(None, description="Error identifier.")
errorCode: str = Field(None, description="An error code.")
errorDescription: str = Field(None, description="Short description of the error.")
class CreateTaskRequestSer(BaseAPIRequestSer):
task: Dict = Field(None, description="Task object.")
languagePool: str = Field("en", description="Sets workers' pool language. Only applies to image captchas.")
callbackUrl: str = Field(None, description="Web address where we can send the results of captcha task processing.")
class CreateTaskBaseSer(BaseAPIRequestSer):
task: Dict = {}
softId: Literal[APP_KEY] = APP_KEY
class CreateTaskRequestTaskSer(MyBaseModel):
type: CaptchaTypeEnm = Field(..., description="Captcha task type name")
class ProxyDataOptionsSer(MyBaseModel):
proxyType: ProxyTypeEnm = Field(..., description="Type of proxy")
proxyAddress: str = Field(
...,
description="Proxy IP address ipv4/ipv6. No host names or IP addresses from local networks",
)
proxyPort: int = Field(..., description="Proxy port")
class BaseAPIResponseSer(MyBaseModel):
errorId: int = 0
errorCode: str = None
errorDescription: str = None
class CreateTaskResponseSer(BaseAPIResponseSer):
taskId: int = Field(None, description="Task ID that you should later use in the `getTaskResult` method.")
taskId: int = None
class GetTaskResultRequestSer(BaseAPIRequestSer):
taskId: int = Field(None, description="An identifier obtained in the createTask method.")
class GetTaskResultRequestSer(BaseAPIResponseSer):
clientKey: str = None
taskId: int = None
class CaptchaOptionsSer(MyBaseModel):
sleep_time: int = 10
url_request: Optional[str] = None
url_response: Optional[str] = None
"""
HTTP API Response
"""
class GetTaskResultResponseSer(BaseAPIResponseSer):
status: ResponseStatusEnm = Field(None, description="Captcha solving process status.")
solution: Dict = Field(None, description="Task result data. Different for each type of task.")
cost: float = Field(None, description="Cost of the task in USD.")
ip: str = Field(None, description="IP from which the task was created.")
createTime: int = Field(None, description="UNIX timestamp date of task creation.")
endTime: int = Field(None, description="UNIX timestamp date of task completion.")
solveCount: int = Field(None, description="Number of workers who tried to complete your task.")
taskId: int = Field(None, description="Task ID.")
"""
Captcha tasks serializers
"""
class TurnstileProxylessOptionsSer(CreateTaskRequestTaskSer):
websiteURL: str = Field(..., description="Address of a target web page. Can be located anywhere on the web site.")
websiteKey: str = Field(..., description="Website key")
class TurnstileOptionsSer(TurnstileProxylessOptionsSer, ProxyDataOptionsSer):
pass
status: ResponseStatusEnm = ResponseStatusEnm.processing.value
solution: dict = {}
cost: float = 0.0
ip: str = None
endTime: int = None
createTime: int = None
solveCount: int = 0
taskId: int = None