g1mp/plugins/seen.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
IRC Bot Plugin for Tracking User Activity with TinyDB

This module implements an IRC bot plugin to track various activities of users in an IRC channel,
including their join times, messages sent, and nickname changes. The plugin uses TinyDB for
persistent storage of user data.

Features:
    - Tracks when users join a channel.
    - Logs user messages with timestamps.
    - Monitors nickname changes.
    - Provides a '!seen' command to retrieve the last activity of a specified user.

Dependencies:
    - irc3: For IRC protocol handling.
    - tinydb: For managing user data in a JSON file.
    - ircstyle: For formatting IRC messages with color and style.
    - logging: For error logging.
    - humanize: For converting time differences into human-readable format.

Usage:
    - Install required packages: pip install irc3 tinydb ircstyle humanize
    - Use within an IRC bot framework that utilizes irc3 plugins.

Storage:
    - User data is stored in 'seen.json' in the same directory as this script.

Author:
    - Zodiac

Date:
    - 02/18/2025

Note:
    - Ensure your bot environment has write permissions for the JSON database file.
"""

from datetime import datetime
from tinydb import TinyDB, Query
import irc3
from irc3.plugins.command import command
import ircstyle
import logging
import humanize

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."""

    requires = [
        'plugins.users',
    ]

    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']:
                response.append(ircstyle.style("🕒 Last join: ", fg="cyan") + 
                                ircstyle.style(humanize.naturaltime(datetime.now() - datetime.fromisoformat(user['last_join'])), fg="white"))

            # Last message
            if 'last_message' in user and user['last_message']:
                try:
                    msg_text = truncate(user['last_message']['text'])
                    response.append(
                        ircstyle.style("💬 Last message: ", fg="green") +
                        ircstyle.style(humanize.naturaltime(datetime.now() - datetime.fromisoformat(user['last_message']['time'])), fg="grey") + " " +
                        ircstyle.style(msg_text, fg="white", italics=True)
                    )
                except 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:
                    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} ", fg="white") +
                        ircstyle.style(humanize.naturaltime(datetime.now() - datetime.fromisoformat(user['last_nick_change']['time'])), fg="white")
                    )
                except 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))