cottongin/Icecast-metadata-IRC-announcer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Activity

Compare commits

base: cottongin:HEAD
Branches Tags
cottongin:master
cottongin:fool-bemrose
cottongin:v1.2.2
cottongin:v1.2.1
cottongin:v1.2.0
cottongin:v1.1.0
cottongin:v1.0.1
cottongin:v1.0.0
..
compare: cottongin:v1.0.0
Branches Tags
cottongin:fool-bemrose
cottongin:master
cottongin:v1.2.2
cottongin:v1.2.1
cottongin:v1.2.0
cottongin:v1.1.0
cottongin:v1.0.1
cottongin:v1.0.0

No commits in common. "HEAD" and "v1.0.0" have entirely different histories.
HEAD ... v1.0.0

16 changed files with 307 additions and 2834 deletions
Show Stats Expand all files Collapse all files

1
.gitignore vendored
View File

@ -30,7 +30,6 @@ env/
.config/ .config/
config*.yaml config*.yaml
!config.yaml.example !config.yaml.example
.archive/
# Logs # Logs
*.log *.log

113
CHANGELOG.md
View File

@ -1,113 +0,0 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
## [1.2.2] - 2025-02-25
### Added
- Improved stream reconnection verification with log entries
- Enhanced metadata endpoint discovery with multiple URL patterns
- Expanded error handling and logging for metadata fetching
### Changed
- Refactored stream monitoring for better stability
- Updated metadata parsing to handle varied source formats
- General project cleanup and housekeeping
### Fixed
- Stream reconnection edge cases and timeout handling
- JSON metadata parsing reliability
- Various validations
## [1.2.1] - 2025-02-25
### Added
- Improved stream reconnection logic with automatic retry
- Enhanced metadata fetching with better error handling
- Expanded logging for better troubleshooting
### Fixed
- Issue with metadata parsing for certain stream formats
- Stream monitoring stability improvements
- Connection timeout handling
## [1.2.0] - 2025-02-25
### Added
- New admin commands:
- !quiet - Disable song announcements while continuing to monitor the stream
- !unquiet - Re-enable song announcements
- Terminal commands for controlling announcement behavior:
- `quiet` - Disable song announcements via terminal
- `unquiet` - Enable song announcements via terminal
- Configuration option `quiet_on_start` to control whether the bot starts with announcements enabled or disabled
- Background task to monitor for quiet/unquiet requests from terminal commands
- Updated documentation in README.md with new commands
- Updated usage information in install.sh
### Changed
- Enhanced the RestartManager to handle quiet and unquiet commands
- Improved the BotManager class to support the new terminal commands
- Updated configuration example to include the new quiet_on_start option
## [1.0.1] - 2025-02-24
### Added
- Configurable logging levels via config.yaml
- Smart URL resolution for metadata fetching
- Connection status verification and reporting
- Admin command system with permissions
- Automatic restart functionality via bot.sh
- Enhanced command handling with prefix configuration
- Private message command control
- New admin commands:
- !start - Start stream monitoring
- !stop - Stop stream monitoring
- !reconnect - Reconnect to stream with status feedback
- !restart - Restart the bot (requires bot.sh)
- !quit - Shutdown the bot
- !help command showing available commands based on user permissions
- ERROR.log file for critical issues
- Detailed debug logging for troubleshooting
### Changed
- Improved metadata fetching with multiple URL patterns
- Enhanced error handling and reporting
- Better stream connection status feedback
- More informative health check messages
- Cleaner logging output with configurable verbosity
- Updated documentation with new features and configuration options
### Fixed
- Metadata fetching issues with different URL patterns
- Command handling in channels vs private messages
- Stream reconnection verification
- Error reporting and logging clarity
- Configuration file structure and validation
### Other
- Added version information
## [1.0.0] - 2025-02-23
### Added
- Initial release
- Basic Icecast stream monitoring
- IRC channel announcements
- Simple command system (!np)
- Basic error handling and reconnection
- Multi-bot support
- Configuration via YAML files
[Unreleased]: https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/compare/v1.2.2...HEAD
[1.2.2]: https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/compare/v1.2.1...v1.2.2
[1.2.1]: https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/compare/v1.2.0...v1.2.1
[1.2.0]: https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/compare/v1.0.1...v1.2.0
[1.0.1]: https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/compare/v1.0.0...v1.0.1
[1.0.0]: https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/releases/tag/v1.0.0

137
DOCS.md
View File

