g1mp/plugins/seen.py

#!/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

def truncate(text, length=50):
    """Truncate text to specified length, appending "..." if truncated."""
    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__)

    def _add_unique_hostmask(self, nick, hostmask):
        """Add hostmask to user data if it's not already present and not 'unknown'."""
        if hostmask == 'unknown':
            return []
        
        user = self.users.get(self.User.nick == nick.lower())
        if user and 'hostmasks' in user:
            if hostmask not in user['hostmasks']:
                user['hostmasks'].append(hostmask)
            return user['hostmasks']
        else:
            return [hostmask]

    @irc3.event(r'(?P<mask>\S+)!.* NICK :?(?P<new_nick>\S+)')
    def handle_nick_change(self, mask, new_nick, **kwargs):
        """Track nickname changes."""
        try:
            now = datetime.now().isoformat()
            host = mask.split('!')[1] if '!' in mask else 'unknown'
            
            self.users.upsert({
                'nick': mask.nick.lower(),
                'hostmasks': self._add_unique_hostmask(mask.nick.lower(), host),
                'last_nick_change': {'old': mask.nick, 'new': new_nick, 'time': now}
            }, self.User.nick == mask.nick.lower())

            self.users.upsert({
                'nick': new_nick.lower(),
                'hostmasks': self._add_unique_hostmask(new_nick.lower(), host),
                'last_join': None,
                'last_message': None,
                'last_nick_change': {'old': mask.nick, 'new': new_nick, 'time': now}
            }, self.User.nick == new_nick.lower())

        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:
            now = datetime.now().isoformat()
            self.users.upsert({
                'nick': mask.nick.lower(),
                'hostmasks': self._add_unique_hostmask(mask.nick.lower(), mask.host),
                'last_join': now,
                'last_message': None,
                'last_nick_change': None
            }, self.User.nick == mask.nick.lower())
        except Exception as e:
            self.log.error(f"Error tracking join for {mask.nick}: {e}")

    @irc3.event(irc3.rfc.PRIVMSG)
    def handle_message(self, target, data, mask, **kwargs):
        """Track user messages."""
        try:
            now = datetime.now().isoformat()
            self.users.upsert({
                'nick': mask.nick.lower(),
                'hostmasks': self._add_unique_hostmask(mask.nick.lower(), mask.host),
                'last_message': {'text': data, 'time': now}
            }, self.User.nick == mask.nick.lower())
        except Exception as e:
            self.log.error(f"Error tracking message from {mask.nick}: {e}")

    @command
    async def seen(self, mask, target, args):
        """Retrieve a user's last activity.
        
        %%seen <nick>
        """
        try:
            requested_nick = args['<nick>'].lower()
            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)

            # Hostmasks
            if 'hostmasks' in user and user['hostmasks']:
                hostmasks_str = ", ".join(user['hostmasks'])
                response.append(
                    ircstyle.style("🌐 Hostmasks: ", fg="cyan") +
                    ircstyle.style(hostmasks_str, fg="white")
                )

            # Last join - only if time exists
            if 'last_join' in user and user['last_join']:
                try:
                    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"))
                except ValueError:
                    response.append(ircstyle.style("🕒 Last join: ", fg="cyan") + ircstyle.style("Invalid Time Format", fg="white"))

            # Last message
            if 'last_message' in user and user['last_message']:
                try:
                    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)
                    )
                except (ValueError, KeyError):
                    response.append(ircstyle.style("💬 Last message: ", fg="green") + ircstyle.style("Invalid or Missing Data", fg="white"))

            # Last nick change
            if 'last_nick_change' in user and user['last_nick_change']:
                try:
                    change_time = datetime.fromisoformat(user['last_nick_change']['time']).strftime('%Y-%m-%d %H:%M:%S')
                    old_nick = user['last_nick_change']['old']
                    new_nick = user['last_nick_change']['new']
                    response.append(
                        ircstyle.style(f"📛 Nickname change: ", fg="purple") +
                        ircstyle.style(f"Changed from {old_nick} to {new_nick} at {change_time}", fg="white")
                    )
                except (ValueError, KeyError):
                    response.append(ircstyle.style(f"📛 Nickname change: ", fg="purple") + ircstyle.style("Invalid or Missing Data", fg="white"))

            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))