218 lines
7.6 KiB
Python
218 lines
7.6 KiB
Python
# rhymes.py
|
|
import json
|
|
import random
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Tuple
|
|
|
|
import database
|
|
|
|
RHYMES_FILE = "resources/rhymes.json"
|
|
RHYME_LOG_FILE = "data/rhyme_log.csv"
|
|
|
|
loaded_rhymes = {}
|
|
|
|
|
|
def _ensure_log_file() -> None:
|
|
"""Create CSV log file if it doesn't exist."""
|
|
Path(RHYME_LOG_FILE).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if not Path(RHYME_LOG_FILE).exists():
|
|
with open(RHYME_LOG_FILE, "w", encoding="utf-8") as f:
|
|
f.write("timestamp,last_word,rhyme_triggered\n")
|
|
|
|
|
|
def load_rhymes() -> Tuple[bool, str]:
|
|
global loaded_rhymes
|
|
"""Load rhymes from JSON file. Returns (success, message)."""
|
|
try:
|
|
with open(RHYMES_FILE, "r", encoding="utf-8") as f:
|
|
loaded_rhymes = json.load(f)
|
|
return True, f'Loaded rhymes file "{RHYMES_FILE}"'
|
|
except FileNotFoundError:
|
|
return False, f'No rhymes file found at "{RHYMES_FILE}"'
|
|
except json.JSONDecodeError as e:
|
|
return False, f'Invalid JSON in "{RHYMES_FILE}": {e}'
|
|
|
|
|
|
def log_rhyme(last_word: str, rhyme_triggered: str) -> None:
|
|
"""Log rhyme trigger to CSV file."""
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
with open(RHYME_LOG_FILE, "a", encoding="utf-8") as f:
|
|
safe_rhyme = rhyme_triggered.replace(",", ";")
|
|
f.write(f"{timestamp},{last_word},{safe_rhyme}\n")
|
|
|
|
|
|
def get_last_word(text: str) -> str:
|
|
"""Extract last alphabetic word from text."""
|
|
truncated = text
|
|
while True:
|
|
if len(truncated) < 2 or truncated[-1].isnumeric():
|
|
return ""
|
|
if truncated[-1].isalpha() and truncated[-2].isalpha():
|
|
break
|
|
truncated = truncated[:-1]
|
|
truncated = truncated.split(" ")[-1]
|
|
return truncated if truncated.isalpha() else ""
|
|
|
|
|
|
def find_rhyme(word: str) -> str:
|
|
global loaded_rhymes
|
|
"""Find matching rhyme for given word."""
|
|
for rhyme in loaded_rhymes:
|
|
if word in rhyme["blacklist"]:
|
|
return ""
|
|
if word.endswith(tuple(rhyme["keys"])):
|
|
log_rhyme(word, rhyme["sound"])
|
|
return random.choice(rhyme["rhymes"])
|
|
return ""
|
|
|
|
|
|
async def get_guild_name(guildId, client) -> str:
|
|
guild = await client.fetch_guild(guildId)
|
|
return "[Server={0}]".format(guild.name)
|
|
|
|
|
|
async def handle_debug_commands(message, client) -> bool:
|
|
"""Handle debug commands (debug, save, load). Returns True if handled."""
|
|
message_content = message.content.lower()
|
|
|
|
if message_content == "debug poilau":
|
|
if message.author.id == 151626081458192384:
|
|
all_states = database.get_all_guild_states()
|
|
dump = {}
|
|
for guild_id, state in all_states.items():
|
|
channel_name = await get_guild_name(guild_id, client)
|
|
cooldown_dt = datetime.fromisoformat(state["cooldown_until"])
|
|
time_remaining = max(0, (cooldown_dt - datetime.now()).total_seconds())
|
|
sleeping_time = "{:.2f} s".format(time_remaining)
|
|
dump[channel_name] = {
|
|
"cooldown_until": state["cooldown_until"],
|
|
"cooldown_remaining": sleeping_time,
|
|
"self-control": state["self_control"],
|
|
"last_updated": state["last_updated"],
|
|
}
|
|
await message.author.send(
|
|
"```json\n{0}```".format(json.dumps(dump, ensure_ascii=False, indent=2))
|
|
)
|
|
return True
|
|
|
|
if message_content == "save poilau":
|
|
if message.author.id == 151626081458192384:
|
|
all_states = database.get_all_guild_states()
|
|
json_str = "```json\n{0}```".format(
|
|
json.dumps(all_states, ensure_ascii=False, indent=2)
|
|
)
|
|
await message.author.send("State persisted in SQLite database")
|
|
await message.author.send(json_str)
|
|
return True
|
|
|
|
if message_content == "load poilau":
|
|
if message.author.id == 151626081458192384:
|
|
success, msg = load_rhymes()
|
|
all_states = database.get_all_guild_states()
|
|
json_str = "```json\n{0}```".format(
|
|
json.dumps(all_states, ensure_ascii=False, indent=2)
|
|
)
|
|
await message.author.send(msg)
|
|
await message.author.send(json_str)
|
|
return True
|
|
|
|
if message_content == "tg fouras" and message.guild:
|
|
# Disable cooldown for this server (set to far future)
|
|
cooldown_date = datetime.now().replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
)
|
|
cooldown_date = cooldown_date.replace(day=cooldown_date.day + 10000)
|
|
database.update_guild_state(
|
|
str(message.guild.id),
|
|
guild_name=message.guild.name,
|
|
cooldown_until=cooldown_date.isoformat(),
|
|
self_control=2.0,
|
|
)
|
|
await message.channel.send("ok :'(")
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
async def handle_rhyme_logic(message, client) -> bool:
|
|
"""Main rhyme detection logic. Returns True if rhyme was triggered."""
|
|
message_content = message.content.lower()
|
|
last_word = get_last_word(message_content)
|
|
|
|
if message.author != client.user and message.guild and last_word:
|
|
rhyme = find_rhyme(last_word)
|
|
guild_id = str(message.guild.id)
|
|
guild_name = message.guild.name
|
|
|
|
if rhyme:
|
|
guild_state = database.get_guild_state(guild_id)
|
|
|
|
# Update guild name if changed
|
|
if guild_state["guild_name"] != guild_name:
|
|
database.update_guild_state(
|
|
guild_id,
|
|
guild_name=guild_name,
|
|
cooldown_until=guild_state["cooldown_until"],
|
|
self_control=guild_state["self_control"],
|
|
)
|
|
|
|
# Check cooldown
|
|
cooldown_dt = datetime.fromisoformat(guild_state["cooldown_until"])
|
|
now_dt = datetime.now()
|
|
|
|
if now_dt >= cooldown_dt:
|
|
self_control = guild_state["self_control"]
|
|
|
|
# Probability check
|
|
if random.random() < self_control:
|
|
new_self_control = self_control * 0.9
|
|
database.update_guild_state(
|
|
guild_id,
|
|
guild_name=guild_name,
|
|
cooldown_until=now_dt.isoformat(),
|
|
self_control=new_self_control,
|
|
)
|
|
return False
|
|
|
|
# Calculate new cooldown duration
|
|
wait_time = random.randint(0, 900)
|
|
if bool(random.getrandbits(1)):
|
|
wait_time = random.randint(900, 10800)
|
|
|
|
new_cooldown_dt = now_dt.replace(second=0, microsecond=0)
|
|
new_cooldown_dt = new_cooldown_dt.replace(
|
|
minute=new_cooldown_dt.minute + wait_time // 60
|
|
)
|
|
new_cooldown_dt = new_cooldown_dt.replace(
|
|
hour=new_cooldown_dt.hour + wait_time // 3600
|
|
)
|
|
|
|
database.update_guild_state(
|
|
guild_id,
|
|
guild_name=guild_name,
|
|
cooldown_until=new_cooldown_dt.isoformat(),
|
|
self_control=self_control + 1.0,
|
|
)
|
|
|
|
await message.channel.send(rhyme)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
async def handle_message(message, client) -> bool:
|
|
"""Main entry point for message handling."""
|
|
# Initialize database and log file on first run
|
|
database.ensure_db()
|
|
_ensure_log_file()
|
|
|
|
# Handle debug commands first
|
|
if await handle_debug_commands(message, client):
|
|
return True
|
|
|
|
# Process rhyme logic
|
|
return await handle_rhyme_logic(message, client)
|