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

initial re-implementation of basic debugging

Browse Source
This commit is contained in:
cottongin 2025-02-24 22:44:06 -08:00
parent afff3d1161
commit 0face72fd4
Signed by: cottongin
GPG Key ID: A0BD18428A296890
1 changed files with 155 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

184
main.py
View File

@ -303,7 +303,38 @@ class IcecastBot:
Returns: Returns:
bool: True if the song should be announced, False if it matches any ignore patterns. bool: True if the song should be announced, False if it matches any ignore patterns.
""" """
return not any(pattern.lower() in song.lower() for pattern in self.ignore_patterns) try:
if not song:
print("Empty song title, not announcing")
return False
if not self.ignore_patterns:
print("No ignore patterns configured, announcing all songs")
return True
# Check each pattern
for pattern in self.ignore_patterns:
try:
if not pattern:
continue
if not isinstance(pattern, str):
print(f"Invalid ignore pattern (not a string): {pattern}")
continue
if pattern.lower() in song.lower():
print(f"Song '{song}' matched ignore pattern '{pattern}', not announcing")
return False
except Exception as e:
print(f"Error checking ignore pattern '{pattern}': {str(e)}")
continue
print(f"Song '{song}' passed all ignore patterns, will announce")
return True
except Exception as e:
print(f"Exception in should_announce_song: {str(e)}")
import traceback
traceback.print_exc()
# Default to not announcing if there's an error
return False
def setup_handlers(self): def setup_handlers(self):
"""Set up all IRC event handlers and command patterns. """Set up all IRC event handlers and command patterns.
@ -731,6 +762,7 @@ class IcecastBot:
str: The current song title, or an error message if fetching failed. str: The current song title, or an error message if fetching failed.
""" """
try: try:
print(f"Fetching metadata from {self.stream_url}/{self.stream_endpoint}")
# Try different URL patterns # Try different URL patterns
base_urls = [ base_urls = [
self.stream_url, # Original URL self.stream_url, # Original URL
@ -741,33 +773,53 @@ class IcecastBot:
for base_url in base_urls: for base_url in base_urls:
try: try:
url = f"{base_url}/status-json.xsl" url = f"{base_url}/status-json.xsl"
print(f"Trying URL: {url}")
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
async with session.get(url) as response: async with session.get(url, timeout=10) as response:
if response.status == 200: if response.status == 200:
data = await response.text() data = await response.text()
print(f"Received response from {url}")
json_data = json.loads(data) try:
if 'icestats' in json_data: json_data = json.loads(data)
sources = json_data['icestats'].get('source', []) if 'icestats' in json_data:
if not isinstance(sources, list): sources = json_data['icestats'].get('source', [])
sources = [sources] if not isinstance(sources, list):
sources = [sources]
# Find our stream # Find our stream
for src in sources: for src in sources:
if src.get('listenurl', '').endswith(self.stream_endpoint): if src.get('listenurl', '').endswith(self.stream_endpoint):
title = src.get('title') or src.get('song') or src.get('current_song') title = src.get('title') or src.get('song') or src.get('current_song')
if title: if title:
return title print(f"Found title: {title}")
return title
except json.JSONDecodeError as e:
print(f"JSON decode error for {url}: {str(e)}")
continue
except aiohttp.ClientError as e: except aiohttp.ClientError as e:
print(f"Client error for {url}: {str(e)}")
continue continue
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f"JSON decode error for {url}: {str(e)}")
continue
except asyncio.TimeoutError:
print(f"Timeout fetching metadata from {url}")
continue
except Exception as e:
print(f"Unexpected error fetching from {url}: {str(e)}")
continue continue
print("All URL patterns failed, returning 'Unable to fetch metadata'")
return "Unable to fetch metadata" return "Unable to fetch metadata"
except Exception as e: except Exception as e:
return "Error fetching metadata" print(f"Exception in fetch_json_metadata: {str(e)}")
import traceback
traceback.print_exc()
return f"Error fetching metadata: {str(e)}"
async def monitor_metadata(self): async def monitor_metadata(self):
"""Monitor the Icecast stream for metadata changes. """Monitor the Icecast stream for metadata changes.
@ -826,14 +878,45 @@ class IcecastBot:
# Look for metadata marker but fetch from JSON # Look for metadata marker but fetch from JSON
if b"StreamTitle='" in buffer: if b"StreamTitle='" in buffer:
new_song = await self.fetch_json_metadata() try:
if new_song and new_song != self.current_song and "Unable to fetch metadata" not in new_song: print("Detected StreamTitle marker in buffer, fetching metadata")
self.current_song = new_song new_song = await self.fetch_json_metadata()
await self.announce_song(new_song)
# Clear buffer after metadata marker # Check if we should announce the song
buffer = buffer[buffer.find(b"';", buffer.find(b"StreamTitle='")) + 2:] if new_song and new_song != self.current_song and "Unable to fetch metadata" not in new_song:
last_json_check = current_time print(f"Song changed from '{self.current_song}' to '{new_song}'")
self.current_song = new_song
# Try to announce the song
try:
await self.announce_song(new_song)
except Exception as e:
print(f"Error announcing song: {str(e)}")
import traceback
traceback.print_exc()
else:
# No song change or unable to fetch metadata
print(f"No song change detected or unable to fetch metadata: {new_song}")
# Clear buffer after metadata marker
marker_pos = buffer.find(b"StreamTitle='")
end_pos = buffer.find(b"';", marker_pos)
if end_pos > marker_pos:
buffer = buffer[end_pos + 2:]
print("Buffer cleared after metadata marker")
else:
print(f"Could not find end of metadata marker, truncating buffer")
buffer = buffer[-8192:] # Keep last 8KB to avoid losing the end marker
# Update last check time
last_json_check = current_time
except Exception as e:
print(f"Error processing metadata marker: {str(e)}")
import traceback
traceback.print_exc()
# Reset buffer to avoid getting stuck in a loop
buffer = b""
# Keep buffer size reasonable # Keep buffer size reasonable
if len(buffer) > 65536: if len(buffer) > 65536:
@ -841,14 +924,34 @@ class IcecastBot:
# Fallback JSON check if ICY updates aren't coming through # Fallback JSON check if ICY updates aren't coming through
if current_time - last_json_check >= json_check_interval: if current_time - last_json_check >= json_check_interval:
new_song = await self.fetch_json_metadata() try:
if "Unable to fetch metadata" in new_song: print("Performing fallback JSON check")
if time.time() - last_data_received > data_timeout: new_song = await self.fetch_json_metadata()
break
if new_song and new_song != self.current_song: if "Unable to fetch metadata" in new_song:
self.current_song = new_song print("Unable to fetch metadata in fallback check")
await self.announce_song(new_song) if time.time() - last_data_received > data_timeout:
last_json_check = current_time print("Data timeout exceeded, breaking loop")
break
elif new_song and new_song != self.current_song:
print(f"Song changed in fallback check from '{self.current_song}' to '{new_song}'")
self.current_song = new_song
try:
await self.announce_song(new_song)
except Exception as e:
print(f"Error announcing song in fallback check: {str(e)}")
import traceback
traceback.print_exc()
else:
print(f"No song change detected in fallback check: {new_song}")
last_json_check = current_time
except Exception as e:
print(f"Error in fallback JSON check: {str(e)}")
import traceback
traceback.print_exc()
# Still update the check time to avoid rapid retries
last_json_check = current_time
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
@ -888,12 +991,31 @@ class IcecastBot:
- The announcement format is configured - The announcement format is configured
""" """
try: try:
print(f"Attempting to announce song: {song}")
if self.channel and self.should_announce_song(song): if self.channel and self.should_announce_song(song):
print(f"Song passed filters, preparing to announce")
# Use the stored channel object directly # Use the stored channel object directly
if hasattr(self.channel, 'name') and self.channel.name.startswith('#'): if hasattr(self.channel, 'name') and self.channel.name.startswith('#'):
await self.channel.message(self.reply.format(song=song)) try:
formatted_message = self.reply.format(song=song)
print(f"Sending message to channel {self.channel.name}: {formatted_message}")
await self.channel.message(self.reply.format(song=song))
print(f"Successfully announced song: {song}")
except Exception as e:
print(f"Error sending message to channel: {str(e)}")
import traceback
traceback.print_exc()
else:
print(f"Channel object invalid or not a channel: {self.channel}")
else:
if not self.channel:
print("Channel object is None or invalid")
elif not self.should_announce_song(song):
print(f"Song '{song}' matched ignore patterns, not announcing")
except Exception as e: except Exception as e:
pass print(f"Exception in announce_song: {str(e)}")
import traceback
traceback.print_exc()
async def start(self): async def start(self):
"""Start the IRC bot and begin processing events.""" """Start the IRC bot and begin processing events."""