197 lines
7.8 KiB
Python
197 lines
7.8 KiB
Python
#!/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."""
|
|
|
|
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))
|