@ -1,137 +0,0 @@
# Icecast IRC Bot Documentation v1.1.0
This document is automatically generated from the codebase.
## Overview
An IRC bot that monitors an Icecast stream and announces track changes.
## Configuration
See `config.yaml.example` for a full example configuration file.
No command documentation available (help template not found)
## Methods
### admin_required
Decorator to mark a command as requiring admin privileges.
**Arguments:**
- `f`: The command handler function to wrap.
- `Returns`:
### announce_song
Announce a song in the IRC channel.
**Arguments:**
- `song`: The song title to announce.
- `Only announces if`:
### fetch_json_metadata
Fetch metadata from the Icecast JSON status endpoint.
**Returns:**
str: The current song title, or an error message if fetching failed.
### format_help_section
Format a help section according to the template.
**Arguments:**
- `section_config`: Configuration dictionary for the section.
- `prefix`: Command prefix to use.
- `Returns`:
- `List[str]`: List of formatted help lines for each command.
### get_version
Get the current version from VERSION file.
### help_command_fallback
Fallback help command implementation using hardcoded format.
**Arguments:**
- `message`: The IRC message object that triggered this command.
### is_admin
Check if a user has admin privileges.
**Arguments:**
- `user`: Full IRC user string (nickname!username@hostname) or User object.
- `Returns`:
- `bool`: True if user has admin privileges, False otherwise.
### load_config
Load and validate the bot configuration from a YAML file.
**Arguments:**
- `config_path`: Path to the YAML configuration file. If None, uses default path.
- `Returns`:
- `dict`: The loaded and validated configuration dictionary with default values applied.
### monitor_metadata
Monitor the Icecast stream for metadata changes.
### restart_monitoring
Restart the metadata monitoring task and verify the reconnection.
**Returns:**
bool: True if reconnection was successful and verified, False otherwise.
### setup_handlers
Set up all IRC event handlers and command patterns.
### should_announce_song
Check if a song should be announced based on configured ignore patterns.
**Arguments:**
- `song`: The song title to check.
- `Returns`:
- `bool`: True if the song should be announced, False if it matches any ignore patterns.
### start
Start the IRC bot and begin processing events.
### start_monitoring
Start the metadata monitoring task.
### stop_monitoring
Stop the metadata monitoring task.

112
README.md
View File

