# bypass-vuotlink-sdk

Documents SDK version 0.5.0 (see `pyproject.toml`).

Python SDK for the bypass-vuotlink API — a service that resolves shortlink/safelink URLs
(e.g. vuotnhanh.com) to their final destination by running a real browser workflow.

## Install

```bash
pip install bypass-vuotlink-sdk==0.5.0
```

## Quick start

```python
from bypass_vuotlink_sdk import BypassVuotLink

# Sync
with BypassVuotLink(base_url="http://localhost:8000") as client:
    result = client.resolve("https://vuotnhanh.com/abc123")
    print(result.final_url)

# Async
from bypass_vuotlink_sdk import AsyncBypassVuotLink

async with AsyncBypassVuotLink(base_url="http://localhost:8000") as client:
    result = await client.resolve("https://vuotnhanh.com/abc123")
    print(result.final_url)
```

## Classes

### BypassVuotLink (sync)

```python
class BypassVuotLink:
    def __init__(
        self,
        base_url: str,
        *,
        timeout: float = 300.0,      # seconds
        api_key: str | None = None,  # sent as X-API-Key header
        extra_headers: dict[str, str] | None = None,
    ) -> None: ...

    @classmethod
    def from_config(cls, config: ClientConfig) -> BypassVuotLink: ...

    @classmethod
    def from_env(cls) -> BypassVuotLink:
        # reads: BYPASS_BASE_URL, BYPASS_TIMEOUT, BYPASS_API_KEY
        ...

    def resolve(
        self, url: str, *, max_hops: int | None = None,
        auto_solve_captcha: bool | None = None,
        is_free: bool = False,
        timeout: float | None = 300.0, on_event=None,
    ) -> ResolveResult: ...
    def list_workflows(self) -> WorkflowsResult: ...
    def check_workflow(self, url: str) -> WorkflowInfo: ...
    def check_workflow_domain(self, domain: str) -> WorkflowInfo: ...
    def fetch_code(
        self, url: str, brand: Literal["funlink", "toplinks", "ontops", "ontops-dr", "gtraffic", "gtraffic-dr", "layma", "link4m"], *,
        source: Literal["live", "background"] = "live",
    ) -> FetchCodeResult: ...
    def consume_code(self, keyword_text: str, image_url: str | None = None) -> ConsumeCodeResult: ...
    def claim_not_found_task(self, *, lease_seconds: int = 900) -> NotFoundTask: ...
    def verify_not_found_task(self, keyword_id: str, url: str) -> VerifyNotFoundResult: ...
    def reset_not_found(self) -> ResetNotFoundResult: ...
    def close(self) -> None: ...

    # context manager
    def __enter__(self) -> BypassVuotLink: ...
    def __exit__(self, ...) -> None: ...
```

### AsyncBypassVuotLink (async)

Same interface as `BypassVuotLink` but all methods are async, plus support for realtime event streaming:

```python
class AsyncBypassVuotLink:
    def __init__(self, base_url: str, *, timeout: float = 300.0,
                 api_key: str | None = None,
                 extra_headers: dict[str, str] | None = None) -> None: ...

    @classmethod
    def from_config(cls, config: ClientConfig) -> AsyncBypassVuotLink: ...

    @classmethod
    def from_env(cls) -> AsyncBypassVuotLink: ...

    async def resolve(
        self, url: str, *, max_hops: int | None = None,
        auto_solve_captcha: bool | None = None,
        is_free: bool = False,
        timeout: float | None = 300.0, on_event=None,
    ) -> ResolveResult: ...
    async def watch_events(
        self, *, ping_interval: float | None = 20.0, events_ws_token: str | None = None,
    ) -> AsyncIterator[dict[str, Any]]: ...
    async def list_workflows(self) -> WorkflowsResult: ...
    async def check_workflow(self, url: str) -> WorkflowInfo: ...
    async def check_workflow_domain(self, domain: str) -> WorkflowInfo: ...
    async def fetch_code(
        self, url: str, brand: Literal["funlink", "toplinks", "ontops", "ontops-dr", "gtraffic", "gtraffic-dr", "layma", "link4m"], *,
        source: Literal["live", "background"] = "live",
    ) -> FetchCodeResult: ...
    async def consume_code(self, keyword_text: str, image_url: str | None = None) -> ConsumeCodeResult: ...
    async def claim_not_found_task(self, *, lease_seconds: int = 900) -> NotFoundTask: ...
    async def verify_not_found_task(self, keyword_id: str, url: str) -> VerifyNotFoundResult: ...
    async def reset_not_found(self) -> ResetNotFoundResult: ...
    async def close(self) -> None: ...

    # async context manager
    async def __aenter__(self) -> AsyncBypassVuotLink: ...
    async def __aexit__(self, ...) -> None: ...
```

