Metadata-Version: 2.4
Name: piratetok-live-py
Version: 0.3.0
Summary: TikTok Live WebSocket connector — real-time chat, gifts, likes, and viewer events. No authentication required.
Author: Zmole Cristian
License-Expression: 0BSD
Project-URL: Homepage, https://piratetok.rosint.org
Project-URL: Funding, https://piratetok.rosint.org/donate.html
Project-URL: Repository, https://github.com/PirateTok/live-py
Project-URL: Issues, https://github.com/PirateTok/live-py/issues
Keywords: tiktok,tiktok-live,live,stream,chat,gifts,websocket,protobuf,realtime,piratetok
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: betterproto>=2.0.0b7
Requires-Dist: websockets>=15.0
Requires-Dist: curl_cffi>=0.7.0
Requires-Dist: python-socks[asyncio]>=2.5
Provides-Extra: dev
Requires-Dist: mypy; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Dynamic: license-file

<p align="center">
  <img src="https://raw.githubusercontent.com/PirateTok/.github/main/profile/assets/og-banner-v2.png" alt="PirateTok" width="640" />
</p>

# piratetok-live-py

Connect to any TikTok Live stream and receive real-time events in Python. No signing server, no API keys, no authentication required.

```python
import asyncio
from piratetok_live import TikTokLiveClient, EventType

async def main():
    # Create client — no API key, no signing server, just a username
    client = TikTokLiveClient("username_here")

    # Register handlers with decorators — events carry decoded protobuf data
    @client.on(EventType.chat)
    def on_chat(evt):
        nick = evt.data.get("user", {}).get("nickname", "?")
        print(f"[chat] {nick}: {evt.data.get('content')}")

    @client.on(EventType.gift)
    def on_gift(evt):
        nick = evt.data.get("user", {}).get("nickname", "?")
        gift = evt.data.get("gift", {})
        print(f"[gift] {nick} sent {gift.get('name')} x{evt.data.get('repeatCount')} ({gift.get('diamondCount', 0)} diamonds)")

    @client.on(EventType.like)
    def on_like(evt):
        nick = evt.data.get("user", {}).get("nickname", "?")
        print(f"[like] {nick} ({evt.data.get('total')} total)")

    # Runs the whole session — auth, room resolution, WSS, heartbeat, reconnection.
    # Returns after client.disconnect() or once max_retries consecutive attempts failed.
    # Need to do other work meanwhile? asyncio.create_task(client.connect())
    await client.connect()

asyncio.run(main())
```

## Install

```
pip install piratetok-live-py
```

Requires Python >= 3.11.

## Other languages