@ -1,21 +1,17 @@
# Icecast-metadata-IRC-announcer # Icecast-metadata-IRC-announcer
[![Version](https://img.shields.io/badge/dynamic/raw?color=blue&label=version&query=.&url=https%3A%2F%2Fcode.cottongin.xyz%2Fcottongin%2FIcecast-metadata-IRC-announcer%2Fraw%2Fbranch%2Fmaster%2FVERSION)](https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer/releases)
A simple asynchronous Python bot that monitors an Icecast stream and announces track changes to an IRC channel. Supports running multiple instances with different configurations. A simple asynchronous Python bot that monitors an Icecast stream and announces track changes to an IRC channel. Supports running multiple instances with different configurations.
Note: This is a work in progress. It has only been tested on **Python 3.12.6**. Note: This is a work in progress. It has only been tested on **Python 3.12.6**.
## Features ## Features
- Monitors Icecast streams for metadata changes - Monitors Icecast stream metadata and announces track changes
- Announces new tracks in IRC channels - Configurable via YAML files and command line arguments
- Supports multiple IRC networks and channels - Supports running multiple bot instances simultaneously
- Customizable announcement formats - Pattern-based song title filtering
- Command system with admin privileges - Responds to !np commands in IRC channels
- Automatic reconnection on network issues - Automatic reconnection and error recovery
- Multiple bot instances can be managed together
- Systemd service integration
## Dependencies ## Dependencies
@ -37,75 +33,29 @@ Create a YAML config file (default: `config.yaml`):
```yaml ```yaml
irc: irc:
host: "irc.libera.chat" host: "irc.example.net"
port: 6667 port: 6667
nick: "IcecastBot" nick: "MusicBot"
user: "icecastbot" user: "musicbot"
realname: "Icecast IRC Bot" realname: "Music Announcer Bot"
channel: "#yourchannel" channel: "#music"
stream: stream:
url: "https://your.stream.url" # Base URL without /stream or .mp3 url: "https://stream.example.com"
endpoint: "/stream" # The endpoint part (e.g. /stream, /radio.mp3) endpoint: "stream"
health_check_interval: 300 # How often to check health status (in seconds) health_check_interval: 300
announce: announce:
format: "\x02Now playing:\x02 {song}" # Format for song announcements format: "\x02Now playing:\x02 {song}"
ignore_patterns: # Don't announce songs matching these patterns ignore_patterns:
- "Unknown" - "Unknown"
- "Unable to fetch metadata" - "Unable to fetch metadata"
- "Error fetching metadata" # Add more patterns to ignore
commands:
prefix: "!" # Command prefix (e.g. !np, !help)
require_nick_prefix: false # If true, commands must be prefixed with "botname: " or "botname, "
allow_private_commands: false # If true, allows commands in private messages
help: # Help message templates
specific_format: "\x02{prefix}{cmd}\x02: {desc}" # Format for specific command help
list_format: "(\x02{cmd}\x02, {desc})" # Format for commands in list
list_separator: " | " # Separator between commands in list
admin:
users: # List of users who can use admin commands (use "*" for anyone)
- "*"
``` ```
## Usage ## Usage
## Manager Commands ### Single Bot Mode
The bot manager supports the following commands:
```bash
icecast-irc-bot-manager list # List running bots
icecast-irc-bot-manager start --config FILE # Start a bot
icecast-irc-bot-manager stop BOT_ID # Stop a bot
icecast-irc-bot-manager restart BOT_ID # Restart a bot
icecast-irc-bot-manager quiet BOT_ID # Disable song announcements
icecast-irc-bot-manager unquiet BOT_ID # Enable song announcements
```
### Running with Automatic Restart Support
The recommended way to run the bot is using the provided `bot.sh` script, which handles automatic restarts when using the `!restart` command:
```bash
# Make the script executable
chmod +x bot.sh
# Run a single bot
./bot.sh config.yaml
# Run multiple bots
./bot.sh config1.yaml config2.yaml config3.yaml
```
### Manual Running
You can also run the bot directly with Python:
#### Single Bot Mode
Run with config file: Run with config file:
```bash ```bash
@ -125,38 +75,32 @@ Available command line arguments:
- `--irc-channel`: IRC channel to join - `--irc-channel`: IRC channel to join
- `--stream-url`: Icecast base URL - `--stream-url`: Icecast base URL
- `--stream-endpoint`: Stream endpoint - `--stream-endpoint`: Stream endpoint
- `--cmd-prefix`: Command prefix character(s)
#### Multiple Bot Mode ### Multiple Bot Mode
Run multiple instances with different configs: Run multiple instances with different configs:
```bash ```bash
python main.py config1.yaml config2.yaml config3.yaml python main.py config1.yaml config2.yaml config3.yaml
``` ```
## Commands Or use the launcher script:
```bash
python launch.py
```
The bot supports the following commands: ## IRC Commands
- `!np` - Show the currently playing song - `!np`: Shows the currently playing track
- `!help` - Show available commands or help for a specific command
Admin commands (only available to users listed in the `admin.users` config): ## Logging
- `!start` - Start stream monitoring The bot logs important events to stdout with timestamps. Log level is set to INFO by default.
- `!stop` - Stop stream monitoring
- `!quiet` - Disable song announcements but continue monitoring
- `!unquiet` - Enable song announcements
- `!reconnect` - Reconnect to the stream
- `!restart` - Restart the bot
- `!quit` - Shutdown the bot
## Error Handling ## Error Handling
- Automatically reconnects on connection drops - Automatically reconnects on connection drops
- Retries stream monitoring on errors - Retries stream monitoring on errors
- Smart metadata URL resolution - Logs errors for debugging
- Connection status verification and reporting
- Health checks every 5 minutes (configurable) - Health checks every 5 minutes (configurable)
## License ## License

38
TODO.md
View File

@ -1,26 +1,16 @@
## TODO 🚧 ## TODO
- ⭐️ Better, more clear, logging - Better, more clear, logging
- Notification system for stream admin (and bot admin) - Add a help command
- Send notification if unexpected change in metadata - Add commands to control the stream connection from IRC
- Loss of network/stream - Also to start/stop the service entirely
- Live notifications (alt. for PodPing ish) - Add a version command
- Implement `quiet mode` - Add a list of commands to the README
- ⭐️ Add whitelist feature
- Add commands:
- To check the status of the bot
- To check the status of the stream
- To check the status of the IRC connection
- Version command
- Enable joining multiple channels with a single bot instance - Enable joining multiple channels with a single bot instance
- Add a command to force the bot to reconnect to the stream
## Complete! 🎉 - Add a command to check the status of the bot
- Add a command to check the status of the stream
- ✅ Move this to a TODO.md file 😊 - Add a command to check the status of the IRC connection
- 🟢 Added commands: - Add a command to check the status of the bot
- ✅ Add `quiet mode` relevant commands - Add a command to check the status of the stream
- ✅ Add a help command - Move this to a TODO.md file :)
- ✅ To control the stream connection from IRC
- ✅ To start/stop the service entirely
- ✅ To quiet/unquiet the metadata announcing
- ✅ Add a list of commands to the README

1
VERSION
View File

@ -1 +0,0 @@
-testing

39
config.yaml.example
View File

@ -1,34 +1,19 @@
irc: irc:
host: "irc.libera.chat" host: "irc.someircserver.net"
port: 6667 port: 6667 # asif does not support ssl as of 2025-02-23
nick: "IcecastBot" nick: "bot"
user: "icecastbot" user: "bot"
realname: "Icecast IRC Bot" realname: "Botty Bot - https://bot.site"
channel: "#yourchannel" channel: "#channel"
stream: stream:
url: "https://your.stream.url" # Base URL without /stream or .mp3 url: "http://your-icecast-server.com:8000/" # this is the *base* url of the icecast server
endpoint: "/stream" # The endpoint part (e.g. /stream, /radio.mp3) endpoint: "stream" # this is the endpoint of the icecast server (e.g. /stream or .mp3)
health_check_interval: 300 # How often to log health status (in seconds) health_check_interval: 300
announce: announce:
format: "\x02Now playing:\x02 {song}" # Format for song announcements format: "\x02Now playing:\x02 {song}"
ignore_patterns: # Don't announce songs matching these patterns ignore_patterns:
- "Unknown" - "Unknown"
- "Unable to fetch metadata" - "Unable to fetch metadata"
- "Error fetching metadata" # Add more patterns to ignore here
quiet_on_start: false # If true, bot starts in quiet mode (no announcements)
commands:
prefix: "!" # Command prefix (e.g. !np, !help)
require_nick_prefix: false # If true, commands must be prefixed with "botname: " or "botname, "
allow_private_commands: false # If true, allows commands in private messages
help: # Help message templates
specific_format: "\x02{prefix}{cmd}\x02: {desc}" # Format for specific command help
list_format: "(\x02{cmd}\x02, {desc})" # Format for commands in list
list_separator: " | " # Separator between commands in list
admin:
users: # List of users who can use admin commands (use "*" for anyone)
- "*"

99
generate_docs.py
View File

@ -1,99 +0,0 @@
#!/usr/bin/env python3
import inspect
import yaml
from pathlib import Path
from main import IcecastBot
def get_version():
"""Get the current version from VERSION file."""
try:
with open('VERSION') as f:
return f.read().strip()
except FileNotFoundError:
return "Unknown"
def format_docstring(obj):
"""Format a docstring into markdown."""
doc = inspect.getdoc(obj)
if not doc:
return ""
# Split into description and args
parts = doc.split('\n\n')
formatted = [parts[0]] # Description
for part in parts[1:]:
if part.startswith('Args:'):
formatted.append("\n**Arguments:**\n")
# Parse args section
args = part.replace('Args:', '').strip().split('\n')
for arg in args:
if ':' in arg:
name, desc = arg.split(':', 1)
formatted.append(f"- `{name.strip()}`: {desc.strip()}")
elif part.startswith('Returns:'):
formatted.append("\n**Returns:**\n")
formatted.append(part.replace('Returns:', '').strip())
return '\n'.join(formatted)
def generate_command_docs(config_path='config.yaml.example'):
"""Generate command documentation from help templates."""
try:
with open(config_path) as f:
config = yaml.safe_load(f)
except FileNotFoundError:
return "No command documentation available (config file not found)"
help_config = config.get('commands', {}).get('help', {})
if not help_config:
return "No command documentation available (help template not found)"
docs = ["## Commands\n"]
# Regular commands
if 'commands' in help_config.get('sections', {}):
docs.append("### Regular Commands\n")
for cmd, desc in help_config['sections']['commands']['commands'].items():
docs.append(f"- `{cmd}`: {desc}")
docs.append("")
# Admin commands
if 'admin' in help_config.get('sections', {}):
docs.append("### Admin Commands\n")
docs.append("These commands are only available to users listed in the `admin.users` config section.\n")
for cmd, desc in help_config['sections']['admin']['commands'].items():
docs.append(f"- `{cmd}`: {desc}")
docs.append("")
return '\n'.join(docs)
def generate_docs():
"""Generate full documentation in markdown format."""
version = get_version()
docs = [
f"# Icecast IRC Bot Documentation v{version}\n",
"This document is automatically generated from the codebase.\n",
"## Overview\n",
format_docstring(IcecastBot),
"\n## Configuration\n",
"See `config.yaml.example` for a full example configuration file.\n",
generate_command_docs(),
"\n## Methods\n"
]
# Document public methods
for name, method in inspect.getmembers(IcecastBot, predicate=inspect.isfunction):
if not name.startswith('_'): # Only public methods
docs.append(f"### {name}\n")
docs.append(format_docstring(method))
docs.append("\n")
# Write to DOCS.md
with open('DOCS.md', 'w') as f:
f.write('\n'.join(docs))
if __name__ == '__main__':
generate_docs()

654
icecast-irc-bot-manager.py
View File

@ -1,654 +0,0 @@
#!/usr/bin/env python3
import argparse
import asyncio
import json
import os
import signal
import sys
import tempfile
import time
from pathlib import Path
from typing import Dict, List, Optional
# ANSI color codes for terminal output if needed
class Colors:
COLORS = [
'\033[94m', # BLUE
'\033[92m', # GREEN
'\033[95m', # MAGENTA
'\033[93m', # YELLOW
'\033[96m', # CYAN
'\033[91m', # RED
'\033[38;5;208m', # ORANGE
'\033[38;5;165m', # PURPLE
'\033[38;5;39m', # DEEP BLUE
'\033[38;5;82m', # LIME
]
ENDC = '\033[0m'
BOLD = '\033[1m'
# Additional colors for output
GREY = '\033[37m'
WHITE = '\033[97m'
RED = '\033[91m'
ORANGE = '\033[38;5;208m' # Using 256-color code for orange
CYAN = '\033[96m'
MAGENTA = '\033[95m'
class BotManager:
"""Manages multiple Icecast IRC bot instances."""
def __init__(self):
self.bots: Dict[str, asyncio.subprocess.Process] = {}
self.config_dir = Path('.') # Use current directory instead of /etc/icecast-irc-bot
self.socket_dir = Path(tempfile.gettempdir())
self.state_file = Path(tempfile.gettempdir()) / 'icecast-irc-bots.json'
self.monitor_task = None
self.venv_python = os.getenv('VIRTUAL_ENV', '/opt/icecast-irc-bot/venv') + '/bin/python3'
# If venv_python doesn't exist, use the current Python interpreter
if not os.path.exists(self.venv_python):
self.venv_python = sys.executable
def _save_state(self):
"""Save the current state of running bots."""
try:
# Load existing state
existing_state = self._load_state()
# Update with current bots
current_state = {
bot_id: {
'pid': process.pid,
'config': str(self.config_dir / f"{bot_id}.yaml")
}
for bot_id, process in self.bots.items()
}
# Merge states, with current bots taking precedence
merged_state = {**existing_state, **current_state}
# Save merged state
with open(self.state_file, 'w') as f:
json.dump(merged_state, f)
print(f"Saved state with {len(merged_state)} bots")
except Exception as e:
print(f"Error saving state: {e}")
pass
def _load_state(self):
"""Load the state of running bots."""
try:
if self.state_file.exists():
with open(self.state_file, 'r') as f:
state = json.load(f)
return state
return {}
except Exception:
return {}
async def monitor_processes(self):
"""Monitor running bot processes and clean up dead ones."""
while True:
try:
for bot_id in list(self.bots.keys()):
process = self.bots[bot_id]
try:
# Check if process exists in system
os.kill(process.pid, 0)
# Check if process has terminated
if process.returncode is not None:
await self._cleanup_process(process)
del self.bots[bot_id]
except ProcessLookupError:
await self._cleanup_process(process)
del self.bots[bot_id]
except Exception:
pass
except Exception:
pass
await asyncio.sleep(5) # Check every 5 seconds
async def _cleanup_process(self, process: asyncio.subprocess.Process):
"""Clean up a process and its resources.
Args:
process: The process to clean up
"""
try:
print(f"Cleaning up process with PID {process.pid}")
# Wait for the process to finish if it hasn't
if process.returncode is None:
try:
print(f"Terminating process {process.pid}")
process.terminate()
await asyncio.wait_for(process.wait(), timeout=5.0)
except (asyncio.TimeoutError, ProcessLookupError) as e:
print(f"Error terminating process: {e}")
try:
print(f"Killing process {process.pid}")
process.kill()
await process.wait()
except ProcessLookupError as e:
print(f"Process already gone: {e}")
pass
# Drain any remaining output to prevent deadlocks
if process.stdout:
try:
data = await process.stdout.read()
if data:
print(f"Remaining stdout: {data.decode().strip()}")
except (ValueError, IOError) as e:
print(f"Error reading stdout: {e}")
pass
if process.stderr:
try:
data = await process.stderr.read()
if data:
print(f"Remaining stderr: {data.decode().strip()}")
except (ValueError, IOError) as e:
print(f"Error reading stderr: {e}")
pass
except Exception as e:
print(f"Exception during cleanup: {e}")
pass
async def start_bot(self, config_path: Path) -> bool:
"""Start a new bot instance with the given config file.
Args:
config_path: Path to the config file
Returns:
bool: True if bot was started successfully
"""
try:
# Generate a unique ID for this bot based on the config file name
bot_id = config_path.stem
# Check if a bot with this ID is already running
state = self._load_state()
if bot_id in state:
try:
pid = state[bot_id]['pid']
os.kill(pid, 0) # Check if process exists
return False
except ProcessLookupError:
# Process doesn't exist, remove from state
pass
# Start the bot process using venv Python
process = await asyncio.create_subprocess_exec(
self.venv_python, 'main.py',
'--config', str(config_path),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
# Verify process started successfully
try:
os.kill(process.pid, 0)
except ProcessLookupError:
await self._cleanup_process(process)
return False
# Start the monitor task if not already running
if self.monitor_task is None:
self.monitor_task = asyncio.create_task(self.monitor_processes())
# Monitor the process output for a short time to ensure it starts properly
try:
startup_timeout = 5.0 # Give it 5 seconds to start
start_time = time.time()
success = False
while time.time() - start_time < startup_timeout:
# Check if process has exited
if process.returncode is not None:
await self._cleanup_process(process)
return False
# Read any available output during startup
try:
line = await asyncio.wait_for(process.stderr.readline(), timeout=0.1)
if line:
line = line.decode().strip()
print(f"Bot output: {line}")
# Consider the bot started successfully after a short delay
# instead of looking for specific output
success = True
except asyncio.TimeoutError:
# No output available, continue monitoring
continue
except Exception as e:
print(f"Error reading bot output: {e}")
pass
# Consider the bot started successfully if it's still running after the timeout
if not success and process.returncode is None:
success = True
# Start background task to monitor process output
async def monitor_output(process, bot_id):
try:
while True:
if process.returncode is not None:
break
try:
line = await process.stderr.readline()
if not line:
break
# Print the bot's output
print(f"Bot {bot_id} output: {line.decode().strip()}")
except Exception as e:
print(f"Error monitoring bot {bot_id}: {e}")
break
except Exception as e:
print(f"Monitor task exception for bot {bot_id}: {e}")
pass
asyncio.create_task(monitor_output(process, bot_id))
# Store the process
self.bots[bot_id] = process
self._save_state()
return True
except Exception:
await self._cleanup_process(process)
return False
except Exception:
return False
async def stop_bot(self, bot_id: str) -> bool:
"""Stop a running bot instance.
Args:
bot_id: ID of the bot to stop, or "all" to stop all bots
Returns:
bool: True if bot(s) were stopped successfully
"""
if bot_id == "all":
print("Stopping all bots...")
success = True
state = self._load_state()
for bid in list(state.keys()):
print(f"Stopping bot {bid}...")
if not await self._stop_single_bot(bid):
success = False
return success
# Stop a single bot
return await self._stop_single_bot(bot_id)
async def _stop_single_bot(self, bot_id: str) -> bool:
"""Stop a single bot instance.
Args:
bot_id: ID of the bot to stop
Returns:
bool: True if bot was stopped successfully
"""
# Check both local bots and state file
state = self._load_state()
process = None
if bot_id in self.bots:
process = self.bots[bot_id]
elif bot_id in state:
try:
pid = state[bot_id]['pid']
print(f"Stopping bot {bot_id} with PID {pid}")
# Try to kill the process
try:
# First try a gentle termination
os.kill(pid, signal.SIGTERM)
print(f"Sent SIGTERM to process {pid}")
# Wait a bit for the process to terminate
for i in range(50): # 5 seconds
await asyncio.sleep(0.1)
try:
os.kill(pid, 0) # Check if process exists
except ProcessLookupError:
print(f"Process {pid} terminated successfully")
break
else:
# Process didn't terminate, force kill
print(f"Process {pid} didn't terminate, sending SIGKILL")
try:
os.kill(pid, signal.SIGKILL)
print(f"Sent SIGKILL to process {pid}")
except ProcessLookupError:
print(f"Process {pid} already terminated")
pass
except ProcessLookupError:
print(f"Process {pid} not found")
pass
# Remove only this bot from state
if bot_id in state:
print(f"Removing {bot_id} from state file")
del state[bot_id]
with open(self.state_file, 'w') as f:
json.dump(state, f)
print(f"State file updated, remaining bots: {list(state.keys())}")
return True
except Exception as e:
print(f"Error stopping bot {bot_id}: {e}")
return False
else:
print(f"Bot {bot_id} not found")
return False
if process:
try:
print(f"Cleaning up process for bot {bot_id}")
await self._cleanup_process(process)
del self.bots[bot_id]
# Update state file - only remove this bot
state = self._load_state()
if bot_id in state:
print(f"Removing {bot_id} from state file")
del state[bot_id]
with open(self.state_file, 'w') as f:
json.dump(state, f)
print(f"State file updated, remaining bots: {list(state.keys())}")
return True
except Exception as e:
print(f"Error cleaning up process for bot {bot_id}: {e}")
return False
async def restart_bot(self, bot_id: str) -> bool:
"""Restart a running bot instance.
Args:
bot_id: ID of the bot to restart
Returns:
bool: True if bot was restarted successfully
"""
# Find the config file for this bot
config_path = self.config_dir / f"{bot_id}.yaml"
if not config_path.exists():
return False
# Stop the bot if it's running
if bot_id in self.bots:
if not await self.stop_bot(bot_id):
return False
# Start the bot with the same config
return await self.start_bot(config_path)
async def quiet_bot(self, bot_id: str) -> bool:
"""Disable song announcements for a running bot.
Args:
bot_id: ID of the bot to quiet
Returns:
bool: True if command was sent successfully
"""
# Check if bot exists in state
state = self._load_state()
if bot_id not in state:
print(f"Bot {bot_id} not found")
return False
try:
# Get the socket path for this bot
socket_path = Path(tempfile.gettempdir()) / f"icecast_bot_{bot_id}.sock"
if not socket_path.exists():
print(f"Socket for bot {bot_id} not found at {socket_path}")
return False
# Send quiet command
print(f"Sending quiet command to bot {bot_id}")
reader, writer = await asyncio.open_unix_connection(str(socket_path))
writer.write(b'quiet')
await writer.drain()
writer.close()
await writer.wait_closed()
print(f"Quiet command sent to bot {bot_id}")
return True
except Exception as e:
print(f"Error sending quiet command to bot {bot_id}: {e}")
return False
async def unquiet_bot(self, bot_id: str) -> bool:
"""Enable song announcements for a running bot.
Args:
bot_id: ID of the bot to unquiet
Returns:
bool: True if command was sent successfully
"""
# Check if bot exists in state
state = self._load_state()
if bot_id not in state:
print(f"Bot {bot_id} not found")
return False
try:
# Get the socket path for this bot
socket_path = Path(tempfile.gettempdir()) / f"icecast_bot_{bot_id}.sock"
if not socket_path.exists():
print(f"Socket for bot {bot_id} not found at {socket_path}")
return False
# Send unquiet command
print(f"Sending unquiet command to bot {bot_id}")
reader, writer = await asyncio.open_unix_connection(str(socket_path))
writer.write(b'unquiet')
await writer.drain()
writer.close()
await writer.wait_closed()
print(f"Unquiet command sent to bot {bot_id}")
return True
except Exception as e:
print(f"Error sending unquiet command to bot {bot_id}: {e}")
return False
async def list_bots(self) -> bool:
"""List all running bots.
Returns:
bool: True if any bots are running
"""
state = self._load_state()
# Check if any bots are running
if not state:
print("No bots running")
return False
# Track unique PIDs to avoid duplicates
seen_pids = set()
unique_bots = {}
# Filter out duplicates based on PID
for bot_id, info in state.items():
pid = info.get('pid')
if pid and pid not in seen_pids:
seen_pids.add(pid)
unique_bots[bot_id] = info
if not unique_bots:
print("No bots running")
return False
# Print header
print("\nRunning Bots:")
print("-" * 80)
print(f"{'ID':<20} {'PID':<8} {'Status':<10} {'Command':<40}")
print("-" * 80)
# Print each bot's status
for bot_id, info in unique_bots.items():
pid = info.get('pid')
config = info.get('config', '')
# Check if process is still running
try:
os.kill(pid, 0)
status = "running"
except ProcessLookupError:
status = "stopped"
continue # Skip stopped processes
except Exception:
status = "unknown"
# Get command line
try:
with open(f"/proc/{pid}/cmdline", 'rb') as f:
cmdline = f.read().replace(b'\0', b' ').decode()
except Exception:
cmdline = f"Unknown (PID: {pid})"
print(f"{bot_id:<20} {pid:<8} {status:<10} {cmdline:<40}")
print("-" * 80)
print()
return True
async def cleanup(self):
"""Clean up all running bots."""
if self.monitor_task:
self.monitor_task.cancel()
try:
await self.monitor_task
except asyncio.CancelledError:
pass
await self.stop_bot("all")
async def main():
parser = argparse.ArgumentParser(description='Icecast IRC Bot Manager')
parser.add_argument('--config', type=str, help='Path to config file or directory')
parser.add_argument('command', choices=['start', 'stop', 'restart', 'list', 'quiet', 'unquiet'], help='Command to execute')
parser.add_argument('bot_id', nargs='?', help='Bot ID for start/stop/restart/quiet/unquiet commands, or "all" to stop all bots')
args = parser.parse_args()
manager = BotManager()
should_cleanup = False # Only cleanup for certain commands
try:
if args.command == 'list':
if await manager.list_bots():
# If list_bots returns True, we've printed the list
pass
elif args.command == 'start':
should_cleanup = True # Need cleanup for start command
if not args.config:
print("Error: --config required for start command")
sys.exit(1)
config_path = Path(args.config)
if config_path.is_dir():
# Start all bots in directory
success = True
for config_file in config_path.glob('*.yaml'):
if not await manager.start_bot(config_file):
success = False
if not success:
sys.exit(1)
else:
# Start single bot
if not await manager.start_bot(config_path):
sys.exit(1)
# If we started any bots successfully, keep running until interrupted
if manager.bots:
try:
# Keep the manager running
while True:
await asyncio.sleep(1)
if not manager.bots:
break
except asyncio.CancelledError:
pass
elif args.command == 'stop':
# Don't need cleanup for stop command as it already cleans up
should_cleanup = False
if not args.bot_id:
print("Error: bot_id required for stop command (use 'all' to stop all bots)")
sys.exit(1)
if not await manager.stop_bot(args.bot_id):
sys.exit(1)
elif args.command == 'restart':
should_cleanup = True # Need cleanup for restart command
if not args.bot_id:
print("Error: bot_id required for restart command")
sys.exit(1)
if args.bot_id == "all":
print("Error: restart all is not supported")
sys.exit(1)
if not await manager.restart_bot(args.bot_id):
sys.exit(1)
# If we restarted successfully, keep running until interrupted
if manager.bots:
try:
# Keep the manager running
while True:
await asyncio.sleep(1)
if not manager.bots:
break
except asyncio.CancelledError:
pass
elif args.command == 'quiet':
should_cleanup = False # Don't need cleanup for quiet command
if not args.bot_id:
print("Error: bot_id required for quiet command")
sys.exit(1)
if args.bot_id == "all":
print("Error: quiet all is not supported")
sys.exit(1)
if not await manager.quiet_bot(args.bot_id):
sys.exit(1)
elif args.command == 'unquiet':
should_cleanup = False # Don't need cleanup for unquiet command
if not args.bot_id:
print("Error: bot_id required for unquiet command")
sys.exit(1)
if args.bot_id == "all":
print("Error: unquiet all is not supported")
sys.exit(1)
if not await manager.unquiet_bot(args.bot_id):
sys.exit(1)
except KeyboardInterrupt:
should_cleanup = True # Need cleanup for keyboard interrupt
except Exception:
should_cleanup = True # Need cleanup for errors
sys.exit(1)
finally:
# Only clean up if we need to
if should_cleanup:
await manager.cleanup()
if __name__ == '__main__':
# Set up signal handlers for graceful shutdown
def signal_handler(sig, frame):
# Don't exit immediately, let the cleanup happen
asyncio.get_event_loop().stop()
for sig in (signal.SIGTERM, signal.SIGINT):
signal.signal(sig, signal_handler)
# Run the manager
asyncio.run(main())

26
icecast-irc-bot-manager.service
View File

@ -1,26 +0,0 @@
[Unit]
Description=Icecast IRC Bot Manager
After=network.target
Wants=network.target
Documentation=https://code.cottongin.xyz/cottongin/Icecast-metadata-IRC-announcer
[Service]
Type=simple
User=icecast-bot
Group=icecast-bot
Environment=VIRTUAL_ENV=/opt/icecast-irc-bot/venv
Environment=PATH=/opt/icecast-irc-bot/venv/bin:$PATH
ExecStart=/opt/icecast-irc-bot/venv/bin/python3 /usr/local/bin/icecast-irc-bot-manager --config /etc/icecast-irc-bot/config.yaml
Restart=on-failure
RestartSec=5s
StandardOutput=journal
StandardError=journal
# Security settings
NoNewPrivileges=yes
ProtectSystem=full
ProtectHome=yes
PrivateTmp=yes
[Install]
WantedBy=multi-user.target

73
install.sh
View File

@ -1,73 +0,0 @@
#!/bin/bash
# Exit on any error
set -e
# Check if running as root
if [ "$EUID" -ne 0 ]; then
echo "Please run as root"
exit 1
fi
# Create icecast-bot user and group if they don't exist
if ! getent group icecast-bot >/dev/null; then
groupadd icecast-bot
fi
if ! getent passwd icecast-bot >/dev/null; then
useradd -r -g icecast-bot -s /bin/false icecast-bot
fi
# Create necessary directories
mkdir -p /etc/icecast-irc-bot
mkdir -p /var/log/icecast-irc-bot
mkdir -p /opt/icecast-irc-bot
# Set ownership
chown icecast-bot:icecast-bot /etc/icecast-irc-bot
chown icecast-bot:icecast-bot /var/log/icecast-irc-bot
chown icecast-bot:icecast-bot /opt/icecast-irc-bot
# Create and activate virtual environment
python3 -m venv /opt/icecast-irc-bot/venv
export VIRTUAL_ENV="/opt/icecast-irc-bot/venv"
export PATH="$VIRTUAL_ENV/bin:$PATH"
unset PYTHONHOME
# Install Python dependencies in virtual environment
pip3 install --upgrade pip
pip3 install hatchling
pip3 install -r requirements.txt
pip3 install .
# Install manager script
cp icecast-irc-bot-manager.py /usr/local/bin/icecast-irc-bot-manager
chmod +x /usr/local/bin/icecast-irc-bot-manager
# Update manager script shebang to use venv Python
sed -i "1c#\!$VIRTUAL_ENV/bin/python3" /usr/local/bin/icecast-irc-bot-manager
# Install service file
cp icecast-irc-bot-manager.service /etc/systemd/system/
systemctl daemon-reload
# Copy example config if it doesn't exist
if [ ! -f /etc/icecast-irc-bot/config.yaml ]; then
cp config.yaml.example /etc/icecast-irc-bot/config.yaml
chown icecast-bot:icecast-bot /etc/icecast-irc-bot/config.yaml
chmod 640 /etc/icecast-irc-bot/config.yaml
fi
echo "Installation complete!"
echo
echo "Next steps:"
echo "1. Configure your bot(s) in /etc/icecast-irc-bot/config.yaml"
echo "2. Start the service: systemctl start icecast-irc-bot-manager"
echo "3. Enable at boot: systemctl enable icecast-irc-bot-manager"
echo
echo "Usage:"
echo "icecast-irc-bot-manager list # List running bots"
echo "icecast-irc-bot-manager start --config FILE # Start a bot"
echo "icecast-irc-bot-manager stop BOT_ID # Stop a bot"
echo "icecast-irc-bot-manager restart BOT_ID # Restart a bot"
echo "icecast-irc-bot-manager quiet BOT_ID # Disable song announcements"
echo "icecast-irc-bot-manager unquiet BOT_ID # Enable song announcements"

12
launch.py Normal file
View File

@ -0,0 +1,12 @@
#!/usr/bin/env python3
import asyncio
from main import run_multiple_bots
CONFIG_FILES = [
'config_station1.yaml',
'config_station2.yaml',
'config_station3.yaml'
]
if __name__ == "__main__":
asyncio.run(run_multiple_bots(CONFIG_FILES))

121
logger.py
View File

@ -1,121 +0,0 @@
#!/usr/bin/env python3
"""
Logger module for the Icecast metadata IRC announcer.
Uses Loguru for simple yet powerful logging.
"""
import os
import sys
from pathlib import Path
from loguru import logger
class LogManager:
"""
Manages logging configuration for the Icecast metadata IRC announcer.
This class provides a simple interface for configuring and using Loguru
throughout the project. It handles log file rotation, formatting, and
different log levels.
"""
def __init__(self, config=None):
"""
Initialize the LogManager with optional configuration.
Args:
config: Optional dictionary with logging configuration.
If None, default configuration is used.
"""
self.config = config or {}
self.log_dir = Path(self.config.get('log_dir', 'logs'))
self.log_level = self.config.get('level', 'INFO').upper()
self.log_format = self.config.get('format',
"<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
"<level>{level: <8}</level> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
"<level>{message}</level>"
)
# Create log directory if it doesn't exist
if not self.log_dir.exists():
self.log_dir.mkdir(parents=True, exist_ok=True)
# Configure logger
self._configure_logger()
def _configure_logger(self):
"""Configure Loguru logger with appropriate sinks and formats."""
# Remove default handler
logger.remove()
# Add console handler
logger.add(
sys.stderr,
format=self.log_format,
level=self.log_level,
colorize=True
)
# Add file handler with rotation
log_file = self.log_dir / "icecast_bot.log"
logger.add(
str(log_file),
format=self.log_format,
level=self.log_level,
rotation="10 MB", # Rotate when file reaches 10MB
compression="zip", # Compress rotated logs
retention="1 week", # Keep logs for 1 week
backtrace=True, # Include backtrace in error logs
diagnose=True # Include variables in error logs
)
def get_logger(self, name=None):
"""
Get a logger instance with the given name.
Args:
name: Optional name for the logger. If None, the calling module's name is used.
Returns:
A Loguru logger instance.
"""
if name:
return logger.bind(name=name)
return logger
@staticmethod
def set_level(level):
"""
Set the log level for all handlers.
Args:
level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
"""
for handler_id in logger._core.handlers:
logger.configure(handler_id, level=level)
# Create a default instance for easy import
log_manager = LogManager()
get_logger = log_manager.get_logger
# Export the logger directly for simple usage
log = logger
# For convenience, export common log levels
debug = logger.debug
info = logger.info
warning = logger.warning
error = logger.error
critical = logger.critical
exception = logger.exception
# Example usage:
# from logger import log, debug, info, error
# log.info("This is an info message")
# debug("This is a debug message")
#
# # Or with a named logger:
# from logger import get_logger
# logger = get_logger("my_module")
# logger.info("This is a message from my_module")

1530
main.py Executable file → Normal file
View File

File diff suppressed because it is too large Load Diff

36
pyproject.toml
View File

@ -1,36 +0,0 @@
[project]
name = "icecast-irc-bot"
dynamic = ["version"]
description = "Icecast metadata IRC announcer bot"
authors = [
{name = "cottongin", email = "cottongin@cottongin.xyz"},
]
dependencies = [
"asif",
"aiohttp",
"pyyaml",
]
requires-python = ">=3.11"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.version]
source = "regex"
path = "VERSION"
pattern = "^(?P<version>[0-9]+\\.[0-9]+\\.[0-9]+)$"
[tool.hatch.build.targets.wheel]
packages = ["."]
[tool.hatch.build]
include = [
"*.py",
"README*",
"LICENSE*",
"config.yaml.example"
]
[tool.setuptools.dynamic]
version = {file = "VERSION"}

1
requirements.txt
View File

@ -1,4 +1,3 @@
asif asif
aiohttp aiohttp
pyyaml pyyaml
loguru>=0.7.0