Chaturbate Poller

Python library and CLI tool for interacting with the Chaturbate Events API. Monitor and analyze chat activity, tips, room status changes, and other events in real-time with support for structured logging, automated error handling, and optional InfluxDB integration.

Features

• Real-time Event Tracking

  • Monitor chat messages, tips, room status changes, and other events
  • Configurable polling intervals with automatic rate limiting
  • Support for both production and testbed environments

• Unified Configuration

  • Centralized configuration management with validated options
  • Consistent handling across CLI and programmatic interfaces
  • Environment-based configuration with .env file support

• Error Handling

  • Automatic retries with exponential backoff for transient errors
  • Error classification and reporting
  • Connection recovery after network interruptions

• Event Processing

  • Event message formatting with enum-based event types
  • Rich, structured event messages for readability
  • Extensible formatting system for custom event handling

• Logging

  • Structured JSON logs for machine parsing in non-TTY environments
  • Rich console output with formatting
  • Configurable verbosity levels

• Data Persistence & Analytics

  • Optional InfluxDB integration for time-series storage
  • Pre-configured sample queries for common analytics
  • Docker build and runtime setup

Installation

Here are a few ways to install the package:

Using uv (Recommended)

Install with uv:

uv pip install chaturbate-poller

Using pip

Make sure you have Python 3.12+ installed:

# Create and activate virtual environment
python3 -m venv .venv
source .venv/bin/activate

# Install the package
pip install chaturbate-poller

Using uvx (CLI tool isolation)

Run the CLI without installing it in your Python environment:

uvx chaturbate_poller start

Environment Configuration (Optional)

Create a .env file with your credentials:

# Required for API access
CB_USERNAME="your_chaturbate_username"
CB_TOKEN="your_chaturbate_token"

# Optional: InfluxDB settings (if using --database flag)
INFLUXDB_URL="http://influxdb:8086"
INFLUXDB_TOKEN="your_influxdb_token"
INFLUXDB_ORG="chaturbate-poller"
INFLUXDB_BUCKET="my-bucket"
USE_DATABASE="false"  # Set to "true" to enable InfluxDB integration

API Token: You'll need to generate your token at chaturbate.com/statsapi/authtoken/ with "Events API" permission enabled.

Quick Start

# With uv
uv run chaturbate_poller start --username your_username --token your_token

# Using testbed mode (for development/testing)
uv run chaturbate_poller start --testbed --verbose

# With pip installation
python -m chaturbate_poller start --username your_username --token your_token

Usage

CLI Usage

The CLI uses a unified configuration system for validation:

chaturbate_poller start [OPTIONS]

Common Options

Option	Description	Default
--username TEXT	Your Chaturbate username	From .env file
--token TEXT	Your API token	From .env file
--timeout FLOAT	API request timeout in seconds	10.0
--database / --no-database	Enable InfluxDB integration	Disabled
--testbed / --no-testbed	Use testbed environment	Disabled
--verbose / --no-verbose	Enable detailed logging	Disabled
--help	Show help message and exit

For a complete list of the available CLI options:

chaturbate_poller --help

Docker

Run with dependency management and health monitoring:

# Pull the latest image
docker pull ghcr.io/mountaingod2/chaturbate_poller:latest

# Run with environment variables
docker run -d \
  --name chaturbate-poller \
  -e CB_USERNAME="your_chaturbate_username" \
  -e CB_TOKEN="your_chaturbate_token" \
  ghcr.io/mountaingod2/chaturbate_poller:latest --verbose

Docker Compose

For a complete setup including InfluxDB for data persistence:

1. Clone the configuration:

   # Copy the example environment file
   cp .env.example .env
   
   # Edit with your credentials
   nano .env

2. Launch the services:

   docker-compose up -d

3. Pass additional arguments:

   POLLER_ARGS="--verbose --testbed" docker-compose up -d

4. Access InfluxDB at http://localhost:8086

InfluxDB Integration

When enabled with the --database flag, events are stored in InfluxDB for analytics and visualization.

Sample Queries

Here are some useful InfluxDB Flux queries to analyze your Chaturbate data:

// Event count by type (last 24 hours)
from(bucket: "events")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "chaturbate_events")
  |> filter(fn: (r) => r._field == "method")
  |> group(columns: ["_value"])
  |> count()
  |> sort(columns: ["_value"], desc: true)

