Metadata-Version: 2.4
Name: atsq
Version: 1.0.0a4
Summary: Asyncio TeamSpeak ServerQuery client for TeamSpeak 3 and TeamSpeak 6 over SSH
Project-URL: Repository, https://github.com/dev-lukas/atsq
Author: Lukas Roth
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Communications :: Conferencing
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: asyncssh>=2.14
Description-Content-Type: text/markdown

# atsq

Asyncio TeamSpeak ServerQuery client for **TeamSpeak 3** and **TeamSpeak 6**, over SSH.

TeamSpeak 6 removed the classic raw/telnet ServerQuery — SSH query (port 10022) is the
only line-protocol interface left. `atsq` speaks that protocol against both server
generations with one async API, runs on modern Python (3.12–3.14+), and its whole test
suite executes against real `teamspeak:3.13` and `teamspeaksystems/teamspeak6-server`
containers.

## Why

- [`py-ts3`](https://github.com/benediktschmitt/py-ts3) is unmaintained and imports
  `telnetlib` at module load — removed from the standard library in Python 3.13.
- TeamSpeak 6 servers only offer SSH query (and HTTP WebQuery).
- Bots want asyncio-native ergonomics: awaitable commands, `@client.on` event handlers,
  automatic keepalive and reconnect — the `discord.py` feel.

## Install

```
uv add atsq          # or: pip install atsq
```

Requires Python ≥ 3.12. The only runtime dependency is
[asyncssh](https://asyncssh.readthedocs.io/).

## Usage

One-shot administrative session:

```python
import atsq

async with await atsq.connect("ts.example.com", 10022,
                             password="...", server_id=1) as ts:
    for row in await ts.client_list("uid"):
        print(row["clid"], row["client_nickname"])
    cid = await ts.channel_create("Lounge", channel_flag_permanent=1)
```

Long-running bot with events and automatic reconnect:

```python
client = atsq.Client("ts.example.com", 10022, password="...",
                    server_id=1,                    # or server_port=9987
                    nickname="My Bot",              # re-applied on reconnect
                    register_events=atsq.ALL_EVENTS) # or "server", or a list

@client.on("cliententerview")
async def on_join(event: atsq.Event) -> None:
    if event.get("reasonid") == "0" and event.get("client_type") == "0":
        print("joined:", event["client_unique_identifier"])

@client.on("clientleftview")
async def on_leave(event: atsq.Event) -> None:
    print("left:", event.get("clid"))

await client.run_forever()   # reconnects with backoff; keepalive automatic
```

Pull-style event consumption (instead of handlers):

```python
async for event in client.events():
    handle(event)
# or: event = await client.wait_for_event(timeout=240)
```

Anything without a typed wrapper goes through the generic escape-safe `exec()`,
including pipelined bulk commands (many parameter blocks, one round trip):

```python
rows = await ts.exec("servergrouplist")
await ts.exec("clientmove", clid=5, cid=42)
await ts.exec("channeladdperm", cid=60, blocks=[
    {"permsid": "i_channel_needed_join_power", "permvalue": 75},
    {"permsid": "i_channel_needed_subscribe_power", "permvalue": 60},
])
```

Wire constants are available as `StrEnum`s that compare directly against
event/row values:

```python
from atsq import ReasonId, TargetMode, ClientType, LEAVE_REASONS

if event["reasonid"] == ReasonId.CONNECT and event["client_type"] == ClientType.VOICE:
    ...
if event.get("reasonid") in LEAVE_REASONS:
    ...
```

File transfer (icons, avatars, channel files) — same API against TS3 and TS6:

```python
ft = atsq.FileTransfer(client)
icon_id = await ft.upload_icon(png_bytes)            # crc32-named, returns the id
data = await ft.download("/atsq.bin", cid=42)
rows = await ft.file_list(cid=42, path="/")          # [] for empty dirs
await ft.delete_file("/atsq.bin", cid=42)
```

### Errors

```python
try:
    await ts.use(99)
except atsq.QueryError as e:        # error id != 0; str(e) carries the server msg
    print(e.error_id, e.msg)
except atsq.QueryTimeoutError:      # no response in time (connection is closed)
    ...
except atsq.ConnectionClosedError:  # connection gone
    ...
```

`atsq.FloodError` (a `QueryError`, id 524) signals server flood protection — add your
client's IP to the server's `query_ip_allowlist.txt` to be exempt.

### Defaults worth knowing

- **Keepalive**: automatic `whoami` after 240 s idle (servers kick at ~300 s).
  Configure via `keepalive_interval`; `0` disables.
- **Flood protection**: an `error 524` is retried automatically after the wait
  the server asks for (`flood_retries`, default 2; `0` disables). Allowlisted
  IPs (`query_ip_allowlist.txt`) never hit it in the first place.
- **Snapshots** work via plain `exec("serversnapshotcreate")` /
  `exec("serversnapshotdeploy", version=..., data=...)` — note deploy
  deselects the session; call `use` again afterwards.
- **Reconnect** (`run_forever`): exponential backoff 5 s → 300 s; a server message
  containing "banned" waits 300 s. `use`/`servernotifyregister` and `on_ready` re-run
  after every reconnect.
- **Host keys**: verification is off by default (TeamSpeak servers generate ephemeral
  query host keys). Pin one in production: `atsq.connect(..., known_hosts=...)`
  (forwarded to asyncssh).
- **close() sends `quit`**: on TS6 a query client that silently drops the SSH
  connection never produces a `notifyclientleftview`; a clean `quit` does (on both
  generations). See [docs/dialects.md](docs/dialects.md).

## TS3 vs TS6

Probed against real servers — the wire dialects are near-identical, and `atsq`
auto-detects the generation from the greeting (`client.dialect`). All recorded
differences and server-config notes live in [docs/dialects.md](docs/dialects.md).

Enable SSH query on a TS3 server with `TS3SERVER_QUERY_PROTOCOLS=raw,ssh`; on TS6 with
`TSSERVER_QUERY_SSH_ENABLED=1` (password via `TSSERVER_QUERY_ADMIN_PASSWORD`).

## Migrating from py-ts3

| py-ts3 | atsq |
|---|---|
| `TS3ServerConnection("telnet://user:pass@host:10011")` | `await atsq.connect(host, 10022, username=..., password=...)` (SSH) |
| `conn.exec_("clientlist", "uid")` | `await ts.exec("clientlist", "uid")` or `await ts.client_list("uid")` |
| response `resp[0]["cldbid"]` | same shape: `rows[0]["cldbid"]` (`list[dict[str, str]]`) |
| `conn.wait_for_event(timeout=240)` | `await client.wait_for_event(timeout=240)` or `@client.on(...)` |
| `event[0]["reasonid"]` | `event["reasonid"]` (Event is a `Mapping[str, str]`) |
| `conn.send_keepalive()` | automatic (or `await ts.send_keepalive()`) |
| `ts3.query.TS3QueryError` | `atsq.QueryError` (str() still contains the server msg) |
| `ts3.query.TS3TimeoutError` | `atsq.QueryTimeoutError` |
| query builder `.pipe()` | `exec(cmd, blocks=[{...}, {...}], **shared)` |
| `ts3.definitions` constants | `atsq.ReasonId` / `TargetMode` / `ClientType` / `LEAVE_REASONS` |
| `ts3.filetransfer.TS3FileTransfer` | `atsq.FileTransfer` (asyncio, TS3+TS6) |
| manual reconnect loop | `await client.run_forever()` |

## Development

```
uv sync
uv run pytest                        # unit + fake-transport tests (no docker)
./scripts/run-integration-tests.sh   # full suite vs real TS3 + TS6 in docker
uv run ruff check src tests scripts && uv run mypy
```

`scripts/probe_dialect.py` records raw protocol transcripts from a live server —
rerun it when a new TS6 build lands and diff against `tests/unit/fixtures/`.

## License

MIT