### ClientConfig

```python
from dataclasses import dataclass
from typing import ClassVar

@dataclass
class ClientConfig:
    DEFAULT_TIMEOUT: ClassVar[float] = 300.0

    base_url: str
    timeout: float = 300.0
    api_key: str | None = None
    extra_headers: dict[str, str] = field(default_factory=dict)

    @classmethod
    def from_env(cls) -> ClientConfig:
        # BYPASS_BASE_URL  (required)
        # BYPASS_TIMEOUT   (optional, default 300.0)
        # BYPASS_API_KEY   (optional)
        ...

    def build_headers(self) -> dict[str, str]: ...
```

### ResolveResult

```python
@dataclass(frozen=True)
class ResolveResult:
    requested_url: str   # the URL you passed in
    final_url: str       # the resolved destination URL
    status_code: int | None
    title: str           # page title at final_url
    workflow: str        # which workflow handled it (e.g. "vuotnhanh")
    hops_used: int       # how many workflow hops were actually followed
    hops_exceeded: bool  # True if max_hops was hit before a terminal workflow was reached
    free_interrupted: bool  # True when is_free stopped before a non-free workflow
    free_interrupted_workflow: str | None
    solved_workflows: list[SolvedWorkflowHop]

@dataclass(frozen=True)
class SolvedWorkflowHop:
    workflow: str
    resolved_url: str
    price: int
    price_auto_solve_captcha: int
    is_support_free: bool
```

### Stream Events Dataclasses

```python
@dataclass(frozen=True)
class StartedEvent:
    requested_url: str
    workflow: str

@dataclass(frozen=True)
class CaptchaPendingEvent:
    app_token: str
    captcha_site: str

@dataclass(frozen=True)
class CaptchaSolvedEvent:
    app_token: str
    status: str = "success"

@dataclass(frozen=True)
class ErrorEvent:
    error_type: str
    detail: str
    workflow: str | None = None
    maintenance: bool = False
    maintenance_message: str | None = None
```

### FetchCodeResult

```python
@dataclass(frozen=True)
class FetchCodeResult:
    brand: str
    requested_url: str
    dest_url: str
    code: str            # confirmation code fetched, not submitted anywhere
```

### ConsumeCodeResult

```python
@dataclass(frozen=True)
class ConsumeCodeResult:
    code: str
    resolved_url: str
    brand: str
```

### NotFoundTask / VerifyNotFoundResult / ResetNotFoundResult

```python
@dataclass(frozen=True)
class NotFoundTask:
    keyword_id: str
    keyword_text: str
    image_url: str | None
    search_query: str | None
    brand: str | None
    saved_at: datetime
    claim_expires_at: datetime   # lease expiry — task becomes claimable again if unreported

@dataclass(frozen=True)
class VerifyNotFoundResult:
    keyword_id: str
    verified: bool
    resolved_url: str | None   # set when verified
    reason: str | None         # set when not verified (e.g. hostname mismatch)

@dataclass(frozen=True)
class ResetNotFoundResult:
    reset_count: int
```

### WorkflowInfo / WorkflowsResult

```python
@dataclass(frozen=True)
class WorkflowInfo:
    name: str                 # workflow name, e.g. "funlink"
    link_formats: list[str]   # supported URL formats for this workflow
    maintenance: bool
    maintenance_message: str | None
    code_ttl_seconds: int | None
    is_dead: bool
    max_codes_per_link: int | None
    has_captcha: bool
    is_support_free: bool
    price: int | None
    price_auto_solve_captcha: int | None

@dataclass(frozen=True)
class WorkflowsResult:
    workflows: list[WorkflowInfo]
```

## Exceptions

```
BypassVuotlinkError          # base — catch this to handle all SDK errors
├── UnsupportedUrlError      # HTTP 422 — no workflow supports this URL type
├── UnsafeUrlError           # HTTP 400 — URL is private/unsafe
├── BrowserExecutionError    # HTTP 502 — browser workflow crashed
├── CodeExhaustedError       # stream error_type CODE_EXHAUSTED
└── ApiError                 # any other unexpected HTTP status
      .status_code: int
      .detail: str
```

## Environment variable config

```bash
export BYPASS_BASE_URL="http://bypass-api:8000"
export BYPASS_TIMEOUT="90"
export BYPASS_API_KEY="secret"
```

## API reference (server)

The SDK wraps these endpoints:

```
POST /api/v1/browser
GET  /api/v1/browser/workflows
GET  /api/v1/browser/workflows/check-domain?domain=layma.net
POST /api/v1/browser/fetch-code
POST /api/v1/codes/consume
POST /api/v1/codes/not-found/claim?lease_seconds=900
POST /api/v1/codes/not-found/verify
POST /api/v1/codes/reset-not-found
WS   /api/v1/events/ws
```
