Metadata-Version: 2.5
Name: bypass-vuotlink-sdk
Version: 0.5.44
Summary: Python SDK for the bypass-vuotlink API
Requires-Python: >=3.10
Requires-Dist: httpx>=0.24
Requires-Dist: websockets>=13
Description-Content-Type: text/markdown

# bypass-vuotlink-sdk

Python SDK for the [bypass-vuotlink](https://github.com/htilssu/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

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

### Async

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

## Configuration

```python
from bypass_vuotlink_sdk import BypassVuotLink, ClientConfig

# Pass parameters directly
client = BypassVuotLink(
    base_url="http://localhost:8000",
    timeout=90.0,
    api_key="secret",
)

# Or via a config object
config = ClientConfig(base_url="http://localhost:8000", api_key="secret")
client = BypassVuotLink.from_config(config)

# Or from environment variables
# BYPASS_BASE_URL, BYPASS_TIMEOUT, BYPASS_API_KEY
client = BypassVuotLink.from_env()
```

## Result

```python
result.requested_url  # the URL you passed in
result.final_url      # resolved destination
result.status_code    # HTTP status at final_url
result.title          # page title
result.workflow       # workflow used (e.g. "vuotnhanh")
result.hops_used      # how many workflow hops were actually followed
result.hops_exceeded  # True if max_hops was hit before a terminal workflow was reached
result.free_interrupted  # True when free mode stopped before an unsupported workflow
result.free_interrupted_workflow  # workflow name that stopped free mode, or None
result.solved_workflows  # successful hops in execution order

for hop in result.solved_workflows:
    print(hop.workflow, hop.resolved_url)
    print(hop.price, hop.price_auto_solve_captcha)
    print(hop.is_support_free)
```

`solved_workflows` is part of the final `/api/v1/browser` result only. It does
not expose the CAPTCHA token. Each item records the workflow output URL, free-usage
support, and the current base/automatic-CAPTCHA prices loaded from workflow metadata.

## Supported workflows

```python
workflows = client.list_workflows()

for workflow in workflows.workflows:
    print(workflow.name)
    print(workflow.link_formats)
    print(workflow.maintenance, workflow.maintenance_message)
    print(workflow.code_ttl_seconds)
    print(workflow.max_codes_per_link)
    print(workflow.has_captcha)
    print(workflow.is_support_free)
    print(workflow.price)
    print(workflow.price_auto_solve_captcha)
```

Kiểm tra một URL thuộc workflow nào mà không chạy workflow:

```python
workflow = client.check_workflow("https://layma.net/example")
print(
    workflow.name,
    workflow.price,
    workflow.has_captcha,
    workflow.is_support_free,
)
```

Kiểm tra trực tiếp theo domain hoặc hostname:

```python
workflow = client.check_workflow_domain("layma.net")
print(workflow.name, workflow.is_support_free, workflow.price)
```

`resolve()` raises `WorkflowMaintenanceError` when the matched workflow is
temporarily disabled. The exception exposes `workflow` and
`maintenance_message`.

Example item:

```python
workflow.name          # "funlink"
workflow.link_formats  # ["https://funlink.io/..."]
workflow.code_ttl_seconds  # 600, or None when using an older server
workflow.max_codes_per_link  # per-workflow pool target, or None on older servers
workflow.has_captcha         # True only for workflows exposing the captcha flow
workflow.is_support_free     # True when the workflow allows free usage
workflow.price               # per-workflow price, or None on older servers
workflow.price_auto_solve_captcha  # auto CAPTCHA surcharge, or None on older servers
```

Layma mặc định để người dùng giải thủ công qua event `captcha_pending` khi
không truyền option. Truyền `True` để chọn auto solve một cách tường minh:

```python
manual = client.resolve(url, auto_solve_captcha=False)
automatic = client.resolve(url, auto_solve_captcha=True)
```

Free mode kiểm tra từng workflow trước khi chạy. Nếu chuỗi hop gặp workflow
không hỗ trợ free, SDK trả về URL tại hop đó và thông tin workflow bị chặn:

```python
result = client.resolve(url, is_free=True)
if result.free_interrupted:
    print(result.free_interrupted_workflow, result.final_url)
```

The SDK ignores unknown response fields, supplies defaults for fields missing
from older servers, and filters items marked `is_dead=true`. This keeps patch
and additive API contract changes backward compatible.

## Chained links (max_hops)

Some links resolve through more than one bypass workflow in a row (e.g. a
shortlink resolving to a funlink link, which itself resolves to a toplinks
link). By default `resolve()` follows up to 10 such hops before returning.
Pass `max_hops` to cap that:

```python
# Solve only the first link and return immediately, without following any
# further chained hop.
result = client.resolve(url, max_hops=1)
```

`result.hops_used` reports how many hops were actually followed. If the chain
doesn't reach a terminal workflow within `max_hops`, `result.hops_exceeded` is
`True` (`hops_used == max_hops`) and `result.final_url` is just wherever the
last hop left off, not necessarily the fully-resolved destination.

## Fetch a code without submitting it

```python
result = client.fetch_code(
    "https://funlink.io/some/destination",
    "funlink",  # or "toplinks", "ontops", "ontops-dr", "gtraffic", "gtraffic-dr", "layma", "link4m"
    source="live",  # or "background"
)
result.code       # confirmation code
result.dest_url   # the destination it was fetched from
```

## Code pool

```python
# Pop a prefetched code for a cached keyword_text/image_url pair.
result = client.consume_code("some keyword", image_url="https://...")
result.code
result.resolved_url
```

## Not-found cache maintenance

When a destination cannot be auto-resolved, a background scan broadcasts each
unclaimed `search_task` once per minute over `/api/v1/events/ws`. Tasks include
the keyword, extracted URL, search query, and a ready-to-open Google search URL.

Only cache entries used within the rolling previous 24 hours are eligible for
socket tasks. Older entries are not emitted as `search_task` and cannot create
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]"
}
```

```python
claim = await client.claim_search_task(keyword_id, claimant_id="worker-1")
if not claim.claimed:
    return  # Another worker already owns this mission.

