# bypass-vuotlink-sdk

Documents SDK version 0.5.44 (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.44
```

## 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 submit_search_result(self, keyword_id: str, url: str, *, claimant_id: str | None = None, events_ws_token: str | None = None) -> VerifyNotFoundResult: ...
    def claim_search_task(self, keyword_id: str, *, claimant_id: str | None = None) -> SearchTaskClaimResultEvent: ...
    def verify_not_found_task(self, keyword_id: str, url: str, *, claimant_id: str | None = None, events_ws_token: str | None = None) -> 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]]: ...
    # Stream generic events (e.g. captcha_task, captcha_pending, captcha_solved).
    # Treat prefetch "captcha_task" as a mission; live requests use "captcha_pending".
    # When receiving "captcha_solved", cleanup/delete the corresponding alert messages via appToken.
    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 submit_search_result(self, keyword_id: str, url: str, *, claimant_id: str | None = None, events_ws_token: str | None = None) -> VerifyNotFoundResult: ...
    async def claim_search_task(self, keyword_id: str, *, claimant_id: str | None = None) -> SearchTaskClaimResultEvent: ...
    async def verify_not_found_task(self, keyword_id: str, url: str, *, claimant_id: str | None = None, events_ws_token: str | None = None) -> 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
    workflow: str | None = None
    source: str | None = None

@dataclass(frozen=True)
class CaptchaTaskEvent:
    app_token: str
    captcha_site: str
    workflow: str | None = None
    source: str | None = None

@dataclass(frozen=True)
class CaptchaSolvedEvent:
    app_token: str
    status: str = "success"
    workflow: str | None = None
    source: str | None = None

@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
```

### VerifyNotFoundResult / ResetNotFoundResult

```python
@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
```

### SearchTaskEvent / SearchResultVerifiedEvent

```python
@dataclass(frozen=True)
class SearchTaskEvent:
    keyword_id: str
    keyword_text: str
    image_url: str | None
    search_query: str
    search_url: str
    brand: str
    extracted_url: str

@dataclass(frozen=True)
class SearchTaskClaimResultEvent:
    keyword_id: str
    claimant_id: str
    claimed: bool
    claim_expires_at: str | None

@dataclass(frozen=True)
class SearchTaskSubmitResultEvent:
    keyword_id: str
    claimant_id: str
    verified: bool
    resolved_url: str | None
    reason: str | None

@dataclass(frozen=True)
class SearchResultVerifiedEvent:
    keyword_id: str
    url: str
    brand: str
    status: str
```

### 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/reset-not-found
WS   /api/v1/events/ws
```

## Realtime WebSocket Events

The generic event socket also broadcasts destination-search work:

Task sources are limited to cache entries whose `last_used_at` is within the
rolling previous 24 hours. Older cache entries are not dispatched as
`search_task` or Layma prefetch CAPTCHA tasks.

```json
{"event":"search_task","keywordId":"...","keywordText":"hubet","keyword":"hubet","imageUrl":"https://...","searchQuery":"hubet official","searchUrl":"https://www.google.com/search?q=hubet+official","brand":"toplinks","extractedUrl":"https://[hide]example.[hide]"}
```

Claim with `claim_search_task(keyword_id, claimant_id="worker-1")` before doing
the work. Claimed missions are suppressed for 10 minutes; unsolved claims are
cleared and become eligible on the next one-minute scan. Claim outcomes are
broadcast as `search_task_claim_result`.

Submit a candidate over the same WebSocket with
`submit_search_result(keyword_id, url, claimant_id=claim.claimant_id)`.
There is no REST endpoint for search mission results. The submitter receives
`search_task_submit_result` for accepted and rejected URLs and must reuse the
successful claim's `claimant_id`. When the URL
matches and is cached, all listeners also receive:

```json
{"event":"search_result_verified","keywordId":"...","url":"https://matched.example/path","brand":"toplinks","status":"success"}
```

`verify_not_found_task()` is retained as an alias for compatibility.

### Captcha Event & Notification Workflow

Example pattern to stream WebSocket events, dispatch Layma prefetch `captcha_task` missions, handle live `captcha_pending` checkpoints, and delete sent messages once solved (`captcha_solved`):

Layma background prefetch sends `captcha_task` with `workflow="layma"` and
`source="prefetch"`. Live browser checkpoints send `captcha_pending` with
`source="live"`.

```python
import asyncio
from collections import defaultdict
from bypass_vuotlink_sdk import AsyncBypassVuotLink

# Maps app_token -> list of (admin_id, message_id)
pending_captcha_messages: dict[str, list[tuple[int, int]]] = defaultdict(list)

async def get_admins_with_notifications_enabled() -> list[int]:
    # TODO: [Cần confirm nguồn dữ liệu admin] Lấy danh sách ID admin bật thông báo từ DB/bot config
    return [123456789, 987654321]

async def send_admin_captcha_alert(admin_id: int, app_token: str, captcha_site: str) -> int:
    # TODO: [Cần confirm integration với bot Telegram/Discord] Gọi API gửi message tới admin
    print(f"Sending solve request to admin {admin_id} for appToken={app_token}")
    message_id = 1000 + admin_id % 100  # Placeholder message ID
    return message_id

async def delete_admin_message(admin_id: int, message_id: int) -> None:
    # TODO: [Cần confirm API bot delete message] Xóa message yêu cầu solve captcha tương ứng
    print(f"Deleting message {message_id} sent to admin {admin_id}")

async def listen_captcha_events(client: AsyncBypassVuotLink, events_ws_token: str) -> None:
    async for raw_event in client.watch_events(events_ws_token=events_ws_token):
        event_kind = raw_event.get("event")
        app_token = str(raw_event.get("appToken", ""))

        if event_kind in {"captcha_task", "captcha_pending"}:
            captcha_site = str(raw_event.get("captchaSite", ""))
            admin_ids = await get_admins_with_notifications_enabled()
            for admin_id in admin_ids:
                try:
                    msg_id = await send_admin_captcha_alert(admin_id, app_token, captcha_site)
                    pending_captcha_messages[app_token].append((admin_id, msg_id))
                except Exception as exc:
                    print(f"Failed to send alert to admin {admin_id}: {exc}")

        elif event_kind == "captcha_solved":
            # Captcha solved (success or failed/cancelled) -> remove corresponding request messages
            sent_messages = pending_captcha_messages.pop(app_token, [])
            for admin_id, msg_id in sent_messages:
                try:
                    await delete_admin_message(admin_id, msg_id)
                except Exception as exc:
                    # Edge case: message was already manually deleted or missing
                    print(f"Failed to delete message {msg_id} for admin {admin_id}: {exc}")
```