// Total tips received (last 7 days)
from(bucket: "events")
  |> range(start: -7d)
  |> filter(fn: (r) => r._measurement == "chaturbate_events")
  |> filter(fn: (r) => r.method == "tip")
  |> filter(fn: (r) => r._field == "object.tip.tokens")
  |> sum()

// Top chatters by message count (last 24 hours)
from(bucket: "events")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "chaturbate_events")
  |> filter(fn: (r) => r.method == "chatMessage")
  |> filter(fn: (r) => r._field == "object.user.username")
  |> group(columns: ["_value"])
  |> count()
  |> sort(columns: ["_value"], desc: true)
  |> limit(n: 10)

For more examples, check out the /config/chaturbate_poller/influxdb_queries.flux file.

Programmatic Usage

You can integrate the library into your own Python applications:

Basic Example

import asyncio
from chaturbate_poller import ChaturbateClient

async def main():
    async with ChaturbateClient("your_username", "your_token") as client:
        url = None
        while True:
            response = await client.fetch_events(url)
            for event in response.events:
                # Process each event
                print(f"Event type: {event.method}")
                print(event.model_dump_json(indent=2))

            # Use the next URL for pagination
            url = response.next_url

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

Event Formatting

The library includes event message formatting:

import asyncio
from chaturbate_poller import ChaturbateClient, format_message

async def main():
    async with ChaturbateClient("your_username", "your_token") as client:
        url = None
        while True:
            response = await client.fetch_events(url)
            for event in response.events:
                # Format events using the system
                formatted_message = format_message(event)
                if formatted_message:
                    print(formatted_message)
                else:
                    print(f"Unformatted event: {event.method}")

            url = response.next_url

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

Custom Event Handlers

import asyncio
from chaturbate_poller import ChaturbateClient, format_message
from chaturbate_poller.models.event import Event

async def handle_tip(event: Event) -> None:
    """Process tip events."""
    if event.object.user and event.object.tip:
        formatted_message = format_message(event)
        print(formatted_message)

        # Custom logic for large tips
        amount = event.object.tip.tokens
        if amount >= 100:
            await send_special_thanks(event.object.user.username)

async def handle_chat(event: Event) -> None:
    """Process chat messages."""
    formatted_message = format_message(event)
    if formatted_message:
        print(formatted_message)

async def send_special_thanks(username: str) -> None:
    """Send special thanks for large tips."""
    print(f"Special thanks to {username} for the generous tip!")

async def main():
    async with ChaturbateClient("your_username", "your_token") as client:
        url = None
        while True:
            response = await client.fetch_events(url)
            for event in response.events:
                if event.method == "tip":
                    await handle_tip(event)
                elif event.method == "chatMessage":
                    await handle_chat(event)
                else:
                    formatted_message = format_message(event)
                    if formatted_message:
                        print(formatted_message)
            url = response.next_url

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

With InfluxDB Integration

import asyncio
from chaturbate_poller import ChaturbateClient, format_message
from chaturbate_poller.database.influxdb_handler import InfluxDBHandler

async def main():
    influx_handler = InfluxDBHandler()

    async with ChaturbateClient("your_username", "your_token") as client:
        url = None
        while True:
            response = await client.fetch_events(url)
            for event in response.events:
                formatted_message = format_message(event)
                if formatted_message:
                    print(formatted_message)

                # Store in InfluxDB if configured
                if influx_handler.url:
                    influx_handler.write_event(
                        measurement="chaturbate_events",
                        data=event.model_dump()
                    )
            url = response.next_url

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

Development

Setup Development Environment

1. Clone the repository:

   git clone https://github.com/MountainGod2/chaturbate_poller.git
   cd chaturbate_poller

2. Install dependencies:

   # Using uv (recommended)
   uv sync --all-extras
   
   # Or using pip
   pip install -e ".[dev,docs]"

3. Set up pre-commit hooks:

   uv run pre-commit install

Running Tests

# Run all tests with coverage
uv run pytest --cov-report=html --cov-report=term-missing --cov-report=xml

Documentation

Building Docs Locally

# Install documentation dependencies
uv sync --group=docs

# Build HTML documentation
uv run sphinx-build -b html docs docs/_build/html

Then open docs/_build/html/index.html in your browser.

Online Documentation

Visit the documentation for Jupyter notebook example and API reference.

Changelog

View the complete CHANGELOG.md for version history and updates.

Contributing

Contributions are welcome! Here's how to get started:

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

For more details, please read the Contributing Guidelines.

License

This project is licensed under the MIT License - see the LICENSE file for details.