dandri/meshcore-bot
1
0
Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-09-01 16:48:26 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
fc63039a48918caf4f2a8f10335c7aa5141076db
meshcore-bot/modules/commands/stats_command.py
T
agessaman fc63039a48 fix: address Codex review of tonight's work 
Correctness:

- {path_distance} was always blank in production. The resolution code that
  builds repeater_info dropped latitude/longitude in every branch, so the
  calculator could never find a coordinate. My tests passed hand-built
  dicts straight to the calculator and never exercised the builder, which
  is why they stayed green. Coordinates are now carried through all four
  construction sites, and the new tests drive _lookup_repeater_names via
  its lookup_func hook so the real builder runs.

- The #80 route guard was defeated two ways in the channel handler. When
  the RF data was an uncorrelated fallback, control fell through to the
  raw-hex and routing_info fallbacks below, which took the route from the
  unrelated packet anyway; the guard had actually made that path
  reachable. message.routing_info was also assigned unconditionally, and
  the path command reads it. "Not attributable" is now a terminal branch
  and the routing_info hand-off checks provenance.

- Same fix was incomplete for DMs: routing_info was captured and turned
  into path_info before the provenance check ran, so the later check only
  declined to overwrite an already-wrong value. Guarded at the source.

- Rendering could transmit for real. Capture only intercepts
  send_response, but advert calls send_advert() directly and
  send_response_chunked never checked capture_sink. Chunked sends are now
  captured, and rendering is opt-in via BaseCommand.render_safe (default
  False) instead of a denylist that cannot be complete. This also closes
  the DM-only leak: schedule is not marked safe, so {cmd:schedule} can no
  longer broadcast configuration to a channel.

- Multi-part rendered output was rejoined into one oversized send.
  Scheduled messages are now split to the RF body budget and sent through
  send_channel_messages_chunked, on character boundaries so multi-byte
  text is not corrupted.

- _last_path_distance_km is instance state that was only set on success,
  so an invalid path request could show the previous request's distance.
  Reset at the start of every execute().

- Stale-contact retries were only counted on non-OK results, so timeouts
  and exceptions left a contact eligible forever and could recreate the
  storm. All failed attempts count now.

Web viewer:

- A failed config reload was reported as a successful save. The API and
  UI now distinguish "saved and active" from "saved, restart needed".

- Preview count was unbounded; clamped to 1-20.

- update_ini_values is a read-modify-replace with no locking, so two
  concurrent viewer saves could lose one. Serialised behind a lock.
2026-08-22 00:16:03 -07:00

