codey/discord-havoc
1
0
Fork 0
Code Issues Pull Requests Wiki Activity

Compare commits

base: codey:5d881369e90921b43051bf4a693c276c9b6ca5f5
Branches Tags
codey:main
..
compare: codey:863ebeb91992ea21780cababa4943484bd3d4604
Branches Tags
codey:main

1 Commits
5d881369e9 ... 863ebeb919

Author SHA1 Message Date
codey
863ebeb919 add quote.py to repo/not loaded 2025-06-27 16:12:39 -04:00
7 changed files with 262 additions and 2149 deletions
Expand all files Collapse all files

250
cogs/quote.py
View File

@@ -16,52 +16,43 @@ import aiosqlite as sqlite3
from discord.ext import bridge, commands from discord.ext import bridge, commands
from disc_havoc import Havoc from disc_havoc import Havoc
class DB: class DB:
"""DB Utility for Quote Cog""" """DB Utility for Quote Cog"""
def __init__(self, bot: Havoc): def __init__(self, bot: Havoc):
self.bot: Havoc = bot self.bot: Havoc = bot
self.db_path = os.path.join( self.db_path = os.path.join("/", "usr", "local", "share",
"/", "usr", "local", "share", "sqlite_dbs", "quotes.db" "sqlite_dbs", "quotes.db")
)
self.hp_chanid = 1157529874936909934 self.hp_chanid = 1157529874936909934
async def get_quote_count(self): async def get_quote_count(self):
"""Get Quote Count""" """Get Quote Count"""
async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
async with await db_conn.execute( async with await db_conn.execute("SELECT COUNT (*) FROM quotes") as db_cursor:
"SELECT COUNT (*) FROM quotes"
) as db_cursor:
result = await db_cursor.fetchone() result = await db_cursor.fetchone()
return result[-1] return result[-1]
async def remove_quote(self, quote_id: int): async def remove_quote(self, quote_id: int):
"""Remove Quote from DB""" """Remove Quote from DB"""
try: try:
async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
async with await db_conn.execute( async with await db_conn.execute("DELETE FROM quotes WHERE id = ?", (quote_id,)) as _:
"DELETE FROM quotes WHERE id = ?", (quote_id,)
) as _:
await db_conn.commit() await db_conn.commit()
return True return True
except Exception as e: # noqa except Exception as e: # noqa
_channel = self.bot.get_channel(self.hp_chanid) _channel = self.bot.get_channel(self.hp_chanid)
if isinstance(_channel, discord.TextChannel): if isinstance(_channel, discord.TextChannel):
await _channel.send(traceback.format_exc()) await _channel.send(traceback.format_exc())
return False return False
async def add_quote( async def add_quote(self, message_id: int, channel_id: int,
self, quoted_member_id: int,
message_id: int, message_time: int,
channel_id: int, quoter_friendly: str,
quoted_member_id: int, quoted_friendly: str,
message_time: int, channel_friendly: str,
quoter_friendly: str, message_content: str,
quoted_friendly: str, ):
channel_friendly: str,
message_content: str,
):
"""Add Quote to DB""" """Add Quote to DB"""
params = ( params = (
quoter_friendly, quoter_friendly,
@@ -80,23 +71,16 @@ class DB:
async with sqlite3.connect(self.db_path, timeout=2) as db_conn: async with sqlite3.connect(self.db_path, timeout=2) as db_conn:
# pylint: disable=line-too-long # pylint: disable=line-too-long
db_conn.row_factory = sqlite3.Row db_conn.row_factory = sqlite3.Row
async with await db_conn.execute( async with await db_conn.execute("INSERT INTO quotes (added_by, added_at, quoted_user_display, quoted_user_memberid, quoted_channel_display, quoted_channel_id, quoted_message_id, quoted_message_time, added_by_friendly, quoted_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
"INSERT INTO quotes (added_by, added_at, quoted_user_display, quoted_user_memberid, quoted_channel_display, quoted_channel_id, quoted_message_id, quoted_message_time, added_by_friendly, quoted_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", params) as _: # pylint: enable=line-too-long
params,
) as _: # pylint: enable=line-too-long
await db_conn.commit() await db_conn.commit()
return True return True
except Exception as e: # noqa except Exception as e: # noqa
return traceback.format_exc() return traceback.format_exc()
async def fetch_quote( async def fetch_quote(self, random: bool = False, quoteid: Optional[int] = None,
self, added_by: Optional[str] = None, quoted_user: Optional[str] = None,
random: bool = False, content: Optional[str] = None):
quoteid: Optional[int] = None,
added_by: Optional[str] = None,
quoted_user: Optional[str] = None,
content: Optional[str] = None,
):
"""Fetch Quote from DB""" """Fetch Quote from DB"""
try: try:
query_head = "SELECT id, added_by_friendly, added_at, quoted_user_display, quoted_channel_display, quoted_message_time, quoted_message FROM quotes" query_head = "SELECT id, added_by_friendly, added_at, quoted_user_display, quoted_channel_display, quoted_message_time, quoted_message FROM quotes"
@@ -124,34 +108,36 @@ class DB:
results = await db_cursor.fetchall() results = await db_cursor.fetchall()
if not results: if not results:
return { return {
"err": "No results for query", 'err': 'No results for query',
} }
if random or quoteid: if random or quoteid:
chosen = results[-1] chosen = results[-1]
return {str(k): v for k, v in chosen.items()} return {
str(k): v for k,v in chosen.items()
}
else: else:
return [{str(k): v for k, v in _.items()} for _ in results] return [
except Exception as e: # noqa { str(k): v for k,v in _.items() }
for _ in results
]
except Exception as e: # noqa
return traceback.format_exc() return traceback.format_exc()
class Quote(commands.Cog): class Quote(commands.Cog):
"""Quote Cog for Havoc""" """Quote Cog for Havoc"""
def __init__(self, bot: Havoc): def __init__(self, bot: Havoc):
self.bot: Havoc = bot self.bot: Havoc = bot
self.db = DB(self.bot) self.db = DB(self.bot)
def is_homeserver(): # type: ignore def is_homeserver(): # type: ignore
"""Check if channel/interaction is within homeserver""" """Check if channel/interaction is within homeserver"""
def predicate(ctx): def predicate(ctx):
try: try:
return ctx.guild.id == 1145182936002482196 return ctx.guild.id == 1145182936002482196
except Exception as e: # noqa except Exception as e: # noqa
traceback.print_exc() traceback.print_exc()
return False return False
return commands.check(predicate) return commands.check(predicate)
@commands.message_command(name="Add Quote") @commands.message_command(name="Add Quote")
@@ -160,73 +146,58 @@ class Quote(commands.Cog):
hp_chanid = 1157529874936909934 hp_chanid = 1157529874936909934
try: try:
if message.author.bot: if message.author.bot:
return await ctx.respond( return await ctx.respond("Quotes are for real users, not bots.", ephemeral=True)
"Quotes are for real users, not bots.", ephemeral=True
)
quoter_friendly = ctx.author.display_name quoter_friendly = ctx.author.display_name
quoted_message_id = message.id quoted_message_id = message.id
quoted_channel_friendly = f"#{ctx.channel.name}" quoted_channel_friendly = f'#{ctx.channel.name}'
quoted_channel_id = ctx.channel.id quoted_channel_id = ctx.channel.id
message_content = message.content message_content = message.content
message_author_friendly = message.author.display_name message_author_friendly = message.author.display_name
message_author_id = message.author.id message_author_id = message.author.id
message_time = int(message.created_at.timestamp()) message_time = int(message.created_at.timestamp())
message_escaped = discord.utils.escape_mentions( message_escaped = discord.utils.escape_mentions(discord.utils.escape_markdown(message_content)).strip()
discord.utils.escape_markdown(message_content)
).strip()
if len(message_escaped) < 3: if len(message_escaped) < 3:
return await ctx.respond( return await ctx.respond("**Error**: Message (text content) is not long enough to quote.", ephemeral=True)
"**Error**: Message (text content) is not long enough to quote.",
ephemeral=True,
)
if len(message_escaped) > 512: if len(message_escaped) > 512:
return await ctx.respond( return await ctx.respond("**Error**: Message (text content) is too long to quote.", ephemeral=True)
"**Error**: Message (text content) is too long to quote.",
ephemeral=True,
)
result = await self.db.add_quote( result = await self.db.add_quote(message_id=quoted_message_id,
message_id=quoted_message_id, channel_id=quoted_channel_id,
channel_id=quoted_channel_id, quoted_member_id=message_author_id,
quoted_member_id=message_author_id, message_time=message_time,
message_time=message_time, quoter_friendly=quoter_friendly,
quoter_friendly=quoter_friendly, quoted_friendly=message_author_friendly,
quoted_friendly=message_author_friendly, channel_friendly=quoted_channel_friendly,
channel_friendly=quoted_channel_friendly, message_content=message_content)
message_content=message_content,
)
if not result: if not result:
return await ctx.respond("Failed!", ephemeral=True) return await ctx.respond("Failed!", ephemeral=True)
else: else:
return await ctx.respond("OK!", ephemeral=True) return await ctx.respond("OK!", ephemeral=True)
except Exception as e: # noqa except Exception as e: # noqa
_channel = self.bot.get_channel(hp_chanid) _channel = self.bot.get_channel(hp_chanid)
if not isinstance(_channel, discord.TextChannel): if not isinstance(_channel, discord.TextChannel):
return return
await _channel.send(traceback.format_exc()) await _channel.send(traceback.format_exc())
@bridge.bridge_command(aliases=["rand"])
@is_homeserver() # pylint: disable=too-many-function-args @bridge.bridge_command(aliases=['rand'])
@is_homeserver() # pylint: disable=too-many-function-args
async def randquote(self, ctx): async def randquote(self, ctx):
"""Get a random quote""" """Get a random quote"""
try: try:
random_quote = await self.db.fetch_quote(random=True) random_quote = await self.db.fetch_quote(random=True)
if random_quote.get("err"): if random_quote.get('err'):
return await ctx.respond("Failed to get a quote") return await ctx.respond("Failed to get a quote")
quote_id = random_quote.get("id") quote_id = random_quote.get('id')
quoted_friendly = random_quote.get("quoted_user_display", "Unknown") quoted_friendly = random_quote.get('quoted_user_display', 'Unknown')
adder_friendly = random_quote.get("added_by_friendly", "Unknown") adder_friendly = random_quote.get('added_by_friendly', 'Unknown')
message_time = datetime.datetime.fromtimestamp( message_time = datetime.datetime.fromtimestamp(random_quote.get('quoted_message_time'))
random_quote.get("quoted_message_time") message_channel = random_quote.get('quoted_channel_display')
) quote_added_at = datetime.datetime.fromtimestamp(random_quote.get('added_at'))
message_channel = random_quote.get("quoted_channel_display") quote_content = random_quote.get('quoted_message')
quote_added_at = datetime.datetime.fromtimestamp(
random_quote.get("added_at")
)
quote_content = random_quote.get("quoted_message")
embed = discord.Embed( embed = discord.Embed(
colour=discord.Colour.orange(), colour=discord.Colour.orange(),
@@ -236,38 +207,32 @@ class Quote(commands.Cog):
embed.add_field(name="Original Message Time", value=message_time) embed.add_field(name="Original Message Time", value=message_time)
embed.add_field(name="Channel", value=message_channel) embed.add_field(name="Channel", value=message_channel)
embed.add_field(name="Quote ID", value=quote_id) embed.add_field(name="Quote ID", value=quote_id)
embed.footer = discord.EmbedFooter( embed.footer = discord.EmbedFooter(text=f"Added by {adder_friendly} {quote_added_at}")
text=f"Added by {adder_friendly} {quote_added_at}"
)
return await ctx.respond(embed=embed) return await ctx.respond(embed=embed)
except Exception as e: # noqa except Exception as e: # noqa
error = await ctx.respond(traceback.format_exc()) error = await ctx.respond(traceback.format_exc())
await asyncio.sleep(10) await asyncio.sleep(10)
await error.delete() await error.delete()
@bridge.bridge_command(aliases=["qg"]) @bridge.bridge_command(aliases=['qg'])
@is_homeserver() # pylint: disable=too-many-function-args @is_homeserver() # pylint: disable=too-many-function-args
async def quoteget(self, ctx, quoteid): async def quoteget(self, ctx, quoteid):
"""Get a specific quote by ID""" """Get a specific quote by ID"""
try: try:
if not str(quoteid).strip().isnumeric(): if not str(quoteid).strip().isnumeric():
return await ctx.respond("**Error**: Quote ID must be numeric.") return await ctx.respond("**Error**: Quote ID must be numeric.")
fetched_quote = await self.db.fetch_quote(quoteid=quoteid) fetched_quote = await self.db.fetch_quote(quoteid=quoteid)
if fetched_quote.get("err"): if fetched_quote.get('err'):
return await ctx.respond("**Error**: Quote not found") return await ctx.respond("**Error**: Quote not found")
quote_id = fetched_quote.get("id") quote_id = fetched_quote.get('id')
quoted_friendly = fetched_quote.get("quoted_user_display", "Unknown") quoted_friendly = fetched_quote.get('quoted_user_display', 'Unknown')
adder_friendly = fetched_quote.get("added_by_friendly", "Unknown") adder_friendly = fetched_quote.get('added_by_friendly', 'Unknown')
message_time = datetime.datetime.fromtimestamp( message_time = datetime.datetime.fromtimestamp(fetched_quote.get('quoted_message_time'))
fetched_quote.get("quoted_message_time") message_channel = fetched_quote.get('quoted_channel_display')
) quote_added_at = datetime.datetime.fromtimestamp(fetched_quote.get('added_at'))
message_channel = fetched_quote.get("quoted_channel_display") quote_content = fetched_quote.get('quoted_message')
quote_added_at = datetime.datetime.fromtimestamp(
fetched_quote.get("added_at")
)
quote_content = fetched_quote.get("quoted_message")
embed = discord.Embed( embed = discord.Embed(
colour=discord.Colour.orange(), colour=discord.Colour.orange(),
@@ -277,48 +242,40 @@ class Quote(commands.Cog):
embed.add_field(name="Original Message Time", value=message_time) embed.add_field(name="Original Message Time", value=message_time)
embed.add_field(name="Channel", value=message_channel) embed.add_field(name="Channel", value=message_channel)
embed.add_field(name="Quote ID", value=quote_id) embed.add_field(name="Quote ID", value=quote_id)
embed.footer = discord.EmbedFooter( embed.footer = discord.EmbedFooter(text=f"Added by {adder_friendly} {quote_added_at}")
text=f"Added by {adder_friendly} {quote_added_at}"
)
return await ctx.respond(embed=embed) return await ctx.respond(embed=embed)
except Exception as e: # noqa except Exception as e: # noqa
error = await ctx.respond(traceback.format_exc()) error = await ctx.respond(traceback.format_exc())
await asyncio.sleep(10) await asyncio.sleep(10)
await error.delete() await error.delete()
@bridge.bridge_command(aliases=["qs"]) @bridge.bridge_command(aliases=['qs'])
@is_homeserver() # pylint: disable=too-many-function-args @is_homeserver() # pylint: disable=too-many-function-args
async def quotesearch(self, ctx, *, content: str): async def quotesearch(self, ctx, *, content: str):
"""Search for a quote (by content)""" """Search for a quote (by content)"""
try: try:
found_quotes = await self.db.fetch_quote(content=content) found_quotes = await self.db.fetch_quote(content=content)
if isinstance(found_quotes, dict) and found_quotes.get("err"): if isinstance(found_quotes, dict) and found_quotes.get('err'):
return await ctx.respond( return await ctx.respond(f"Quote search failed: {found_quotes.get('err')}")
f"Quote search failed: {found_quotes.get('err')}"
)
embeds = [] embeds = []
for quote in found_quotes: for quote in found_quotes:
quote_id = quote.get("id") quote_id = quote.get('id')
quoted_friendly = quote.get("quoted_user_display", "Unknown") quoted_friendly = quote.get('quoted_user_display', 'Unknown')
adder_friendly = quote.get("added_by_friendly", "Unknown") adder_friendly = quote.get('added_by_friendly', 'Unknown')
message_time = datetime.datetime.fromtimestamp( message_time = datetime.datetime.fromtimestamp(quote.get('quoted_message_time'))
quote.get("quoted_message_time") message_channel = quote.get('quoted_channel_display')
) quote_added_at = datetime.datetime.fromtimestamp(quote.get('added_at'))
message_channel = quote.get("quoted_channel_display") quote_content = quote.get('quoted_message')
quote_added_at = datetime.datetime.fromtimestamp(quote.get("added_at"))
quote_content = quote.get("quoted_message")
# await ctx.respond(f"**{quoted_friendly}**: {quote}")ed_friendly = quote.get('quoted_user_display', 'Unknown') # await ctx.respond(f"**{quoted_friendly}**: {quote}")ed_friendly = quote.get('quoted_user_display', 'Unknown')
adder_friendly = quote.get("added_by_friendly", "Unknown") adder_friendly = quote.get('added_by_friendly', 'Unknown')
message_time = datetime.datetime.fromtimestamp( message_time = datetime.datetime.fromtimestamp(quote.get('quoted_message_time'))
quote.get("quoted_message_time") message_channel = quote.get('quoted_channel_display')
) quote_added_at = datetime.datetime.fromtimestamp(quote.get('added_at'))
message_channel = quote.get("quoted_channel_display") quote = quote.get('quoted_message')
quote_added_at = datetime.datetime.fromtimestamp(quote.get("added_at"))
quote = quote.get("quoted_message")
# await ctx.respond(f"**{quoted_friendly}**: {quote}") # await ctx.respond(f"**{quoted_friendly}**: {quote}")
embed = discord.Embed( embed = discord.Embed(
@@ -329,33 +286,29 @@ class Quote(commands.Cog):
embed.add_field(name="Original Message Time", value=str(message_time)) embed.add_field(name="Original Message Time", value=str(message_time))
embed.add_field(name="Channel", value=message_channel) embed.add_field(name="Channel", value=message_channel)
embed.add_field(name="Quote ID", value=quote_id) embed.add_field(name="Quote ID", value=quote_id)
embed.footer = discord.EmbedFooter( embed.footer = discord.EmbedFooter(text=f"Added by {adder_friendly} {quote_added_at}")
text=f"Added by {adder_friendly} {quote_added_at}"
)
embeds.append(embed) embeds.append(embed)
return await ctx.respond(embeds=embeds) return await ctx.respond(embeds=embeds)
except Exception as e: except Exception as e:
await ctx.respond(f"Error: {type(e).__name__} - {str(e)}") await ctx.respond(f"Error: {type(e).__name__} - {str(e)}")
@bridge.bridge_command(aliases=["nq"]) @bridge.bridge_command(aliases=['nq'])
@is_homeserver() # pylint: disable=too-many-function-args @is_homeserver() # pylint: disable=too-many-function-args
async def nquotes(self, ctx): async def nquotes(self, ctx):
"""Get # of quotes stored""" """Get # of quotes stored"""
try: try:
quote_count = await self.db.get_quote_count() quote_count = await self.db.get_quote_count()
if not quote_count: if not quote_count:
return await ctx.respond("**Error**: No quotes found!") return await ctx.respond("**Error**: No quotes found!")
return await ctx.respond( return await ctx.respond(f"I currently have **{quote_count}** quotes stored.")
f"I currently have **{quote_count}** quotes stored."
)
except Exception as e: except Exception as e:
await ctx.respond(f"Error: {type(e).__name__} - {str(e)}") await ctx.respond(f"Error: {type(e).__name__} - {str(e)}")
@bridge.bridge_command(aliases=["qr"]) @bridge.bridge_command(aliases=['qr'])
@commands.is_owner() @commands.is_owner()
@is_homeserver() # pylint: disable=too-many-function-args @is_homeserver() # pylint: disable=too-many-function-args
async def quoteremove(self, ctx, quoteid): async def quoteremove(self, ctx, quoteid):
"""Remove a quote (by id) """Remove a quote (by id)
Owner only""" Owner only"""
@@ -363,7 +316,7 @@ class Quote(commands.Cog):
if not str(quoteid).strip().isnumeric(): if not str(quoteid).strip().isnumeric():
return await ctx.respond("**Error**: Quote ID must be numeric.") return await ctx.respond("**Error**: Quote ID must be numeric.")
quoteid = int(quoteid) quoteid = int(quoteid)
remove_quote = await self.db.remove_quote(quoteid) remove_quote = await self.db.remove_quote(quoteid)
if not remove_quote: if not remove_quote:
return await ctx.respond("**Error**: Failed!", ephemeral=True) return await ctx.respond("**Error**: Failed!", ephemeral=True)
return await ctx.respond("Removed!", ephemeral=True) return await ctx.respond("Removed!", ephemeral=True)
@@ -371,7 +324,6 @@ class Quote(commands.Cog):
except Exception as e: except Exception as e:
await ctx.respond(f"Error: {type(e).__name__} - {str(e)}") await ctx.respond(f"Error: {type(e).__name__} - {str(e)}")
def setup(bot): def setup(bot):
"""Run on Cog Load""" """Run on Cog Load"""
bot.add_cog(Quote(bot)) bot.add_cog(Quote(bot))

127
cogs/radio.py
View File

@@ -1,6 +1,5 @@
import logging import logging
import traceback import traceback
import asyncio
from typing import Optional from typing import Optional
from discord.ext import bridge, commands, tasks from discord.ext import bridge, commands, tasks
from util.radio_util import get_now_playing, skip from util.radio_util import get_now_playing, skip
@@ -44,86 +43,78 @@ class Radio(commands.Cog):
return commands.check(predicate) return commands.check(predicate)
# @bridge.bridge_command() @bridge.bridge_command()
# @commands.is_owner() @commands.is_owner()
# async def reinitradio(self, ctx) -> None: async def reinitradio(self, ctx) -> None:
# """ """
# Reinitialize serious.FM Reinitialize serious.FM
# """ """
# loop: discord.asyncio.AbstractEventLoop = self.bot.loop loop: discord.asyncio.AbstractEventLoop = self.bot.loop
# loop.create_task(self.radio_init()) loop.create_task(self.radio_init())
# await ctx.respond("Done!", ephemeral=True) await ctx.respond("Done!", ephemeral=True)
# async def radio_init(self) -> None:
# """Init Radio"""
# try:
# (radio_guild, radio_chan) = self.channels["sfm"]
# guild: Optional[discord.Guild] = self.bot.get_guild(radio_guild)
# if not guild:
# return
# channel = guild.get_channel(radio_chan)
# if not isinstance(channel, discord.VoiceChannel):
# return
# if not self.bot.voice_clients:
# await channel.connect()
# try:
# try:
# self.radio_state_loop.cancel()
# except Exception as e:
# logging.debug("Failed to cancel radio_state_loop: %s", str(e))
# self.radio_state_loop.start()
# logging.info("radio_state_loop task started!")
# except Exception as e:
# logging.critical("Could not start task... Exception: %s", str(e))
# traceback.print_exc()
# except Exception as e:
# logging.debug("Exception: %s", str(e))
# traceback.print_exc()
# return
async def radio_init(self) -> None: async def radio_init(self) -> None:
"""Init Radio"""
try: try:
self.radio_state_loop.start() (radio_guild, radio_chan) = self.channels["sfm"]
guild: Optional[discord.Guild] = self.bot.get_guild(radio_guild)
if not guild:
return
channel = guild.get_channel(radio_chan)
if not isinstance(channel, discord.VoiceChannel):
return
if not self.bot.voice_clients:
await channel.connect()
try:
try:
self.radio_state_loop.cancel()
except Exception as e:
logging.debug("Failed to cancel radio_state_loop: %s", str(e))
self.radio_state_loop.start()
logging.info("radio_state_loop task started!")
except Exception as e:
logging.critical("Could not start task... Exception: %s", str(e))
traceback.print_exc()
except Exception as e: except Exception as e:
logging.critical("Failed to start radio state loop: %s", str(e)) logging.debug("Exception: %s", str(e))
traceback.print_exc() traceback.print_exc()
return
@tasks.loop(seconds=5.0) @tasks.loop(seconds=5.0)
async def radio_state_loop(self) -> None: async def radio_state_loop(self) -> None:
"""Radio State Loop""" """Radio State Loop"""
try: try:
# (radio_guild, radio_chan) = self.channels["sfm"] (radio_guild, radio_chan) = self.channels["sfm"]
# try: try:
# vc: discord.VoiceProtocol = self.bot.voice_clients[-1] vc: discord.VoiceProtocol = self.bot.voice_clients[-1]
# except Exception as e: except Exception as e:
# logging.debug( logging.debug(
# "No voice client, establishing new VC connection... (Exception: %s)", "No voice client, establishing new VC connection... (Exception: %s)",
# str(e), str(e),
# ) )
# guild: Optional[discord.Guild] = self.bot.get_guild(radio_guild) guild: Optional[discord.Guild] = self.bot.get_guild(radio_guild)
# if not guild: if not guild:
# return return
# channel = guild.get_channel(radio_chan) channel = guild.get_channel(radio_chan)
# if not isinstance(channel, discord.VoiceChannel): if not isinstance(channel, discord.VoiceChannel):
# return return
# await channel.connect() await channel.connect()
# vc = self.bot.voice_clients[-1] vc = self.bot.voice_clients[-1]
# if not vc.is_playing() or vc.is_paused(): # type: ignore if not vc.is_playing() or vc.is_paused(): # type: ignore
# """ """
# Mypy does not seem aware of the is_playing, play, and is_paused methods, Mypy does not seem aware of the is_playing, play, and is_paused methods,
# but they exist. but they exist.
# """ """
# logging.info("Detected VC not playing... playing!") logging.info("Detected VC not playing... playing!")
# source: discord.FFmpegAudio = discord.FFmpegOpusAudio(self.STREAM_URL) source: discord.FFmpegAudio = discord.FFmpegOpusAudio(self.STREAM_URL)
# vc.play( # type: ignore vc.play( # type: ignore
# source, source,
# after=lambda e: logging.info("Error: %s", e) if e else None, after=lambda e: logging.info("Error: %s", e) if e else None,
# ) )
# Get Now Playing # Get Now Playing
np_track: Optional[str] = await get_now_playing() np_track: Optional[str] = await get_now_playing()
if not self.LAST_NP_TRACK or (np_track and not self.LAST_NP_TRACK == np_track): if np_track and not self.LAST_NP_TRACK == np_track:
logging.critical("Setting: %s", np_track)
self.LAST_NP_TRACK = np_track self.LAST_NP_TRACK = np_track
await self.bot.change_presence( await self.bot.change_presence(
activity=discord.Activity( activity=discord.Activity(

322
cogs/sing.py
View File

@@ -7,137 +7,9 @@ import regex
from util.sing_util import Utility from util.sing_util import Utility
from discord.ext import bridge, commands from discord.ext import bridge, commands
from disc_havoc import Havoc from disc_havoc import Havoc
import random
import string
import json
import os
BOT_CHANIDS = [] BOT_CHANIDS = []
# Global storage for pagination data (survives across interactions)
PAGINATION_DATA = {}
PAGINATION_FILE = "pagination_data.json"
def load_pagination_data():
"""Load pagination data from JSON file"""
global PAGINATION_DATA
try:
if os.path.exists(PAGINATION_FILE):
with open(PAGINATION_FILE, 'r') as f:
PAGINATION_DATA = json.load(f)
except Exception:
PAGINATION_DATA = {}
def save_pagination_data():
"""Save pagination data to JSON file (excluding non-serializable embeds)"""
try:
# Create serializable copy without embeds
serializable_data = {}
for key, value in PAGINATION_DATA.items():
serializable_data[key] = {
"pages": value["pages"],
"song": value["song"],
"artist": value["artist"],
"total_pages": value["total_pages"]
}
with open(PAGINATION_FILE, 'w') as f:
json.dump(serializable_data, f)
except Exception:
pass
import logging
from typing import Optional, Union
from regex import Pattern
import discord
import regex
from util.sing_util import Utility
from discord.ext import bridge, commands
from disc_havoc import Havoc
import random
import string
BOT_CHANIDS = []
# Global storage for pagination data (survives across interactions)
PAGINATION_DATA: dict[str, dict] = {}
class LyricsPaginator(discord.ui.View):
def __init__(self, pages, song, artist):
super().__init__(timeout=None) # No timeout - buttons work permanently
self.pages = pages
self.song = song
self.artist = artist
self.current_page = 0
# Generate a unique ID for this paginator instance
self.custom_id_base = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
# Pre-generate all embeds to avoid creating them on each page turn
self.embeds = []
# URL encode artist and song for the link
import urllib.parse
encoded_artist = urllib.parse.quote(self.artist, safe='')
encoded_song = urllib.parse.quote(self.song, safe='')
lyrics_url = f"https://codey.lol/#{encoded_artist}/{encoded_song}"
for i, page in enumerate(pages):
embed = discord.Embed(title=f"{self.song}", color=discord.Color.blue(), url=lyrics_url)
embed.set_author(name=self.artist)
embed.description = page
# Set footer with just page info for multi-page, or no footer for single page
if len(pages) > 1:
embed.set_footer(text=f"Page {i + 1} of {len(pages)}", icon_url="https://codey.lol/favicon.ico")
self.embeds.append(embed)
# Store pagination data for persistence across bot restarts
if len(pages) > 1:
# Store data (embeds will be recreated as needed since they can't be JSON serialized)
serializable_data = {
"pages": pages,
"song": song,
"artist": artist,
"total_pages": len(pages)
}
# Store in memory with embeds for performance
PAGINATION_DATA[self.custom_id_base] = {
**serializable_data,
"embeds": self.embeds
}
# Save serializable data to file
save_pagination_data()
def get_view_for_page(self, page_num: int):
"""Create a view with properly configured buttons for the given page"""
view = discord.ui.View(timeout=None)
if len(self.pages) > 1:
# Previous button
prev_button = discord.ui.Button(
label="Previous",
style=discord.ButtonStyle.secondary,
emoji="⬅️",
disabled=(page_num == 0),
custom_id=f"lyrics_prev:{self.custom_id_base}",
row=0
)
# Next button
next_button = discord.ui.Button(
label="Next",
style=discord.ButtonStyle.primary,
emoji="➡️",
disabled=(page_num == len(self.pages) - 1),
custom_id=f"lyrics_next:{self.custom_id_base}",
row=0
)
view.add_item(prev_button)
view.add_item(next_button)
return view
class Sing(commands.Cog): class Sing(commands.Cog):
"""Sing Cog for Havoc""" """Sing Cog for Havoc"""
@@ -151,84 +23,6 @@ class Sing(commands.Cog):
r"\x0f|\x1f|\035|\002|\u2064|\x02|(\x03([0-9]{1,2}))|(\x03|\003)(?:\d{1,2}(?:,\d{1,2})?)?", r"\x0f|\x1f|\035|\002|\u2064|\x02|(\x03([0-9]{1,2}))|(\x03|\003)(?:\d{1,2}(?:,\d{1,2})?)?",
regex.UNICODE, regex.UNICODE,
) )
# Persistent interactions handled by on_interaction listener
@commands.Cog.listener()
async def on_interaction(self, interaction: discord.Interaction):
"""Handle persistent lyrics pagination interactions"""
if interaction.type != discord.InteractionType.component:
return
custom_id = interaction.data.get("custom_id", "")
if not (custom_id.startswith("lyrics_prev:") or custom_id.startswith("lyrics_next:")):
return
try:
# Extract the base custom ID and direction
if custom_id.startswith("lyrics_prev:"):
direction = -1
base_id = custom_id.replace("lyrics_prev:", "")
else: # lyrics_next:
direction = 1
base_id = custom_id.replace("lyrics_next:", "")
# Get pagination data from JSON persistence
data = PAGINATION_DATA.get(base_id)
if not data:
await interaction.response.send_message(
"⚠️ **Navigation Expired**\n"
"These buttons are from a previous session. "
"Use `/s <song>` or `/sing <artist> : <song>` to get fresh lyrics!",
ephemeral=True,
)
return
# Recreate embeds if they don't exist (loaded from JSON)
if "embeds" not in data or not data["embeds"]:
paginator = LyricsPaginator(data["pages"], data["song"], data["artist"])
paginator.custom_id_base = base_id
data["embeds"] = paginator.embeds
# Update in-memory data
PAGINATION_DATA[base_id] = data
# Get current page from embed footer
if not interaction.message or not interaction.message.embeds:
await interaction.response.defer()
return
current_embed = interaction.message.embeds[0]
footer_text = current_embed.footer.text if current_embed.footer else ""
# Extract current page number (format: "Page X of Y")
current_page = 0
if "Page " in footer_text and " of " in footer_text:
try:
page_part = footer_text.split("Page ")[1].split(" of ")[0]
current_page = int(page_part) - 1
except:
pass
# Calculate new page
new_page = current_page + direction
total_pages = len(data["pages"])
if new_page < 0 or new_page >= total_pages:
await interaction.response.defer() # Just acknowledge, don't change anything
return
# Create new paginator and get view for new page
paginator = LyricsPaginator(data["pages"], data["song"], data["artist"])
paginator.custom_id_base = base_id # Preserve same custom ID
view = paginator.get_view_for_page(new_page)
# Update the message
await interaction.response.edit_message(embed=data["embeds"][new_page], view=view)
except Exception as e:
try:
await interaction.response.send_message(f"Error handling pagination: {str(e)}", ephemeral=True)
except:
pass
def is_spamchan(): # type: ignore def is_spamchan(): # type: ignore
"""Check if channel is spam chan""" """Check if channel is spam chan"""
@@ -266,18 +60,18 @@ class Sing(commands.Cog):
"**Error**: No song specified, no activity found to read." "**Error**: No song specified, no activity found to read."
) )
# Initial response that we'll edit later await ctx.respond(
message = await ctx.respond("*Searching for lyrics...*") "*Searching...*"
) # Must respond to interactions within 3 seconds, per Discord
parsed = self.utility.parse_song_input(song, activity) parsed = self.utility.parse_song_input(song, activity)
if isinstance(parsed, tuple): if isinstance(parsed, tuple):
(search_artist, search_song, search_subsearch) = parsed (search_artist, search_song, search_subsearch) = parsed
# Update the message to show what we're searching for
await message.edit(content=f"*Searching for '{search_song}' by {search_artist}...*")
# await ctx.respond(f"So, {search_song} by {search_artist}? Subsearch: {search_subsearch} I will try...") # Commented, useful for debugging
search_result: Optional[list] = await self.utility.lyric_search( search_result: Optional[list] = await self.utility.lyric_search(
search_artist, search_song, search_subsearch, is_spam_channel=(ctx.channel.id in BOT_CHANIDS) search_artist, search_song, search_subsearch
) )
if not search_result: if not search_result:
@@ -303,26 +97,29 @@ class Sing(commands.Cog):
search_result_wrapped_short: list[str] = search_result[ search_result_wrapped_short: list[str] = search_result[
2 2
] # Third is short wrapped lyrics ] # Third is short wrapped lyrics
# Now all channels get full pagination - removed the early return restriction if ctx.channel.id not in BOT_CHANIDS:
short_lyrics = " ".join(
search_result_wrapped_short
) # Replace with shortened lyrics for non spamchans
short_lyrics = regex.sub(
r"\p{Vert_Space}", " / ", short_lyrics.strip()
)
return await ctx.respond(
f"**{search_result_song}** by **{search_result_artist}**\n-# {short_lyrics}"
)
# Use the appropriate wrapped lyrics based on channel type out_messages: list = []
is_spam_channel = ctx.channel.id in BOT_CHANIDS footer: str = "" # Placeholder
if is_spam_channel: for c, section in enumerate(search_result_wrapped):
# Spam channels get full-length lyrics if c == len(search_result_wrapped):
pages = search_result_wrapped.copy() footer = f"`Found on: {search_result_src}`"
else: section = regex.sub(r"\p{Vert_Space}", " / ", section.strip())
# Non-spam channels get shorter lyrics for better UX msg: str = f"**{search_result_song}** by **{search_result_artist}**\n-# {section}\n{footer}"
pages = search_result_wrapped.copy() # For now, still use full but we'll modify limits later if c > 1:
msg = "\n".join(msg.split("\n")[1:])
# Add source info to last page out_messages.append(msg.strip())
if pages: for msg in out_messages:
pages[-1] += f"\n\n`Found on: {search_result_src}`" await ctx.send(msg)
# Create and send view with paginator
paginator = LyricsPaginator(pages, search_result_song, search_result_artist)
view = paginator.get_view_for_page(0)
# Edit the existing message to show the lyrics
await message.edit(content=None, embed=paginator.embeds[0], view=view)
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
await ctx.respond(f"ERR: {str(e)}") await ctx.respond(f"ERR: {str(e)}")
@@ -352,29 +149,26 @@ class Sing(commands.Cog):
) )
member_id: int = member.id # if not(member.id == PODY_ID) else 1234134345497837679 # Use Thomas for Pody! member_id: int = member.id # if not(member.id == PODY_ID) else 1234134345497837679 # Use Thomas for Pody!
activity: Optional[discord.Activity] = None activity: Optional[discord.Activity] = None
# Initial response that we'll edit later if IS_SPAMCHAN:
message = await ctx.respond(f"*Reading activity of {member_display}...*" if IS_SPAMCHAN else "*Processing...*") await ctx.respond(f"***Reading activity of {member_display}...***")
for _activity in ctx.interaction.guild.get_member(member_id).activities: for _activity in ctx.interaction.guild.get_member(member_id).activities:
if _activity.type == discord.ActivityType.listening: if _activity.type == discord.ActivityType.listening:
activity = _activity activity = _activity
parsed: Union[tuple, bool] = self.utility.parse_song_input( parsed: Union[tuple, bool] = self.utility.parse_song_input(
song=None, activity=activity song=None, activity=activity
) )
if not parsed: if not parsed:
if IS_SPAMCHAN: return await ctx.respond(
return await message.edit(content=f"Could not parse activity of {member_display}.") f"Could not parse activity of {member_display}.", ephemeral=True
else: )
return await ctx.respond(
f"Could not parse activity of {member_display}.", ephemeral=True
)
if isinstance(parsed, tuple): if isinstance(parsed, tuple):
(search_artist, search_song, search_subsearch) = parsed (search_artist, search_song, search_subsearch) = parsed
await message.edit(content=f"*Searching for '{search_song}' by {search_artist}...*") await ctx.respond(
"*Searching...*"
) # Must respond to interactions within 3 seconds, per Discord
search_result: Optional[list] = await self.utility.lyric_search( search_result: Optional[list] = await self.utility.lyric_search(
search_artist, search_song, search_subsearch, is_spam_channel=(ctx.channel.id in BOT_CHANIDS) search_artist, search_song, search_subsearch
) )
if not search_result: if not search_result:
await ctx.respond("ERR: No search result") await ctx.respond("ERR: No search result")
@@ -399,24 +193,33 @@ class Sing(commands.Cog):
2 2
] # Third index is shortened lyrics ] # Third index is shortened lyrics
# All channels now get full paginated response if not IS_SPAMCHAN:
short_lyrics = " ".join(
search_result_wrapped_short
) # Replace with shortened lyrics for non spamchans
short_lyrics = regex.sub(
r"\p{Vert_Space}", " / ", short_lyrics.strip()
)
return await ctx.respond(
f"**{search_result_song}** by **{search_result_artist}**\n-# {short_lyrics}"
)
# Create paginator for lyrics out_messages: list = []
if not search_result_wrapped: footer: str = ""
return await ctx.respond("No lyrics found.") c: int = 0
for section in search_result_wrapped:
# Use the already smartly-wrapped pages from sing_util c += 1
pages = search_result_wrapped.copy() # Already processed by _smart_lyrics_wrap if c == len(search_result_wrapped):
footer = f"`Found on: {search_result_src}`"
# Add source info to last page # if ctx.guild.id == 1145182936002482196:
if pages: # section = section.upper()
pages[-1] += f"\n\n`Found on: {search_result_src}`" section = regex.sub(r"\p{Vert_Space}", " / ", section.strip())
msg: str = f"**{search_result_song}** by **{search_result_artist}**\n-# {section}\n{footer}"
# Create and send view with paginator if c > 1:
paginator = LyricsPaginator(pages, search_result_song, search_result_artist) msg = "\n".join(msg.split("\n")[1:])
view = paginator.get_view_for_page(0) out_messages.append(msg.strip())
# Edit the existing message to show the lyrics for msg in out_messages:
await message.edit(content=None, embed=paginator.embeds[0], view=view) await ctx.send(msg)
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
return await ctx.respond(f"ERR: {str(e)}") return await ctx.respond(f"ERR: {str(e)}")
@@ -424,5 +227,4 @@ class Sing(commands.Cog):
def setup(bot) -> None: def setup(bot) -> None:
"""Run on Cog Load""" """Run on Cog Load"""
load_pagination_data() # Load existing pagination data from file
bot.add_cog(Sing(bot)) bot.add_cog(Sing(bot))

8
disc_havoc.py
View File

@@ -25,11 +25,10 @@ cogs_list: list[str] = [
"sing", "sing",
"meme", "meme",
"lovehate", "lovehate",
# "ollama", "radio",
# "radio",
] ]
bot_activity = discord.CustomActivity(name="I LIKE TURTLES") bot_activity = discord.CustomActivity(name="I made cookies!")
load_dotenv() load_dotenv()
@@ -43,7 +42,7 @@ class Havoc(bridge.Bot):
command_prefix=".", command_prefix=".",
intents=intents, intents=intents,
owner_ids=OWNERS, owner_ids=OWNERS,
activity=None, activity=bot_activity,
help_command=commands.MinimalHelpCommand(), help_command=commands.MinimalHelpCommand(),
) )
self.BOT_CHANIDS = BOT_CHANIDS self.BOT_CHANIDS = BOT_CHANIDS
@@ -84,7 +83,6 @@ class Havoc(bridge.Bot):
@commands.Cog.listener() @commands.Cog.listener()
async def on_ready(self) -> None: async def on_ready(self) -> None:
"""Run on Bot Ready""" """Run on Bot Ready"""
await self.change_presence(activity=None)
logging.info("%s online!", self.user) logging.info("%s online!", self.user)

6
util/radio_util.py
View File

@@ -25,11 +25,7 @@ async def get_now_playing() -> Optional[str]:
) as request: ) as request:
request.raise_for_status() request.raise_for_status()
response_json = await request.json() response_json = await request.json()
artistsong: str = "N/A - N/A" artistsong = response_json.get("artistsong")
artist: Optional[str] = response_json.get("artist")
song: Optional[str] = response_json.get("song")
if artist and song:
artistsong = f"{artist} - {song}"
return artistsong return artistsong
except Exception as e: except Exception as e:
logging.critical("Now playing retrieval failed: %s", str(e)) logging.critical("Now playing retrieval failed: %s", str(e))