result = await client.submit_search_result(
    keyword_id,
    "https://found-it.example/page",
    claimant_id=claim.claimant_id,
)
if result.verified:
    print("Accepted:", result.resolved_url)
else:
    print("Rejected:", result.reason)
```

A successful claim suppresses repeat broadcasts for 10 minutes. If no verified
result is submitted during that window, the server clears the claim and the
next one-minute scan broadcasts the mission again. Claim results are also
broadcast as `search_task_claim_result` so other connected workers can discard
the same mission immediately.

Claim and result submission both use `/api/v1/events/ws`; there is no REST API
for search missions. The submitter receives `search_task_submit_result` for
both accepted and rejected URLs. Result submission must reuse the
`claim.claimant_id` returned by the successful claim. After a successful match, the server also
broadcasts `search_result_verified` with
the same `keywordId`, the accepted `url`, `brand`, and `status="success"`.
`verify_not_found_task()` remains available as a backward-compatible alias.

```python
# Re-open every not-found entry for another search attempt.
result = client.reset_not_found()
result.reset_count
```

## Watch realtime events & handle captcha notifications

Layma background prefetch always dispatches CAPTCHA solving as a
`captcha_task` mission on `/api/v1/events/ws` instead of using an automatic
solver. Its events include `workflow="layma"` and `source="prefetch"`. Live
browser CAPTCHA checkpoints remain `captcha_pending` with `source="live"`.
Both paths emit `captcha_solved` after the AppCaptcha token is available.

```python
# Connect to /api/v1/events/ws and stream generic realtime events.
# Treat prefetch "captcha_task" as a mission. Live "captcha_pending" remains a
# request-scoped checkpoint. When "captcha_solved" arrives, delete the matching
# notification messages.
import asyncio
from collections import defaultdict
from bypass_vuotlink_sdk import AsyncBypassVuotLink

# Maps app_token -> list of (admin_chat_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", ""))
            source = str(raw_event.get("source", ""))
            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}")
```

## Error handling

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

try:
    result = client.resolve(url)
except UnsupportedUrlError:
    ...  # no workflow supports this URL type (HTTP 422)
except UnsafeUrlError:
    ...  # URL is private or unsafe (HTTP 400)
except BrowserExecutionError:
    ...  # browser workflow failed (HTTP 502)
except CodeExhaustedError:
    ...  # provider has no confirmation code available
except BypassVuotlinkError:
    ...  # catch-all
```