874 lines
39 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Stats command for the MeshCore Bot
Provides comprehensive statistics about bot usage, messages, and activity
"""
import time
from typing import Any, Optional
from ..models import MeshMessage
from .base_command import BaseCommand
class StatsCommand(BaseCommand):
"""Handles the stats command with comprehensive data collection.
This command tracks usage statistics including messages, commands, and routing paths.
It provides insights into bot activity and network performance over the last 24 hours.
"""
# Plugin metadata
# Read-only informational output; safe for scheduled {cmd:...} rendering.
render_safe = True
name = "stats"
keywords = ['stats']
description = "Show statistics for past 24 hours. Use 'stats messages', 'stats channels', 'stats paths', or 'stats adverts' for specific stats."
category = "analytics"
# Documentation
short_description = "Show bot usage statistics for past 24 hours"
usage = "stats [messages|channels|paths|adverts]"
examples = ["stats", "stats channels"]
parameters = [
{"name": "type", "description": "messages, channels, or paths (optional)"}
]
# Web-viewer settings schema (see modules/settings_schema.py)
settings_schema = [
{"key": "data_retention_days", "label": "Data retention", "type": "int",
"min": 1, "default": 7, "unit": "days",
"help": "Days of stats data to keep (recommended 7-30)."},
{"key": "auto_cleanup", "label": "Auto cleanup", "type": "bool",
"default": True,
"help": "Automatically delete data older than the retention window."},
{"key": "track_all_messages", "label": "Track all messages", "type": "bool",
"default": True,
"help": "On: record all messages. Off: only command executions."},
{"key": "track_command_details", "label": "Track command details", "type": "bool",
"default": True,
"help": "Record detailed command execution info."},
{"key": "anonymize_users", "label": "Anonymize users", "type": "bool",
"default": False,
"help": "Replace user IDs with anonymous identifiers in stats."},
]
def __init__(self, bot: Any):
"""Initialize the stats command.
Args:
bot: The bot instance.
"""
super().__init__(bot)
self._load_config()
self._init_stats_tables()
def _load_config(self) -> None:
"""Load configuration settings for stats command."""
self.stats_enabled = self.get_config_value('Stats_Command', 'enabled', fallback=None, value_type='bool')
if self.stats_enabled is None:
self.stats_enabled = self.get_config_value('Stats_Command', 'stats_enabled', fallback=True, value_type='bool')
# Optional: collect_stats controls dashboard/stat table writes independently
# of whether the user-facing stats command is enabled.
self.collect_stats = self.get_config_value('Stats_Command', 'collect_stats', fallback=None, value_type='bool')
if self.collect_stats is None:
self.collect_stats = True
self.data_retention_days = self.get_config_value('Stats_Command', 'data_retention_days', fallback=7, value_type='int')
self.auto_cleanup = self.get_config_value('Stats_Command', 'auto_cleanup', fallback=True, value_type='bool')
self.track_all_messages = self.get_config_value('Stats_Command', 'track_all_messages', fallback=True, value_type='bool')
self.track_command_details = self.get_config_value('Stats_Command', 'track_command_details', fallback=True, value_type='bool')
self.anonymize_users = self.get_config_value('Stats_Command', 'anonymize_users', fallback=False, value_type='bool')
def _init_stats_tables(self) -> None:
"""Initialize database tables for stats tracking.
Creates tables for message stats, command stats, and path stats if they
don't already exist. Also sets up necessary indexes for performance.
"""
try:
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Create message_stats table for tracking all messages
cursor.execute('''
CREATE TABLE IF NOT EXISTS message_stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp INTEGER NOT NULL,
sender_id TEXT NOT NULL,
channel TEXT,
content TEXT NOT NULL,
is_dm BOOLEAN NOT NULL,
hops INTEGER,
snr REAL,
rssi INTEGER,
path TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create command_stats table for tracking bot commands
cursor.execute('''
CREATE TABLE IF NOT EXISTS command_stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp INTEGER NOT NULL,
sender_id TEXT NOT NULL,
command_name TEXT NOT NULL,
channel TEXT,
is_dm BOOLEAN NOT NULL,
response_sent BOOLEAN NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create path_stats table for tracking longest paths
cursor.execute('''
CREATE TABLE IF NOT EXISTS path_stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp INTEGER NOT NULL,
sender_id TEXT NOT NULL,
channel TEXT,
path_length INTEGER NOT NULL,
path_string TEXT NOT NULL,
hops INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create indexes for better performance
cursor.execute('CREATE INDEX IF NOT EXISTS idx_message_timestamp ON message_stats(timestamp)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_message_sender ON message_stats(sender_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_message_channel ON message_stats(channel)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_command_timestamp ON command_stats(timestamp)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_command_sender ON command_stats(sender_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_command_name ON command_stats(command_name)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_path_timestamp ON path_stats(timestamp)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_path_length ON path_stats(path_length)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_path_sender ON path_stats(sender_id)')
conn.commit()
self.logger.info("Stats tables initialized successfully")
except Exception as e:
self.logger.error(f"Failed to initialize stats tables: {e}")
raise
def record_message(self, message: MeshMessage) -> None:
"""Record a message in the stats database.
Args:
message: The message to record statistics for.
"""
if not self.collect_stats or not self.track_all_messages:
return
try:
# Anonymize user if configured
sender_id = message.sender_id or 'unknown'
if self.anonymize_users and sender_id != 'unknown':
# Create a simple hash-based anonymization
import hashlib
sender_id = f"user_{hashlib.md5(sender_id.encode()).hexdigest()[:8]}"
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO message_stats
(timestamp, sender_id, channel, content, is_dm, hops, snr, rssi, path)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
message.timestamp or int(time.time()),
sender_id,
message.channel,
message.content,
message.is_dm,
message.hops,
message.snr,
message.rssi,
message.path
))
conn.commit()
except Exception as e:
self.logger.error(f"Error recording message stats: {e}")
def record_command(self, message: MeshMessage, command_name: str, response_sent: bool = True) -> None:
"""Record a command execution in the stats database.
Args:
message: The message that triggered the command.
command_name: The name of the command executed.
response_sent: Whether a response was sent back to the user.
"""
if not self.collect_stats or not self.track_command_details:
return
try:
# Anonymize user if configured
sender_id = message.sender_id or 'unknown'
if self.anonymize_users and sender_id != 'unknown':
# Create a simple hash-based anonymization
import hashlib
sender_id = f"user_{hashlib.md5(sender_id.encode()).hexdigest()[:8]}"
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO command_stats
(timestamp, sender_id, command_name, channel, is_dm, response_sent)
VALUES (?, ?, ?, ?, ?, ?)
''', (
message.timestamp or int(time.time()),
sender_id,
command_name,
message.channel,
message.is_dm,
response_sent
))
conn.commit()
except Exception as e:
self.logger.error(f"Error recording command stats: {e}")
def record_path_stats(self, message: MeshMessage) -> None:
"""Record path statistics for longest path tracking.
Args:
message: The message containing path information.
"""
if not self.collect_stats or not self.track_all_messages:
return
# Only record if we have meaningful path data
if not message.hops or message.hops <= 0 or not message.path:
return
# Only record paths that contain actual node IDs (hex characters or comma-separated)
# Skip descriptive paths like "Routed through X hops"
if not self._is_valid_path_format(message.path):
return
try:
# Anonymize user if configured
sender_id = message.sender_id or 'unknown'
if self.anonymize_users and sender_id != 'unknown':
import hashlib
sender_id = f"user_{hashlib.md5(sender_id.encode()).hexdigest()[:8]}"
# Format the path string properly (e.g., "75,24,1d,5f,bd")
path_string = self._format_path_for_display(message.path)
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO path_stats
(timestamp, sender_id, channel, path_length, path_string, hops)
VALUES (?, ?, ?, ?, ?, ?)
''', (
message.timestamp or int(time.time()),
sender_id,
message.channel,
message.hops, # Use hops as path length
path_string,
message.hops
))
conn.commit()
except Exception as e:
self.logger.error(f"Error recording path stats: {e}")
def _is_valid_path_format(self, path: str) -> bool:
"""Check if path contains actual node IDs rather than descriptive text.
Args:
path: The path string to validate.
Returns:
bool: True if the path structure appears valid, False otherwise.
"""
if not path:
return False
# If path contains spaces and common descriptive words, it's likely descriptive text
descriptive_words = ['routed', 'through', 'hops', 'direct', 'unknown', 'path']
path_lower = path.lower()
if any(word in path_lower for word in descriptive_words):
return False
# If path contains only hex characters and commas, it's valid
if all(c in '0123456789abcdefABCDEF,' for c in path):
return True
# If path is a single hex string without separators, it's valid
return bool(all(c in '0123456789abcdefABCDEF' for c in path) and len(path) >= 2)
def _format_path_for_display(self, path: str) -> str:
"""Format path string for display (e.g., '75,24,1d,5f,bd' or '0102,5f7e' for 2-byte hops).
Uses bot.prefix_hex_chars for multi-byte path support; falls back to 2-char (1-byte)
chunks when length is not divisible or for legacy paths.
Args:
path: The raw path string (hex, possibly with commas).
Returns:
str: The formatted path string.
"""
if not path:
return "Direct"
# If path already contains commas, it's likely already formatted
if ',' in path:
return path
# If path contains descriptive text (like "Routed through X hops"),
# extract just the numeric part or return as-is
if ' ' in path and not all(c in '0123456789abcdefABCDEF' for c in path.replace(' ', '')):
# This looks like descriptive text, return as-is
return path
# Use configured hex chars per node for multi-byte path support (2 = 1 byte, 4 = 2 bytes, 6 = 3 bytes)
hex_chars = getattr(self.bot, 'prefix_hex_chars', 2)
if hex_chars <= 0:
hex_chars = 2
# If path is a hex string without separators, chunk by hex_chars per hop
if len(path) > 2 and ',' not in path and all(c in '0123456789abcdefABCDEF' for c in path):
if (len(path) % hex_chars) == 0:
formatted = ','.join([path[i:i + hex_chars] for i in range(0, len(path), hex_chars)])
if formatted:
return formatted
# Legacy fallback: length not divisible by hex_chars, use 2-char (1-byte) chunks
formatted = ','.join([path[i:i + 2] for i in range(0, len(path), 2)])
return formatted if formatted else path
# If it's already a single node ID or short path, return as-is
return path
def get_help_text(self) -> str:
"""Get help text for the stats command.
Returns:
str: The help text for this command.
"""
return self.translate('commands.stats.help')
async def execute(self, message: MeshMessage) -> bool:
"""Execute the stats command.
Handles subcommands for messages, channels, and paths, or shows basic stats
if no subcommand is provided.
Args:
message: The message triggering the command.
Returns:
bool: True if executed successfully, False otherwise.
"""
if not self.stats_enabled:
await self.send_response(message, self.translate('commands.stats.disabled'))
return False
try:
# Perform automatic cleanup if enabled
if self.auto_cleanup:
self.cleanup_old_stats(self.data_retention_days)
# Parse command arguments
content = message.content.strip()
if content.startswith('!'):
content = content[1:].strip()
parts = content.split()
if len(parts) > 1:
subcommand = parts[1].lower()
if subcommand in ['messages', 'message']:
response = await self._get_bot_user_leaderboard()
elif subcommand in ['channels', 'channel']:
response = await self._get_channel_leaderboard()
elif subcommand in ['paths', 'path']:
response = await self._get_path_leaderboard(message)
elif subcommand in ['adverts', 'advert', 'advertisements', 'advertisement']:
# Check for verbose/hash option
show_hashes = len(parts) > 2 and parts[2].lower() in ['hash', 'hashes', 'verbose', 'verb']
response = await self._get_adverts_leaderboard(message, show_hashes=show_hashes)
else:
response = self.translate('commands.stats.unknown_subcommand', subcommand=subcommand)
else:
response = await self._get_basic_stats()
# Record this command execution
self.record_command(message, 'stats', True)
return await self.send_response(message, response)
except Exception as e:
self.logger.error(f"Error executing stats command: {e}")
await self.send_response(message, self.translate('commands.stats.error', error=str(e)))
return False
async def _get_basic_stats(self) -> str:
"""Get basic bot statistics.
Returns:
str: Formatted string containing basic statistics (commands, top user, etc.).
"""
try:
# Get time window (24 hours ago)
now = int(time.time())
day_ago = now - (24 * 60 * 60)
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Bot commands received
cursor.execute('''
SELECT COUNT(*) FROM command_stats
WHERE timestamp >= ?
''', (day_ago,))
commands_received = cursor.fetchone()[0]
# Bot replies sent
cursor.execute('''
SELECT COUNT(*) FROM command_stats
WHERE timestamp >= ? AND response_sent = 1
''', (day_ago,))
bot_replies = cursor.fetchone()[0]
# Top command
cursor.execute('''
SELECT command_name, COUNT(*) as count
FROM command_stats
WHERE timestamp >= ?
GROUP BY command_name
ORDER BY count DESC
LIMIT 1
''', (day_ago,))
top_command_result = cursor.fetchone()
if top_command_result:
top_command = f"{top_command_result[0]} ({top_command_result[1]})"
else:
top_command = self.translate('commands.stats.basic.none')
# Top user
cursor.execute('''
SELECT sender_id, COUNT(*) as count
FROM command_stats
WHERE timestamp >= ?
GROUP BY sender_id
ORDER BY count DESC
LIMIT 1
''', (day_ago,))
top_user_result = cursor.fetchone()
if top_user_result:
top_user = f"{top_user_result[0]} ({top_user_result[1]})"
else:
top_user = self.translate('commands.stats.basic.none')
response = f"""{self.translate('commands.stats.basic.header')}
{self.translate('commands.stats.basic.commands', count=commands_received, replies=bot_replies)}
{self.translate('commands.stats.basic.top_command', command=top_command)}
{self.translate('commands.stats.basic.top_user', user=top_user)}"""
return response
except Exception as e:
self.logger.error(f"Error getting basic stats: {e}")
return self.translate('commands.stats.error', error=str(e))
async def _get_bot_user_leaderboard(self) -> str:
"""Get leaderboard for bot users (people who triggered bot responses).
Returns:
str: Formatted leaderboard string.
"""
try:
# Get time window (24 hours ago)
now = int(time.time())
day_ago = now - (24 * 60 * 60)
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Top bot users (people who triggered commands)
cursor.execute('''
SELECT sender_id, COUNT(*) as count
FROM command_stats
WHERE timestamp >= ?
GROUP BY sender_id
ORDER BY count DESC
LIMIT 5
''', (day_ago,))
top_users = cursor.fetchall()
# Build response
response = self.translate('commands.stats.users.header') + "\n"
if top_users:
for i, (user, count) in enumerate(top_users, 1):
display_user = user[:12] + "..." if len(user) > 15 else user
response += f"{i}. {display_user}: {count}\n"
else:
response += self.translate('commands.stats.users.none') + "\n"
return response
except Exception as e:
self.logger.error(f"Error getting bot user leaderboard: {e}")
return self.translate('commands.stats.error_bot_users', error=str(e))
async def _get_channel_leaderboard(self) -> str:
"""Get leaderboard for channel message activity.
Returns:
str: Formatted leaderboard string.
"""
try:
# Get time window (24 hours ago)
now = int(time.time())
day_ago = now - (24 * 60 * 60)
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Top channels by message count with unique user counts
cursor.execute('''
SELECT channel, COUNT(*) as message_count, COUNT(DISTINCT sender_id) as unique_users
FROM message_stats
WHERE timestamp >= ? AND channel IS NOT NULL
GROUP BY channel
ORDER BY message_count DESC
LIMIT 5
''', (day_ago,))
top_channels = cursor.fetchall()
# Build compact response
response = self.translate('commands.stats.channels.header') + "\n"
if top_channels:
for i, (channel, msg_count, unique_users) in enumerate(top_channels, 1):
display_channel = channel[:12] + "..." if len(channel) > 15 else channel
# Handle singular/plural for messages and users
msg_text = self.translate('commands.stats.channels.msg_singular') if msg_count == 1 else self.translate('commands.stats.channels.msg_plural')
user_text = self.translate('commands.stats.channels.user_singular') if unique_users == 1 else self.translate('commands.stats.channels.user_plural')
response += self.translate('commands.stats.channels.format', rank=i, channel=display_channel, msg_count=msg_count, msg_text=msg_text, user_count=unique_users, user_text=user_text) + "\n"
# Remove trailing newline
response = response.rstrip('\n')
else:
response += self.translate('commands.stats.channels.none')
return response
except Exception as e:
self.logger.error(f"Error getting channel leaderboard: {e}")
return self.translate('commands.stats.error_channels', error=str(e))
async def _get_path_leaderboard(self, message: Optional[MeshMessage] = None) -> str:
"""Get leaderboard for longest paths seen.
Returns:
str: Formatted leaderboard string.
"""
try:
# Get time window (24 hours ago)
now = int(time.time())
day_ago = now - (24 * 60 * 60)
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Top longest paths (one per user, more results with compact format)
cursor.execute('''
SELECT sender_id, path_length, path_string
FROM path_stats p1
WHERE timestamp >= ?
AND path_length = (
SELECT MAX(path_length)
FROM path_stats p2
WHERE p2.sender_id = p1.sender_id
AND p2.timestamp >= ?
)
GROUP BY sender_id
ORDER BY path_length DESC
LIMIT 8
''', (day_ago, day_ago))
longest_paths = cursor.fetchall()
# Build compact response with length checking
response = ""
max_length = self.get_max_message_length(message) if message else 130
if longest_paths:
for i, (sender, path_len, path_str) in enumerate(longest_paths, 1):
# Truncate sender name to fit more data
display_sender = sender[:8] + "..." if len(sender) > 11 else sender
# Full line: rank + sender + ": " + path (same as format template)
full_line = self.translate('commands.stats.paths.format', rank=i, sender=display_sender, path=path_str) + "\n"
# When a single path line would exceed the message limit, use synopsis (rank + sender + hop count only)
if len(full_line) > max_length:
new_line = self.translate('commands.stats.paths.synopsis', rank=i, sender=display_sender, hops=path_len) + "\n"
else:
new_line = full_line
# Check if adding this line would exceed the limit
if len(response + new_line) > max_length:
break
response += new_line
else:
response = self.translate('commands.stats.paths.none') + "\n"
return response
except Exception as e:
self.logger.error(f"Error getting path leaderboard: {e}")
return self.translate('commands.stats.error_paths', error=str(e))
async def _get_adverts_leaderboard(self, message: Optional[MeshMessage] = None, show_hashes: bool = False) -> str:
"""Get leaderboard for nodes with most unique advert packets in last 24 hours.
Args:
message: Optional message for dynamic length calculation.
show_hashes: If True, include packet hashes in output.
Returns:
str: Formatted leaderboard string.
"""
try:
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Check if daily_stats table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='daily_stats'")
has_daily_stats = cursor.fetchone() is not None
if has_daily_stats:
if show_hashes:
# Query with packet hashes for validation
# First get the top nodes, then get their hashes
cursor.execute('''
SELECT
c.public_key,
c.name,
COALESCE(
(SELECT advert_count FROM daily_stats
WHERE date = DATE(c.last_advert_timestamp)
AND public_key = c.public_key), 0
) as unique_adverts
FROM complete_contact_tracking c
WHERE c.last_advert_timestamp >= datetime('now', '-24 hours')
GROUP BY c.public_key, c.name
HAVING unique_adverts > 0
ORDER BY unique_adverts DESC
LIMIT 20
''')
top_nodes = cursor.fetchall()
# Now get packet hashes for each node
top_adverts = []
for public_key, name, count in top_nodes:
# Get packet hashes for this node in the last 24 hours
cursor.execute('''
SELECT packet_hash
FROM unique_advert_packets
WHERE public_key = ?
AND first_seen >= datetime('now', '-24 hours')
ORDER BY first_seen
''', (public_key,))
hash_rows = cursor.fetchall()
packet_hashes = ', '.join([row[0] for row in hash_rows]) if hash_rows else None
top_adverts.append((public_key, name, count, packet_hashes))
else:
# Query nodes that advertised in the last 24 hours
# Get advert_count from daily_stats for the day of last_advert_timestamp
# This gives us the count of unique adverts for that day
cursor.execute('''
SELECT
c.public_key,
c.name,
COALESCE(
(SELECT advert_count FROM daily_stats
WHERE date = DATE(c.last_advert_timestamp)
AND public_key = c.public_key), 0
) as unique_adverts
FROM complete_contact_tracking c
WHERE c.last_advert_timestamp >= datetime('now', '-24 hours')
GROUP BY c.public_key, c.name
HAVING unique_adverts > 0
ORDER BY unique_adverts DESC
LIMIT 20
''')
top_adverts = cursor.fetchall()
else:
# Fallback: use advert_count from complete_contact_tracking
# This is less accurate but works if daily_stats doesn't exist
if show_hashes:
# Get nodes first
cursor.execute('''
SELECT public_key, name, advert_count as unique_adverts
FROM complete_contact_tracking
WHERE last_advert_timestamp >= datetime('now', '-24 hours')
ORDER BY advert_count DESC
LIMIT 20
''')
top_nodes = cursor.fetchall()
# Get packet hashes for each node
top_adverts = []
for public_key, name, count in top_nodes:
cursor.execute('''
SELECT packet_hash
FROM unique_advert_packets
WHERE public_key = ?
AND first_seen >= datetime('now', '-24 hours')
ORDER BY first_seen
''', (public_key,))
hash_rows = cursor.fetchall()
packet_hashes = ', '.join([row[0] for row in hash_rows]) if hash_rows else None
top_adverts.append((public_key, name, count, packet_hashes))
else:
cursor.execute('''
SELECT public_key, name, advert_count as unique_adverts
FROM complete_contact_tracking
WHERE last_advert_timestamp >= datetime('now', '-24 hours')
ORDER BY advert_count DESC
LIMIT 20
''')
top_adverts = cursor.fetchall()
# Build compact response with length checking
response = self.translate('commands.stats.adverts.header') + "\n"
max_length = self.get_max_message_length(message) if message else 130
if top_adverts:
for i, row in enumerate(top_adverts, 1):
if show_hashes:
public_key, name, count, packet_hashes = row
else:
public_key, name, count = row[0], row[1], row[2]
packet_hashes = None
# Truncate name if needed
display_name = name[:15] + "..." if len(name) > 18 else name
# Format: "1. NodeName: 42 adverts"
advert_text = self.translate('commands.stats.adverts.advert_singular') if count == 1 else self.translate('commands.stats.adverts.advert_plural')
if show_hashes and packet_hashes:
# Format with packet hashes: "1. NodeName: 42 adverts\n Hashes: abc123, def456, ..."
main_line = self.translate('commands.stats.adverts.format',
rank=i,
name=display_name,
count=count,
advert_text=advert_text)
# Split packet hashes and format them
hash_list = [h.strip() for h in packet_hashes.split(',') if h.strip()]
# Show first few hashes (truncate if too many)
if len(hash_list) > 10:
hash_display = ', '.join(hash_list[:10]) + f" ... ({len(hash_list)} total)"
else:
hash_display = ', '.join(hash_list)
new_line = f"{main_line}\n {self.translate('commands.stats.adverts.hashes_label', hashes=hash_display)}"
else:
new_line = self.translate('commands.stats.adverts.format',
rank=i,
name=display_name,
count=count,
advert_text=advert_text) + "\n"
# Check if adding this line would exceed the limit
if len(response + new_line.rstrip('\n')) > max_length:
break
response += new_line
# Remove trailing newline
response = response.rstrip('\n')
else:
response += self.translate('commands.stats.adverts.none')
return response
except Exception as e:
self.logger.error(f"Error getting adverts leaderboard: {e}")
return self.translate('commands.stats.error_adverts', error=str(e))
# Rows timestamped beyond this margin ahead of now came from a bad clock on
# the sending node, not from the future. Without an upper bound they are
# never older than the cutoff, so retention never removes them: observed in
# the wild dated 2103-08-15, which also stretches any chart's axis to match.
FUTURE_TIMESTAMP_GRACE_SECONDS = 24 * 60 * 60
def cleanup_old_stats(self, days_to_keep: int = 7) -> None:
"""Clean up old stats data to prevent database bloat.
Deletes rows older than the retention cutoff *and* rows dated
implausibly far in the future.
Args:
days_to_keep: Number of days of data to retain.
"""
try:
now = int(time.time())
cutoff_time = now - (days_to_keep * 24 * 60 * 60)
future_cutoff = now + self.FUTURE_TIMESTAMP_GRACE_SECONDS
deleted = {}
for table in ('message_stats', 'command_stats', 'path_stats'):
deleted[table] = (
self.bot.db_manager.delete_timestamp_rows_in_chunks(
table,
'timestamp',
cutoff_time,
future_cutoff=future_cutoff,
progress_label=table.replace('_', ' '),
)
)
total_deleted = sum(deleted.values())
if total_deleted > 0:
self.logger.info(
f"Cleaned up {total_deleted} old stats entries "
f"({deleted['message_stats']} messages, "
f"{deleted['command_stats']} commands, "
f"{deleted['path_stats']} paths)"
)
except Exception as e:
self.logger.error(f"Error cleaning up old stats: {e}")
def get_stats_summary(self) -> dict[str, Any]:
"""Get a summary of all stats data.
Returns:
Dict[str, Any]: Dictionary containing summary statistics.
"""
try:
with self.bot.db_manager.connection() as conn:
cursor = conn.cursor()
# Total messages
cursor.execute('SELECT COUNT(*) FROM message_stats')
total_messages = cursor.fetchone()[0]
# Total commands
cursor.execute('SELECT COUNT(*) FROM command_stats')
total_commands = cursor.fetchone()[0]
# Unique users
cursor.execute('SELECT COUNT(DISTINCT sender_id) FROM message_stats')
unique_users = cursor.fetchone()[0]
# Unique channels
cursor.execute('SELECT COUNT(DISTINCT channel) FROM message_stats WHERE channel IS NOT NULL')
unique_channels = cursor.fetchone()[0]
return {
'total_messages': total_messages,
'total_commands': total_commands,
'unique_users': unique_users,
'unique_channels': unique_channels
}
except Exception as e:
self.logger.error(f"Error getting stats summary: {e}")
return {}
Reference in New Issue View Git Blame Copy Permalink