| Language | Install | Repo |
|:---------|:--------|:-----|
| **Rust** | `cargo add piratetok-live-rs` | [live-rs](https://github.com/PirateTok/live-rs) |
| **Go** | `go get github.com/PirateTok/live-go` | [live-go](https://github.com/PirateTok/live-go) |
| **JavaScript** | `npm install piratetok-live-js` | [live-js](https://github.com/PirateTok/live-js) |
| **C#** | `dotnet add package PirateTok.Live` | [live-cs](https://github.com/PirateTok/live-cs) |
| **Java** | `com.piratetok:live` | [live-java](https://github.com/PirateTok/live-java) |
| **Lua** | `luarocks install piratetok-live-lua` | [live-lua](https://github.com/PirateTok/live-lua) |
| **Elixir** | `{:piratetok_live, "~> 0.1"}` | [live-ex](https://github.com/PirateTok/live-ex) |
| **Dart** | `dart pub add piratetok_live` | [live-dart](https://github.com/PirateTok/live-dart) |
| **C** | `#include "piratetok.h"` | [live-c](https://github.com/PirateTok/live-c) |
| **PowerShell** | `Install-Module PirateTok.Live` | [live-ps1](https://github.com/PirateTok/live-ps1) |
| **Shell** | `bpkg install PirateTok/live-sh` | [live-sh](https://github.com/PirateTok/live-sh) |

## Features

- **Zero signing dependency** — no API keys, no signing server, no external auth
- **64 decoded event types** — hand-written betterproto dataclasses, no codegen
- **Auto-reconnection** — stale detection, exponential backoff, ttwid retry + reuse, rotation on DEVICE_BLOCKED, retry budget resets after a healthy (30 s+) session
- **Proxy support** — `.proxy(url)` builder, HTTP/HTTPS/SOCKS5 for both HTTP and WSS
- **Enriched User data** — badges, gifter level, moderator status, follow info, fan club
- **Sub-routed convenience events** — `follow`, `share`, `join`, `live_ended`
- **Gift helpers** — `is_combo`, `is_streak_over`, `diamond_total` on every gift event
- **Runtime deps** — `betterproto`, `websockets`, `curl_cffi`, `python-socks`

## Configuration

```python
client = (TikTokLiveClient("username_here")
    .cdn_eu()
    .timeout(15)
    .max_retries(10)
    .stale_timeout(90)
    .heartbeat_interval(10)
    .proxy("socks5://127.0.0.1:1080")
    .user_agent("Mozilla/5.0 ...")
    .cookies("sessionid=abc; sid_tt=abc")
    .language("pt")
    .region("BR")
    .compress(False))
```

| Method | Default | Description |
|:-------|:--------|:------------|
| `.cdn_eu()` / `.cdn_us()` / `.cdn(host)` | Global CDN | WebSocket CDN endpoint |
| `.timeout(seconds)` | `10` | HTTP request timeout in seconds |
| `.max_retries(n)` | `5` | Consecutive failed reconnects before giving up (a 30 s+ session resets the count) |
| `.stale_timeout(seconds)` | `60` | Seconds without data before triggering a reconnect |
| `.heartbeat_interval(seconds)` | `10` | WSS heartbeat interval (also sent as `heartbeat_duration`) |
| `.proxy(url)` | None | HTTP/HTTPS/SOCKS5 proxy for all HTTP + WSS traffic |
| `.user_agent(ua)` | Random from pool | Override the user agent (rotated automatically on reconnect by default) |
| `.cookies(cookies)` | None | Append session cookies alongside ttwid in the WSS cookie header |
| `.language(lang)` | System detected | Override the language sent in WSS params (e.g. `"pt"`, `"ro"`) |
| `.region(reg)` | System detected | Override the region sent in WSS params (e.g. `"BR"`, `"RO"`) |
| `.compress(enabled)` | `True` | Disable gzip compression for WSS payloads (trades bandwidth for CPU) |

## Room info (optional, separate call)

```python
from piratetok_live import check_online, fetch_room_info

result = check_online("username_here")
info = fetch_room_info(result.room_id)

# 18+ rooms
info = fetch_room_info(result.room_id, cookies="sessionid=abc; sid_tt=abc")
```

## Viewers

The top-viewers box (usually top 3) rides the WSS feed — no cookies:

```python
from piratetok_live import top_viewers

@client.on(EventType.room_user_seq)
def on_seq(evt):
    for c in top_viewers(evt.data):
        print(c["rank"], c["user"].get("nickname"), c["score"])
```

The full roster (the web viewer panel) is a separate HTTP call. **TikTok requires session
cookies for this one call** — without them it raises `SessionRequiredError`:

```python
from piratetok_live import check_online, fetch_room_audience

room = check_online("username_here")
audience = fetch_room_audience(room.room_id, room.anchor_id, cookies="sessionid=abc; sid_tt=abc")
# audience.total, audience.anonymous, audience.viewers[i].username / nickname / rank / ...
```

Pass `anchor_id=None` to resolve it from room info (one extra request).

## Examples

```bash
python examples/basic_chat.py <username>       # connect + print chat events
python examples/online_check.py <username>     # check if user is live
python examples/stream_info.py <username>      # fetch room metadata + stream URLs
python examples/gift_tracker.py <username>     # track gifts with diamond totals
python examples/gift_streak.py <username>      # track gifts with GiftStreakTracker deltas
python examples/profile_lookup.py [username]   # fetch profile via SIGI scrape with caching
python examples/audience.py <username> <cookies>  # full viewer roster (needs session cookies)
```

## Tests

```bash
pip install -e ".[test]"
pytest -m "not integration"          # offline: replay, ttwid retry, reconnect loop, WSS, proxy, parsing
```

Replay tests read `testdata/` (gitignored, `captures/*.bin` + `manifests/*.json`) or a
[live-testdata](https://github.com/PirateTok/live-testdata) checkout:

```bash
git clone https://github.com/PirateTok/live-testdata ../live-testdata
PIRATETOK_TESTDATA=../live-testdata pytest tests/test_replay.py
```

Missing testdata fails the replay tests — they never pass vacuously.

## License

0BSD