190
util/sing_util.py
View File

@@ -1,12 +1,10 @@
import logging import logging
import traceback
import aiohttp
import regex import regex
import discord import aiohttp
import textwrap
import traceback
from discord import Activity from discord import Activity
from typing import Optional, Union
logger = logging.getLogger(__name__)
class Utility: class Utility:
@@ -16,134 +14,9 @@ class Utility:
self.api_url: str = "http://127.0.0.1:52111/lyric/search" self.api_url: str = "http://127.0.0.1:52111/lyric/search"
self.api_src: str = "DISC-HAVOC" self.api_src: str = "DISC-HAVOC"
def _smart_lyrics_wrap(self, lyrics: str, max_length: int = 1500, single_page: bool = False, max_verses: int = 100, max_lines: int = 150) -> list[str]:
"""
Intelligently wrap lyrics to avoid breaking verses in the middle
Prioritizes keeping verses intact over page length consistency
Args:
lyrics: Raw lyrics text
max_length: Maximum character length per page (soft limit)
single_page: If True, return only the first page
Returns:
List of lyrics pages
"""
if not lyrics:
return []
# Strip markdown formatting from lyrics
lyrics = discord.utils.escape_markdown(lyrics)
verses = []
current_verse: list[str] = []
# Handle both regular newlines and zero-width space newlines
lines = lyrics.replace("\u200b\n", "\n").split("\n")
empty_line_count = 0
for line in lines:
stripped_line = line.strip()
if not stripped_line or stripped_line in ["", "\u200b"]:
empty_line_count += 1
# One empty line indicates a section break (be more aggressive)
if empty_line_count >= 1 and current_verse:
verses.append("\n".join(current_verse))
current_verse = []
empty_line_count = 0
else:
empty_line_count = 0
current_verse.append(stripped_line)
# Add the last verse if it exists
if current_verse:
verses.append("\n".join(current_verse))
# If we have too few verses (verse detection failed), fallback to line-based splitting
if len(verses) <= 1:
all_lines = lyrics.split("\n")
verses = []
current_chunk = []
for line in all_lines:
current_chunk.append(line.strip())
# Split every 8-10 lines to create artificial "verses"
if len(current_chunk) >= 8:
verses.append("\n".join(current_chunk))
current_chunk = []
# Add remaining lines
if current_chunk:
verses.append("\n".join(current_chunk))
if not verses:
return [lyrics[:max_length]]
# If single page requested, return first verse or truncated
if single_page:
result = verses[0]
if len(result) > max_length:
# Try to fit at least part of the first verse
lines_in_verse = result.split("\n")
truncated_lines = []
current_length = 0
for line in lines_in_verse:
if current_length + len(line) + 1 <= max_length - 3: # -3 for "..."
truncated_lines.append(line)
current_length += len(line) + 1
else:
break
result = "\n".join(truncated_lines) + "..." if truncated_lines else verses[0][:max_length-3] + "..."
return [result]
# Group complete verses into pages, never breaking a verse
# Limit by character count, verse count, AND line count for visual appeal
max_verses_per_page = max_verses
max_lines_per_page = max_lines
pages = []
current_page_verses: list[str] = []
current_page_length = 0
current_page_lines = 0
for verse in verses:
verse_length = len(verse)
# Count lines properly - handle both regular newlines and zero-width space newlines
verse_line_count = verse.count("\n") + verse.count("\u200b\n") + 1
# Calculate totals if we add this verse (including separator)
separator_length = 3 if current_page_verses else 0 # "\n\n\n" between verses
separator_lines = 3 if current_page_verses else 0 # 3 empty lines between verses
total_length_with_verse = current_page_length + separator_length + verse_length
total_lines_with_verse = current_page_lines + separator_lines + verse_line_count
# Check all three limits: character, verse count, and line count
exceeds_length = total_length_with_verse > max_length
exceeds_verse_count = len(current_page_verses) >= max_verses_per_page
exceeds_line_count = total_lines_with_verse > max_lines_per_page
# If adding this verse would exceed any limit AND we already have verses on the page
if (exceeds_length or exceeds_verse_count or exceeds_line_count) and current_page_verses:
# Finish current page with existing verses
pages.append("\n\n".join(current_page_verses))
current_page_verses = [verse]
current_page_length = verse_length
current_page_lines = verse_line_count
else:
# Add verse to current page
current_page_verses.append(verse)
current_page_length = total_length_with_verse
current_page_lines = total_lines_with_verse
# Add the last page if it has content
if current_page_verses:
pages.append("\n\n".join(current_page_verses))
return pages if pages else [lyrics[:max_length]]
def parse_song_input( def parse_song_input(
self, song: str | None = None, activity: Activity | None = None, self, song: Optional[str] = None, activity: Optional[Activity] = None
) -> bool | tuple: ) -> Union[bool, tuple]:
""" """
Parse Song (Sing Command) Input Parse Song (Sing Command) Input
@@ -163,7 +36,7 @@ class Utility:
match activity.name.lower(): match activity.name.lower():
case "codey toons" | "cider" | "sonixd": case "codey toons" | "cider" | "sonixd":
search_artist: str = " ".join( search_artist: str = " ".join(
str(activity.state).strip().split(" ")[1:], str(activity.state).strip().split(" ")[1:]
) )
search_artist = regex.sub( search_artist = regex.sub(
r"(\s{0,})(\[(spotify|tidal|sonixd|browser|yt music)])$", r"(\s{0,})(\[(spotify|tidal|sonixd|browser|yt music)])$",
@@ -178,13 +51,13 @@ class Utility:
search_song = str(activity.details) search_song = str(activity.details)
song = f"{search_artist} : {search_song}" song = f"{search_artist} : {search_song}"
case "spotify": case "spotify":
if not activity.title or not activity.artist: # type: ignore[attr-defined] if not activity.title or not activity.artist: # type: ignore
""" """
Attributes exist, but mypy does not recognize them. Ignored. Attributes exist, but mypy does not recognize them. Ignored.
""" """
return False return False
search_artist = str(activity.artist) # type: ignore[attr-defined] search_artist = str(activity.artist) # type: ignore
search_song = str(activity.title) # type: ignore[attr-defined] search_song = str(activity.title) # type: ignore
song = f"{search_artist} : {search_song}" song = f"{search_artist} : {search_song}"
case "serious.fm" | "cocks.fm" | "something": case "serious.fm" | "cocks.fm" | "something":
if not activity.details: if not activity.details:
@@ -205,27 +78,27 @@ class Utility:
return False return False
search_artist = song.split(search_split_by)[0].strip() search_artist = song.split(search_split_by)[0].strip()
search_song = "".join(song.split(search_split_by)[1:]).strip() search_song = "".join(song.split(search_split_by)[1:]).strip()
search_subsearch: str | None = None search_subsearch: Optional[str] = None
if ( if (
search_split_by == ":" and len(song.split(":")) > 2 search_split_by == ":" and len(song.split(":")) > 2
): # Support sub-search if : is used (per instructions) ): # Support sub-search if : is used (per instructions)
search_song = song.split( search_song = song.split(
search_split_by, search_split_by
)[ )[
1 1
].strip() # Reduce search_song to only the 2nd split of : [the rest is meant to be lyric text] ].strip() # Reduce search_song to only the 2nd split of : [the rest is meant to be lyric text]
search_subsearch = "".join( search_subsearch = "".join(
song.split(search_split_by)[2:], song.split(search_split_by)[2:]
) # Lyric text from split index 2 and beyond ) # Lyric text from split index 2 and beyond
return (search_artist, search_song, search_subsearch) return (search_artist, search_song, search_subsearch)
except Exception as e: except Exception as e:
logger.debug("Exception: %s", str(e)) logging.debug("Exception: %s", str(e))
traceback.print_exc() traceback.print_exc()
return False return False
async def lyric_search( async def lyric_search(
self, artist: str, song: str, sub: str | None = None, is_spam_channel: bool = True, self, artist: str, song: str, sub: Optional[str] = None
) -> list | None: ) -> Optional[list]:
""" """
Lyric Search Lyric Search
@@ -267,7 +140,7 @@ class Utility:
return [(f"ERR: {response.get('errorText')}",)] return [(f"ERR: {response.get('errorText')}",)]
out_lyrics = regex.sub( out_lyrics = regex.sub(
r"<br>", "\u200b\n", response.get("lyrics", ""), r"<br>", "\u200b\n", response.get("lyrics", "")
) )
response_obj: dict = { response_obj: dict = {
"artist": response.get("artist"), "artist": response.get("artist"),
@@ -281,15 +154,24 @@ class Utility:
lyrics = response_obj.get("lyrics") lyrics = response_obj.get("lyrics")
if not lyrics: if not lyrics:
return None return None
# Use different limits based on channel type response_obj["lyrics"] = textwrap.wrap(
if is_spam_channel: text=lyrics.strip(),
# Spam channels: higher limits for more content per page width=1500,
response_obj["lyrics"] = self._smart_lyrics_wrap(lyrics.strip(), max_length=8000, max_verses=100, max_lines=150) drop_whitespace=False,
else: replace_whitespace=False,
# Non-spam channels: much shorter limits for better UX in regular channels break_long_words=True,
response_obj["lyrics"] = self._smart_lyrics_wrap(lyrics.strip(), max_length=2000, max_verses=15, max_lines=25) break_on_hyphens=True,
max_lines=8,
response_obj["lyrics_short"] = self._smart_lyrics_wrap(lyrics.strip(), max_length=500, single_page=True) )
response_obj["lyrics_short"] = textwrap.wrap(
text=lyrics.strip(),
width=750,
drop_whitespace=False,
replace_whitespace=False,
break_long_words=True,
break_on_hyphens=True,
max_lines=1,
)
return [ return [
( (
@@ -304,4 +186,4 @@ class Utility:
] ]
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
return [f"Retrieval failed: {e!s}"] return [f"Retrieval failed: {str(e)}"]

1508
uv.lock generated
View File

File diff suppressed because it is too large Load Diff