diff --git a/plugins/seen.py b/plugins/seen.py new file mode 100644 index 0000000..c893a50 --- /dev/null +++ b/plugins/seen.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +IRC Bot Plugin for Tracking User Activity with TinyDB. + +This plugin tracks user join times, messages, and nickname changes. Provides +a !seen command to check last activity. Stores data in "seen.json". + +Dependencies: + - irc3 + - tinydb + - ircstyle + - logging + - re +""" + +from datetime import datetime +from tinydb import TinyDB, Query +import irc3 +from irc3.plugins.command import command +import ircstyle +import logging +import re + + +def truncate(text, length=50): + """ + Truncate text to specified length, appending "..." if truncated. + + Args: + text (str): Input text + length (int): Maximum length + + Returns: + str: Truncated text + """ + if not isinstance(text, str): + text = str(text) + return text if len(text) <= length else text[:length] + "..." + + +@irc3.plugin +class UserActivityTracker: + """IRC bot plugin to track and report user activity using TinyDB.""" + + def __init__(self, bot): + self.bot = bot + self.db = TinyDB('seen.json') + self.users = self.db.table('users') + self.User = Query() + self.log = logging.getLogger(__name__) + + @irc3.event(r'(?P\S+)!.* NICK :?(?P\S+)') + def handle_nick_change(self, mask, new_nick, **kwargs): + """Track nickname changes and update the old nickname with the new one.""" + self.log.debug(f"Received old_nick: {mask.nick}, new_nick: {new_nick}, mask: {mask}, kwargs: {kwargs}") + try: + now = datetime.now().isoformat() + host = mask.split('!')[1] if '!' in mask else 'unknown' + + # Update the record for the old nickname + old_user = self.users.get(self.User.nick == mask.nick) + if old_user: + # Mark the nickname change in the old user's record + old_user_data = { + 'last_nick_change': {'old': mask.nick, 'new': new_nick, 'time': now}, + 'nick': new_nick + } + self.users.update(old_user_data, self.User.nick == mask.nick) + else: + # Insert a record for the old nickname if it doesn't exist + self.users.insert({ + 'nick': mask.nick, + 'host': host, + 'last_nick_change': {'old': mask.nick, 'new': new_nick, 'time': now}, + 'last_join': None, + 'last_message': None + }) + + # Update or insert the record for the new nickname + new_user = self.users.get(self.User.nick == new_nick) + if new_user: + self.users.update({'host': host}, self.User.nick == new_nick) + else: + self.users.insert({ + 'nick': new_nick, + 'host': host, + 'last_nick_change': None, + 'last_join': None, + 'last_message': None + }) + except Exception as e: + self.log.error(f"Error tracking nick change {mask.nick}→{new_nick}: {e}") + + @irc3.event(irc3.rfc.JOIN) + def handle_join(self, mask, channel, **kwargs): + """Track when users join a channel.""" + try: + nick = mask.nick + host = mask.host + now = datetime.now().isoformat() + user = self.users.get(self.User.nick == nick) + update_data = {'last_join': now, 'host': host} + + if user: + self.users.update(update_data, self.User.nick == nick) + else: + self.users.insert({'nick': nick, 'host': host, 'last_join': now, + 'last_message': None, 'last_nick_change': None}) + except Exception as e: + self.log.error(f"Error tracking join for {nick}: {e}") + + @irc3.event(irc3.rfc.PRIVMSG) + def handle_message(self, target, data, mask, **kwargs): + """Track user messages.""" + try: + now = datetime.now().isoformat() + nick = mask.nick # Get the nickname from the mask + host = mask.host # Get the host from the mask + message = {'text': data, 'time': now} + self.users.upsert( + {'nick': nick, 'host': host, 'last_message': message}, + self.User.nick == nick + ) + except Exception as e: + self.log.error(f"Error tracking message from {nick}: {e}") + + @command(permission=None) + async def seen(self, mask, target, args): + """Retrieve a user's last activity. + + %%seen + """ + try: + requested_nick = args[''] + user = self.users.get(self.User.nick == requested_nick) + + if not user: + msg = ircstyle.style(f"🚫 {requested_nick} has never been observed.", fg="red") + self.bot.privmsg(target, msg) + return + + response = [] + header = ircstyle.style(f"📊 Activity report for {requested_nick}:", fg="blue", bold=True) + response.append(header) + + # Last join + if user.get('last_join'): + join_time = datetime.fromisoformat(user['last_join']).strftime('%Y-%m-%d %H:%M:%S') + response.append( + ircstyle.style("🕒 Last join: ", fg="cyan") + + ircstyle.style(join_time, fg="white") + ) + + # Last message + if user.get('last_message'): + msg_time = datetime.fromisoformat(user['last_message']['time']).strftime('%Y-%m-%d %H:%M:%S') + msg_text = truncate(user['last_message']['text']) + response.append( + ircstyle.style("💬 Last message: ", fg="green") + + ircstyle.style(f"[{msg_time}] ", fg="grey") + + ircstyle.style(msg_text, fg="white", italics=True) + ) + + # Last nick change + if user.get('last_nick_change'): + old_nick = user['last_nick_change']['old'] + new_nick = user['last_nick_change']['new'] + change_time = datetime.fromisoformat(user['last_nick_change']['time']).strftime('%Y-%m-%d %H:%M:%S') + response.append( + ircstyle.style(f"📛 Nickname change: ", fg="purple") + + ircstyle.style(f"Changed from {old_nick} to {new_nick} at {change_time}", fg="white") + ) + + # Handle no tracked activity + if len(response) == 1: # Only header present + response.append(ircstyle.style("No tracked activities.", fg="yellow")) + + for line in response: + self.bot.privmsg(target, line) + + except Exception as e: + self.log.error(f"Error in !seen command: {e}") + self.bot.privmsg( + target, + ircstyle.style("❌ Internal error processing request.", fg="red", bold=True) + )