# bypass-vuotlink-sdk

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

## 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,
        timeout: float | None = 300.0, on_event=None,
    ) -> ResolveResult: ...
    def list_workflows(self) -> WorkflowsResult: ...
    def fetch_code(
        self, url: str, brand: Literal["funlink", "toplinks"], *,
        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:

```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,
        timeout: float | None = 300.0, on_event=None,
    ) -> ResolveResult: ...
    async def list_workflows(self) -> WorkflowsResult: ...
    async def fetch_code(
        self, url: str, brand: Literal["funlink", "toplinks"], *,
        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
```

`resolve()`'s `max_hops` caps how many chained-workflow hops to follow (e.g. a
shortlink resolving to a funlink link resolving to a toplinks link) before
returning whatever URL has been reached so far. `1` solves the first link once
and returns immediately without following any further hop. `None`/omitted
keeps the server default of following up to 10 hops. `hops_used` reports how
many hops were actually followed; when the cap is hit, `hops_exceeded` is
`True` (`hops_used == max_hops`) and `final_url` is just wherever the last hop
left off, not necessarily the fully-resolved destination.

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

Usage — manual-search fallback for keywords the auto-resolve flow couldn't
match to a destination link (marked `not_found` in the cache):

```python
task = client.claim_not_found_task(lease_seconds=900)  # raises ApiError(404) if none claimable
# ... search the web manually using task.keyword_text / task.image_url / task.search_query ...
result = client.verify_not_found_task(task.keyword_id, "https://found-it.example/page")
if result.verified:
    print(result.resolved_url)
```

### 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

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

Usage:

```python
result = client.list_workflows()
for workflow in result.workflows:
    print(workflow.name, workflow.link_formats)
```

## 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
└── ApiError                 # any other unexpected HTTP status
      .status_code: int
      .detail: str
```

```python
from bypass_vuotlink_sdk import (
    BypassVuotLink,
    UnsupportedUrlError,
    UnsafeUrlError,
    BrowserExecutionError,
    BypassVuotlinkError,
)

try:
    result = client.resolve(url)
except UnsupportedUrlError:
    # no workflow supports this URL type (e.g. https://fb.com)
    ...
except UnsafeUrlError:
    # URL points to a private/reserved address
    ...
except BrowserExecutionError:
    # the browser failed to process the page
    ...
except BypassVuotlinkError:
    # catch-all for any other SDK error
    ...
```

## Environment variable config

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

```python
client = BypassVuotLink.from_env()
```

## API reference (server)

The SDK wraps these endpoints:

```
POST /api/v1/browser
Content-Type: application/json

{ "url": "https://vuotnhanh.com/abc123", "max_hops": null }
```

`max_hops` is optional (`null`/omitted or `-1` = follow up to 10 chained
hops, the default; `1` = solve the first link once and return immediately).
Response is a stream of newline-delimited JSON events, the last of which is
always `{"event": "result", ...}` (success) or `{"event": "error", ...}`
(failure) — the SDK's `resolve()` consumes this stream for you.

```json
{
  "event": "result",
  "requested_url": "https://vuotnhanh.com/abc123",
  "final_url": "https://example.com/destination",
  "status_code": 200,
  "title": "Example Domain",
  "workflow": "vuotnhanh",
  "hops_used": 1,
  "hops_exceeded": false
}
```

```
GET /api/v1/browser/workflows
```

Response:

```json
{
  "workflows": [
    {
      "name": "funlink",
      "link_formats": ["https://funlink.io/..."]
    }
  ]
}
```

```
POST /api/v1/browser/fetch-code
Content-Type: application/json

{ "url": "https://funlink.io/destination", "brand": "funlink", "source": "live" }
```

`brand` is `"funlink"` or `"toplinks"`; `source` is `"live"` (default) or
`"background"`. Response:

```json
{ "brand": "funlink", "requested_url": "...", "dest_url": "...", "code": "AB12CD" }
```

```
POST /api/v1/codes/consume
Content-Type: application/json

{ "keyword_text": "some keyword", "image_url": "https://..." }
```

Response: `{"code": "...", "resolved_url": "...", "brand": "..."}`. HTTP 404
if there's no cache entry or no prefetched code available.

```
POST /api/v1/codes/not-found/claim?lease_seconds=900
```

Leases one funlink_cache entry with no resolved_url yet for manual search.
HTTP 404 if none are claimable. Response:

```json
{
  "keyword_id": "...", "keyword_text": "...", "image_url": "...",
  "search_query": "...", "brand": "funlink",
  "saved_at": "2026-01-01T00:00:00+00:00",
  "claim_expires_at": "2026-01-01T00:15:00+00:00"
}
```

```
POST /api/v1/codes/not-found/verify
Content-Type: application/json

{ "keyword_id": "...", "url": "https://found-it.example/page" }
```

Verifies a manually-found candidate URL by matching its hostname against the
claimed entry's OCR'd extracted_url — no browser visit involved. Response:
`{"keyword_id": "...", "verified": true, "resolved_url": "...", "reason": null}`
(or `verified: false` with `reason` set and `resolved_url: null`).

```
POST /api/v1/codes/reset-not-found
```

Re-opens every not-found cache entry for another search attempt. Response:
`{"reset_count": 3}`.

Errors:

| Status | Meaning |
|--------|---------|
| 400    | URL is unsafe or private |
| 404    | Resource not found (cache miss, no codes available, no claimable task, unknown keyword_id) |
| 422    | No workflow supports this URL type, or request body failed validation |
| 502    | Browser workflow execution failed |
