Metadata-Version: 2.4
Name: kryten-py
Version: 0.2.2
Summary: Python library for building CyTube microservices via Kryten bridge and NATS
License: MIT
License-File: LICENSE
Keywords: cytube,nats,microservices,bot,streaming
Author: Development Team
Requires-Python: >=3.11,<4.0
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Communications :: Chat
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Provides-Extra: all
Provides-Extra: dotenv
Provides-Extra: yaml
Requires-Dist: nats-py (>=2.9.0,<3.0.0)
Requires-Dist: pydantic (>=2.0.0,<3.0.0)
Requires-Dist: pydantic-settings (>=2.0.0,<3.0.0)
Requires-Dist: python-dotenv (>=1.0.0,<2.0.0) ; extra == "dotenv" or extra == "all"
Requires-Dist: pyyaml (>=6.0,<7.0) ; extra == "yaml" or extra == "all"
Description-Content-Type: text/markdown

# kryten-py

Python library for building CyTube microservices via Kryten bridge and NATS.

[![Python Version](https://img.shields.io/badge/python-3.11%2B-blue.svg)](https://www.python.org/downloads/)
[![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE)

## Overview

**kryten-py** provides a high-level, type-safe API for interacting with CyTube servers through the Kryten bridge and NATS message bus. It enables rapid development of microservices like:

- **Chat bots** - Automated chat responses and commands
- **DJ bots** - Playlist management and automation
- **Moderation tools** - User management and content filtering
- **Analytics systems** - Event tracking and statistics
- **Integration services** - Connect CyTube to external platforms

## Features

- ✅ **Async-first API** - Built on asyncio for high performance
- ✅ **Type safety** - Full typing with Pydantic models
- ✅ **Decorator-based handlers** - Intuitive event subscription
- ✅ **Automatic reconnection** - Resilient to network failures
- ✅ **18+ command methods** - Complete CyTube control
- ✅ **Health monitoring** - Built-in metrics and status
- ✅ **Easy testing** - Mock client for unit tests
- ✅ **Comprehensive docs** - Examples and API reference

## Installation

### Basic Installation

```bash
pip install kryten-py
```

### With Optional Dependencies

```bash
# For YAML configuration support
pip install kryten-py[yaml]

# For environment variable loading
pip install kryten-py[dotenv]

# Install all extras
pip install kryten-py[all]
```

### Development Installation

```bash
# Clone repository
git clone https://github.com/yourusername/kryten-py.git
cd kryten-py

# Install with Poetry
poetry install

# Or with pip in editable mode
pip install -e ".[all]"
```

## Quick Start

### Simple Echo Bot

```python
import asyncio
from kryten import KrytenClient, ChatMessageEvent


async def main():
    """Simple echo bot that repeats user messages."""
    
    config = {
        "nats": {
            "servers": ["nats://localhost:4222"],
            "user": "kryten",
            "password": "secret"
        },
        "channels": [
            {"domain": "cytu.be", "channel": "lounge"}
        ]
    }
    
    async with KrytenClient(config) as client:
        @client.on("chatmsg")
        async def on_chat(event: ChatMessageEvent):
            """Echo user messages."""
            if event.username != "MyBot":  # Don't echo ourselves
                await client.send_chat(
                    event.channel,
                    f"{event.username} said: {event.message}"
                )
        
        print("Echo bot started!")
        await client.run()  # Run until Ctrl+C


if __name__ == "__main__":
    asyncio.run(main())
```

### Configuration from File

**config.json:**
```json
{
  "nats": {
    "servers": ["nats://localhost:4222"],
    "user": "${NATS_USER}",
    "password": "${NATS_PASSWORD}"
  },
  "channels": [
    {"domain": "cytu.be", "channel": "lounge"}
  ]
}
```

**bot.py:**
```python
from kryten import KrytenClient, KrytenConfig

# Load configuration with environment variable substitution
config = KrytenConfig.from_json("config.json")

async with KrytenClient(config) as client:
    # ... your bot logic
    await client.run()
```

## Core Concepts

### Event Subscription

Use decorators to register event handlers:

```python
@client.on("chatmsg")
async def handle_chat(event: ChatMessageEvent):
    """Handle all chat messages."""
    print(f"{event.username}: {event.message}")

@client.on("chatmsg", channel="lounge")
async def handle_lounge_only(event: ChatMessageEvent):
    """Handle chat only from lounge channel."""
    if "!ping" in event.message:
        await client.send_chat(event.channel, "Pong!")
```

### Command Publishing

Send commands to CyTube:

```python
# Chat
await client.send_chat("lounge", "Hello world!")
await client.send_pm("lounge", "alice", "Private message")

# Playlist management
await client.add_media("lounge", "yt", "dQw4w9WgXcQ")
await client.delete_media("lounge", uid=42)
await client.shuffle_playlist("lounge")

# Playback control
await client.pause("lounge")
await client.play("lounge")
await client.seek("lounge", 30.0)

# Moderation
await client.kick_user("lounge", "spammer", reason="Spam")
await client.ban_user("lounge", "troll")
```

### Health Monitoring

```python
# Get client health status
health = client.health()

print(f"Connected: {health.connected}")
print(f"Events received: {health.events_received}")
print(f"Commands sent: {health.commands_sent}")
print(f"Avg latency: {health.avg_event_latency_ms:.2f}ms")

# Check connection
if client.is_connected:
    print("Client is connected to NATS")

# Get channel info
for channel in client.channels:
    print(f"{channel.domain}/{channel.channel}: {channel.events_received} events")
```

## Testing

### Using MockKrytenClient

```python
import pytest
from kryten import MockKrytenClient, ChatMessageEvent


@pytest.mark.asyncio
async def test_bot_responds_to_ping():
    """Test bot responds to !ping command."""
    
    config = {
        "nats": {"servers": ["nats://localhost:4222"]},
        "channels": [{"domain": "test.local", "channel": "test"}]
    }
    
    client = MockKrytenClient(config)
    
    @client.on("chatmsg")
    async def handle_command(event: ChatMessageEvent):
        if event.message == "!ping":
            await client.send_chat(event.channel, "Pong!")
    
    async with client:
        # Simulate user sending !ping
        await client.simulate_event("chatmsg", {
            "username": "alice",
            "message": "!ping",
            "timestamp": "2024-01-15T10:00:00Z",
            "rank": 1
        })
        
        # Verify bot responded
        commands = client.get_published_commands()
        assert len(commands) == 1
        assert commands[0]["data"]["message"] == "Pong!"
```

### Running Tests

```bash
# Run all tests
pytest

# Run with coverage
pytest --cov=kryten --cov-report=html

# Run specific test file
pytest tests/test_mock.py

# Run tests in parallel
pytest -n auto
```

## Examples

See the [examples/](examples/) directory for complete examples:

- `echo_bot.py` - Simple echo bot
- `dj_bot.py` - Automated DJ with playlist management
- `moderator_bot.py` - Chat moderation tool
- `analytics.py` - Event tracking and statistics
- `multi_channel.py` - Bot operating on multiple channels

## Architecture

```
Your Bot/Service
       ↓
   kryten-py (this library)
       ↓
     NATS Message Bus
       ↓
   Kryten Bridge
       ↓
   CyTube Server
```

- **kryten-py** abstracts NATS complexity and provides high-level API
- **NATS** handles pub/sub messaging between services
- **Kryten Bridge** translates between NATS and CyTube Socket.IO
- **CyTube** manages channels, users, and media streaming

## Configuration

### NATS Settings

```python
{
  "nats": {
    "servers": ["nats://localhost:4222"],  # NATS server URLs
    "user": "username",                     # Optional authentication
    "password": "password",
    "token": "auth_token",                  # Alternative to user/pass
    "connect_timeout": 10,                  # Connection timeout (seconds)
    "reconnect_time_wait": 2,               # Reconnect delay (seconds)
    "max_reconnect_attempts": -1,           # -1 = infinite retries
    "ping_interval": 120,                   # Keepalive ping interval
  }
}
```

### Channel Settings

```python
{
  "channels": [
    {"domain": "cytu.be", "channel": "lounge"},
    {"domain": "cytu.be", "channel": "movies"},
    {"domain": "test.cytube.local", "channel": "testing"}
  ]
}
```

### Client Settings

```python
{
  "retry_attempts": 3,           # Command retry attempts
  "retry_delay": 1.0,            # Initial retry delay (seconds)
  "handler_timeout": 30.0,       # Max handler execution time (seconds)
  "max_concurrent_handlers": 1000,  # Max concurrent handlers
  "log_level": "INFO"            # Logging level
}
```

## API Reference

### KrytenClient

Main client class for interacting with CyTube via NATS.

**Methods:**
- `connect()` - Establish NATS connection
- `disconnect()` - Close connection gracefully
- `on(event_name, channel=None, domain=None)` - Register event handler (decorator)
- `run()` - Start event processing loop
- `stop()` - Stop event processing loop
- `health()` - Get health status and metrics
- `send_chat(channel, message, domain=None)` - Send chat message
- `send_pm(channel, username, message, domain=None)` - Send private message
- `add_media(channel, media_type, media_id, position="end", domain=None)` - Add media to playlist
- `delete_media(channel, uid, domain=None)` - Delete media from playlist
- `move_media(channel, uid, position, domain=None)` - Move media in playlist
- `jump_to(channel, uid, domain=None)` - Jump to media in playlist
- `clear_playlist(channel, domain=None)` - Clear entire playlist
- `shuffle_playlist(channel, domain=None)` - Shuffle playlist
- `set_temp(channel, uid, is_temp=True, domain=None)` - Set temporary flag on media
- `pause(channel, domain=None)` - Pause playback
- `play(channel, domain=None)` - Resume playback
- `seek(channel, time_seconds, domain=None)` - Seek to time
- `kick_user(channel, username, reason=None, domain=None)` - Kick user
- `ban_user(channel, username, reason=None, domain=None)` - Ban user
- `voteskip(channel, domain=None)` - Vote to skip media

**Properties:**
- `is_connected` - Check if connected to NATS
- `channels` - Get list of configured channels

### Event Models

- `RawEvent` - Raw CyTube event with metadata
- `ChatMessageEvent` - Chat message event
- `UserJoinEvent` - User joined channel
- `UserLeaveEvent` - User left channel
- `ChangeMediaEvent` - Media changed in playlist
- `PlaylistUpdateEvent` - Playlist modified

### Exceptions

- `KrytenError` - Base exception
- `KrytenConnectionError` - Connection failed or lost
- `KrytenValidationError` - Invalid configuration or data
- `KrytenTimeoutError` - Operation timed out
- `PublishError` - Failed to publish command
- `HandlerError` - Event handler raised exception

## Requirements

- Python 3.11+
- NATS server (local or remote)
- Kryten bridge (deployed and configured)
- CyTube server (accessible from Kryten bridge)

## Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Make your changes with tests
4. Run tests and linting (`pytest && ruff check`)
5. Commit your changes (`git commit -m 'Add amazing feature'`)
6. Push to branch (`git push origin feature/amazing-feature`)
7. Open a Pull Request

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## Support

- **Issues**: [GitHub Issues](https://github.com/yourusername/kryten-py/issues)
- **Discussions**: [GitHub Discussions](https://github.com/yourusername/kryten-py/discussions)
- **Documentation**: [Read the Docs](https://kryten-py.readthedocs.io)

## Acknowledgments

- Built for the [CyTube](https://github.com/calzoneman/sync) platform
- Uses [NATS](https://nats.io/) for messaging
- Powered by [Pydantic](https://docs.pydantic.dev/) for data validation

## Related Projects

- [Kryten Bridge](https://github.com/yourusername/kryten-robot) - CyTube to NATS gateway
- [CyTube](https://github.com/calzoneman/sync) - Synchronized media streaming platform
- [nats-py](https://github.com/nats-io/nats.py) - Python NATS client

