diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index a35af1b..bf266b2 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -71,7 +71,7 @@ from modules.scheduled_message_admin import ( ) from modules.security_utils import ( validate_external_url, - validate_sql_identifier, + validate_sql_identifier, # noqa: F401 kept for modules.web_viewer.app.validate_sql_identifier ) from modules.settings_schema import ( build_plugin_settings_view, @@ -84,7 +84,7 @@ from modules.web_viewer.dashboard_stats import ( SERIES_METRICS, TOP_KINDS, DashboardStatsService, - humanize_span, + humanize_span, # noqa: F401 kept for modules.web_viewer.app.humanize_span ) # RFC 8594 Sunset date advertised on the deprecated /api/stats endpoint. @@ -163,6 +163,7 @@ from modules.repeater_manager import RepeaterManager, validate_repeater_tables from modules.security_utils import SafeUrlPolicy, create_safe_requests_session, safe_requests_request # noqa: F401 from modules.utils import resolve_path from modules.web_viewer.config_panels import CONFIG_PANELS, PANEL_CATEGORIES +from modules.web_viewer.database_admin import DatabaseAdminMixin from modules.web_viewer.feeds import ( # noqa: F401 _read_limited_requests_response and _validate_feed_interval re-exported FeedSubscriptionsMixin, _read_limited_requests_response, @@ -174,9 +175,10 @@ from modules.web_viewer.mesh_evidence import ( # noqa: F401 NeighborEvidenceKe NeighborEvidenceKeys, ) from modules.web_viewer.multibyte_rollout import MultibyteRolloutMixin +from modules.web_viewer.tracking import ContactTrackingMixin -class BotDataViewer(MeshEvidenceMixin, MultibyteRolloutMixin, FeedSubscriptionsMixin): +class BotDataViewer(DatabaseAdminMixin, ContactTrackingMixin, MeshEvidenceMixin, MultibyteRolloutMixin, FeedSubscriptionsMixin): """Complete web interface using Flask-SocketIO 5.x best practices""" # Whitelist of allowed tables for security @@ -5451,1325 +5453,6 @@ class BotDataViewer(MeshEvidenceMixin, MultibyteRolloutMixin, FeedSubscriptionsM except Exception as e: self.logger.error(f"Error cleaning up old packet stream data: {e}", exc_info=True) - def _get_database_stats(self, top_users_window='all', top_commands_window='all', - top_paths_window='all', top_channels_window='all'): - """Get comprehensive database statistics for dashboard""" - try: - with self._db_connection() as conn: - cursor = conn.cursor() - - # Get all available tables - cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") - tables = [row[0] for row in cursor.fetchall()] - - # Filter tables by ALLOWED_TABLES whitelist for security - tables = [t for t in tables if t in self.ALLOWED_TABLES] - - with self._clients_lock: - client_count = len(self.connected_clients) - - stats = { - 'timestamp': time.time(), - 'connected_clients': client_count, - 'tables': tables - } - - # Contact and tracking statistics - if 'complete_contact_tracking' in tables: - cursor.execute("SELECT COUNT(*) FROM complete_contact_tracking") - stats['total_contacts'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(*) FROM complete_contact_tracking - WHERE last_heard > datetime('now', 'localtime', '-24 hours') - """) - stats['contacts_24h'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(*) FROM complete_contact_tracking - WHERE last_heard > datetime('now', 'localtime', '-7 days') - """) - stats['contacts_7d'] = cursor.fetchone()[0] - - # Contacts heard in 7d with multibyte path evidence. Scope observed_paths to 7d so - # the pie chart matches "last 7 days" (lifetime paths + stale out_bytes_per_hop - # otherwise inflated the percentage). - stats['contacts_7d_multibyte_path'] = 0 - chunk_buckets = self._bucket_hop_chunks(set()) - mb_advert_pks: set[str] = set() - if 'observed_paths' in tables: - try: - chunk_buckets = self._bucket_hop_chunks( - self._get_cached_contact_multibyte_hop_chunks( - cursor, recent_days=7 - ) - ) - # Use date() — julianday(iso8601) often returns NULL for Python isoformat() strings - cursor.execute( - """ - SELECT DISTINCT public_key FROM observed_paths - WHERE packet_type = 'advert' AND public_key IS NOT NULL - AND bytes_per_hop IN (2, 3) - AND date(last_seen) >= date('now', 'localtime', '-7 days') - """ - ) - mb_advert_pks = { - row["public_key"] for row in cursor.fetchall() if row["public_key"] - } - except Exception as e: - self.logger.debug(f"Could not load multibyte path sets for 7d stats: {e}") - try: - cursor.execute( - """ - SELECT public_key, role, out_bytes_per_hop - FROM complete_contact_tracking - WHERE last_heard > datetime('now', 'localtime', '-7 days') - """ - ) - mb_7d = 0 - for row in cursor.fetchall(): - if self._contact_has_multibyte_path_evidence( - row["public_key"], - row["role"], - row["out_bytes_per_hop"], - mb_advert_pks, - chunk_buckets, - ): - mb_7d += 1 - stats['contacts_7d_multibyte_path'] = mb_7d - except Exception as e: - self.logger.debug(f"Could not compute contacts_7d_multibyte_path: {e}") - - cursor.execute(""" - SELECT COUNT(*) FROM complete_contact_tracking - WHERE is_currently_tracked = 1 - """) - stats['tracked_contacts'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT AVG(hop_count) FROM complete_contact_tracking - WHERE hop_count IS NOT NULL - """) - avg_hops = cursor.fetchone()[0] - stats['avg_hop_count'] = round(avg_hops, 1) if avg_hops else 0 - - cursor.execute(""" - SELECT MAX(hop_count) FROM complete_contact_tracking - WHERE hop_count IS NOT NULL - """) - stats['max_hop_count'] = cursor.fetchone()[0] or 0 - - cursor.execute(""" - SELECT COUNT(DISTINCT role) FROM complete_contact_tracking - WHERE role IS NOT NULL - """) - stats['unique_roles'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(DISTINCT device_type) FROM complete_contact_tracking - WHERE device_type IS NOT NULL - """) - stats['unique_device_types'] = cursor.fetchone()[0] - - # Incoming packets: multibyte path share over whatever packet_stream - # actually retains. The key names still say 7d for compatibility, - # but the window is reported honestly alongside them — - # packet_stream is pruned at 3 days, so the old label was never true. - stats['incoming_packets_7d'] = 0 - stats['incoming_packets_7d_multibyte_path'] = 0 - if 'packet_stream' in tables: - try: - cutoff_ts = time.time() - 7 * 86400 - cursor.execute( - """ - SELECT COUNT(*), - SUM(CASE WHEN bytes_per_hop IN (2, 3) THEN 1 ELSE 0 END), - MIN(timestamp) - FROM packet_stream - WHERE type = ? AND timestamp > ? AND route_type_name IS NOT NULL - """, - ("packet", cutoff_ts), - ) - row = cursor.fetchone() - stats['incoming_packets_7d'] = row[0] or 0 - stats['incoming_packets_7d_multibyte_path'] = row[1] or 0 - stats['incoming_packets_from'] = row[2] - stats['incoming_packets_window_label'] = humanize_span( - time.time() - row[2] if row[2] else None - ) - except Exception as e: - self.logger.debug(f"Could not compute incoming packet multibyte stats: {e}") - - # Advertisement statistics using daily tracking table - if 'daily_stats' in tables: - # Total advertisements (all time) - cursor.execute(""" - SELECT SUM(advert_count) FROM daily_stats - """) - total_adverts = cursor.fetchone()[0] - stats['total_advertisements'] = total_adverts or 0 - - # 24h advertisements - cursor.execute(""" - SELECT SUM(advert_count) FROM daily_stats - WHERE date = date('now', 'localtime') - """) - stats['advertisements_24h'] = cursor.fetchone()[0] or 0 - - # 7d advertisements (last 7 days, excluding today) - cursor.execute(""" - SELECT SUM(advert_count) FROM daily_stats - WHERE date >= date('now', 'localtime', '-7 days') AND date < date('now', 'localtime') - """) - stats['advertisements_7d'] = cursor.fetchone()[0] or 0 - - # Nodes per day statistics - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date = date('now', 'localtime') - """) - stats['nodes_24h'] = cursor.fetchone()[0] or 0 - - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date >= date('now', 'localtime', '-6 days') - """) - stats['nodes_7d'] = cursor.fetchone()[0] or 0 - - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - """) - stats['nodes_all'] = cursor.fetchone()[0] or 0 - else: - # Fallback to old method if daily table doesn't exist yet - if 'complete_contact_tracking' in tables: - cursor.execute(""" - SELECT SUM(advert_count) FROM complete_contact_tracking - """) - total_adverts = cursor.fetchone()[0] - stats['total_advertisements'] = total_adverts or 0 - - cursor.execute(""" - SELECT SUM(advert_count) FROM complete_contact_tracking - WHERE last_heard > datetime('now', 'localtime', '-24 hours') - """) - stats['advertisements_24h'] = cursor.fetchone()[0] or 0 - - cursor.execute(""" - SELECT SUM(advert_count) FROM complete_contact_tracking - WHERE last_heard > datetime('now', 'localtime', '-7 days') - """) - stats['advertisements_7d'] = cursor.fetchone()[0] or 0 - - # Repeater contacts (if exists) - if 'repeater_contacts' in tables: - cursor.execute("SELECT COUNT(*) FROM repeater_contacts") - stats['repeater_contacts'] = cursor.fetchone()[0] - - cursor.execute("SELECT COUNT(*) FROM repeater_contacts WHERE is_active = 1") - stats['active_repeater_contacts'] = cursor.fetchone()[0] - - # Cache statistics - cache_tables = [t for t in tables if 'cache' in t] - stats['cache_tables'] = cache_tables - stats['total_cache_entries'] = 0 - stats['active_cache_entries'] = 0 - - for table in cache_tables: - try: - validate_sql_identifier(table) - except ValueError: - self.logger.warning(f"Rejecting invalid table name: {table!r}") - raise - cursor.execute(f"SELECT COUNT(*) FROM {table}") - count = cursor.fetchone()[0] - stats['total_cache_entries'] += count - stats[f'{table}_count'] = count - - # Get active entries (not expired) - cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE expires_at > datetime('now')") - active_count = cursor.fetchone()[0] - stats['active_cache_entries'] += active_count - stats[f'{table}_active'] = active_count - - # Message and command statistics (if stats tables exist) - if 'message_stats' in tables: - cursor.execute("SELECT COUNT(*) FROM message_stats") - stats['total_messages'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(*) FROM message_stats - WHERE timestamp > strftime('%s', 'now', '-24 hours') - """) - stats['messages_24h'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(DISTINCT sender_id) FROM message_stats - WHERE timestamp > strftime('%s', 'now', '-24 hours') - """) - stats['unique_senders_24h'] = cursor.fetchone()[0] - - # Total unique users and channels - cursor.execute("SELECT COUNT(DISTINCT sender_id) FROM message_stats") - stats['unique_users_total'] = cursor.fetchone()[0] - - cursor.execute("SELECT COUNT(DISTINCT channel) FROM message_stats WHERE channel IS NOT NULL") - stats['unique_channels_total'] = cursor.fetchone()[0] - - # Top users (most frequent message senders) - filter by time window - if top_users_window == '24h': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-24 hours')" - elif top_users_window == '7d': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-7 days')" - elif top_users_window == '30d': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-30 days')" - else: # 'all' - time_filter = "" - - query = f""" - SELECT sender_id, COUNT(*) as count - FROM message_stats - {time_filter} - GROUP BY sender_id - ORDER BY count DESC - LIMIT 15 - """ - cursor.execute(query) - stats['top_users'] = [{'user': row[0], 'count': row[1]} for row in cursor.fetchall()] - - # Top channels by message count - filter by time window - if top_channels_window == '24h': - time_filter = "AND timestamp > strftime('%s', 'now', '-24 hours')" - elif top_channels_window == '7d': - time_filter = "AND timestamp > strftime('%s', 'now', '-7 days')" - elif top_channels_window == '30d': - time_filter = "AND timestamp > strftime('%s', 'now', '-30 days')" - else: # 'all' - time_filter = "" - - query = f""" - SELECT channel, COUNT(*) as message_count, COUNT(DISTINCT sender_id) as unique_users - FROM message_stats - WHERE channel IS NOT NULL {time_filter} - GROUP BY channel - ORDER BY message_count DESC - LIMIT 10 - """ - cursor.execute(query) - stats['top_channels'] = [ - {'channel': row[0], 'messages': row[1], 'users': row[2]} - for row in cursor.fetchall() - ] - - if 'command_stats' in tables: - cursor.execute("SELECT COUNT(*) FROM command_stats") - stats['total_commands'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-24 hours') - """) - stats['commands_24h'] = cursor.fetchone()[0] - - # Top commands - filter by time window - if top_commands_window == '24h': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-24 hours')" - elif top_commands_window == '7d': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-7 days')" - elif top_commands_window == '30d': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-30 days')" - else: # 'all' - time_filter = "" - - query = f""" - SELECT command_name, COUNT(*) as count - FROM command_stats - {time_filter} - GROUP BY command_name - ORDER BY count DESC - LIMIT 15 - """ - cursor.execute(query) - stats['top_commands'] = [{'command': row[0], 'count': row[1]} for row in cursor.fetchall()] - - # Bot reply rates (commands that got responses) - calculate for different time windows - # 24 hour reply rate - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-24 hours') AND response_sent = 1 - """) - replied_24h = cursor.fetchone()[0] - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-24 hours') - """) - total_24h = cursor.fetchone()[0] - if total_24h > 0: - stats['bot_reply_rate_24h'] = round((replied_24h / total_24h) * 100, 1) - else: - stats['bot_reply_rate_24h'] = 0 - - # 7 day reply rate - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-7 days') AND response_sent = 1 - """) - replied_7d = cursor.fetchone()[0] - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-7 days') - """) - total_7d = cursor.fetchone()[0] - if total_7d > 0: - stats['bot_reply_rate_7d'] = round((replied_7d / total_7d) * 100, 1) - else: - stats['bot_reply_rate_7d'] = 0 - - # 30 day reply rate - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-30 days') AND response_sent = 1 - """) - replied_30d = cursor.fetchone()[0] - cursor.execute(""" - SELECT COUNT(*) FROM command_stats - WHERE timestamp > strftime('%s', 'now', '-30 days') - """) - total_30d = cursor.fetchone()[0] - if total_30d > 0: - stats['bot_reply_rate_30d'] = round((replied_30d / total_30d) * 100, 1) - else: - stats['bot_reply_rate_30d'] = 0 - - # Path statistics (if path_stats table exists) - if 'path_stats' in tables: - cursor.execute(""" - SELECT sender_id, path_length, path_string, timestamp - FROM path_stats - ORDER BY path_length DESC - LIMIT 1 - """) - longest_path = cursor.fetchone() - if longest_path: - stats['longest_path'] = { - 'user': longest_path[0], - 'path_length': longest_path[1], - 'path_string': longest_path[2], - 'timestamp': longest_path[3] - } - - # Top paths (longest paths) - filter by time window - if top_paths_window == '24h': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-24 hours')" - elif top_paths_window == '7d': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-7 days')" - elif top_paths_window == '30d': - time_filter = "WHERE timestamp > strftime('%s', 'now', '-30 days')" - else: # 'all' - time_filter = "" - - query = f""" - SELECT sender_id, path_length, path_string, timestamp - FROM path_stats - {time_filter} - ORDER BY path_length DESC - LIMIT 5 - """ - cursor.execute(query) - stats['top_paths'] = [ - { - 'user': row[0], - 'path_length': row[1], - 'path_string': row[2], - 'timestamp': row[3] - } - for row in cursor.fetchall() - ] - - # Network health metrics - if 'complete_contact_tracking' in tables: - cursor.execute(""" - SELECT AVG(snr) FROM complete_contact_tracking - WHERE snr IS NOT NULL AND last_heard > datetime('now', 'localtime', '-24 hours') - """) - avg_snr = cursor.fetchone()[0] - stats['avg_snr_24h'] = round(avg_snr, 1) if avg_snr else 0 - - cursor.execute(""" - SELECT AVG(signal_strength) FROM complete_contact_tracking - WHERE signal_strength IS NOT NULL AND last_heard > datetime('now', 'localtime', '-24 hours') - """) - avg_signal = cursor.fetchone()[0] - stats['avg_signal_strength_24h'] = round(avg_signal, 1) if avg_signal else 0 - - # Geographic distribution - only count currently tracked contacts heard in the last 30 days - # Normalize country names to avoid duplicates (e.g., "United States" vs "United States of America") - if 'complete_contact_tracking' in tables: - cursor.execute(""" - SELECT COUNT(DISTINCT - CASE - WHEN country IN ('United States', 'United States of America', 'US', 'USA') - THEN 'United States' - ELSE country - END - ) FROM complete_contact_tracking - WHERE country IS NOT NULL AND country != '' - AND last_heard > datetime('now', 'localtime', '-30 days') - AND is_currently_tracked = 1 - """) - stats['countries'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(DISTINCT state) FROM complete_contact_tracking - WHERE state IS NOT NULL AND state != '' - AND last_heard > datetime('now', 'localtime', '-30 days') - AND is_currently_tracked = 1 - """) - stats['states'] = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(DISTINCT city) FROM complete_contact_tracking - WHERE city IS NOT NULL AND city != '' - AND last_heard > datetime('now', 'localtime', '-30 days') - AND is_currently_tracked = 1 - """) - stats['cities'] = cursor.fetchone()[0] - - return stats - - except Exception as e: - self.logger.error(f"Error getting database stats: {e}") - return {'error': str(e)} - - def _get_database_info(self): - """Get comprehensive database information for database page""" - try: - with self._db_connection() as conn: - cursor = conn.cursor() - - # Get all tables - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") - table_names = [row[0] for row in cursor.fetchall()] - - # Filter tables by ALLOWED_TABLES whitelist for security - table_names = [ - name for name in table_names - if name in self.ALLOWED_TABLES - ] - - # Get table information - tables = [] - total_records = 0 - - for table_name in table_names: - try: - # Get record count - cursor.execute(f"SELECT COUNT(*) FROM {table_name}") - record_count = cursor.fetchone()[0] - total_records += record_count - - # Get table size (approximate) - cursor.execute(f"PRAGMA table_info({table_name})") - columns = cursor.fetchall() - - # Estimate size (rough calculation) - estimated_size = record_count * len(columns) * 50 # Rough estimate - size_str = f"{estimated_size:,} bytes" if estimated_size < 1024 else f"{estimated_size/1024:.1f} KB" - - # Get table description based on name - description = self._get_table_description(table_name) - - tables.append({ - 'name': table_name, - 'record_count': record_count, - 'size': size_str, - 'description': description - }) - - except Exception as e: - self.logger.debug(f"Error getting info for table {table_name}: {e}") - tables.append({ - 'name': table_name, - 'record_count': 0, - 'size': 'Unknown', - 'description': 'Error reading table' - }) - - # Get database file size - import os - try: - db_size_bytes = os.path.getsize(self.db_path) - if db_size_bytes < 1024: - db_size = f"{db_size_bytes} bytes" - elif db_size_bytes < 1024 * 1024: - db_size = f"{db_size_bytes/1024:.1f} KB" - else: - db_size = f"{db_size_bytes/(1024*1024):.1f} MB" - except: - db_size = "Unknown" - - return { - 'total_tables': len(table_names), - 'total_records': total_records, - 'last_updated': time.strftime('%Y-%m-%d %H:%M:%S'), - 'db_size': db_size, - 'tables': tables - } - - except Exception as e: - self.logger.error(f"Error getting database info: {e}") - return { - 'total_tables': 0, - 'total_records': 0, - 'last_updated': 'Error', - 'db_size': 'Unknown', - 'tables': [] - } - - def _is_safe_table_name(self, table_name: str) -> bool: - """Check if table name is in the ALLOWED_TABLES whitelist. - - Args: - table_name: The table name to validate - - Returns: - True if the table is in the allowed whitelist, False otherwise - """ - if not table_name or not isinstance(table_name, str): - return False - return table_name in self.ALLOWED_TABLES - - def _get_table_description(self, table_name): - """Get human-readable description for table""" - descriptions = { - 'packet_stream': 'Real-time packet and command data stream', - 'complete_contact_tracking': 'Contact tracking and device information', - 'repeater_contacts': 'Repeater contact management', - 'message_stats': 'Message statistics and analytics', - 'command_stats': 'Command execution statistics', - 'path_stats': 'Network path statistics', - 'geocoding_cache': 'Geocoding service cache', - 'generic_cache': 'General purpose cache storage' - } - return descriptions.get(table_name, 'Database table') - - def _optimize_database(self): - """Optimize database using VACUUM, ANALYZE, and REINDEX""" - try: - with self._db_connection() as conn: - cursor = conn.cursor() - - # Get initial database size - import os - initial_size = os.path.getsize(self.db_path) - - # Perform VACUUM to reclaim unused space - self.logger.info("Starting database VACUUM...") - cursor.execute("VACUUM") - vacuum_size = os.path.getsize(self.db_path) - vacuum_saved = initial_size - vacuum_size - - # Perform ANALYZE to update table statistics - self.logger.info("Starting database ANALYZE...") - cursor.execute("ANALYZE") - - # Get all tables for REINDEX - cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") - tables = [row[0] for row in cursor.fetchall()] - - # Filter tables by ALLOWED_TABLES whitelist for security - tables = [t for t in tables if t in self.ALLOWED_TABLES] - - # Perform REINDEX on all tables - self.logger.info("Starting database REINDEX...") - reindexed_tables = [] - for table in tables: - try: - cursor.execute(f"REINDEX {table}") - reindexed_tables.append(table) - except Exception as e: - self.logger.debug(f"Could not reindex table {table}: {e}") - - # Get final database size - final_size = os.path.getsize(self.db_path) - total_saved = initial_size - final_size - - # Format size information - def format_size(size_bytes): - if size_bytes < 1024: - return f"{size_bytes} bytes" - elif size_bytes < 1024 * 1024: - return f"{size_bytes/1024:.1f} KB" - else: - return f"{size_bytes/(1024*1024):.1f} MB" - - return { - 'success': True, - 'vacuum_result': f"VACUUM completed - saved {format_size(vacuum_saved)}", - 'analyze_result': f"ANALYZE completed - updated statistics for {len(tables)} tables", - 'reindex_result': f"REINDEX completed - rebuilt indexes for {len(reindexed_tables)} tables", - 'initial_size': format_size(initial_size), - 'final_size': format_size(final_size), - 'total_saved': format_size(total_saved), - 'tables_processed': len(tables), - 'tables_reindexed': len(reindexed_tables) - } - - except Exception as e: - self.logger.error(f"Error optimizing database: {e}") - return { - 'success': False, - 'error': str(e) - } - - - def _get_contact_detail(self, public_key: str) -> dict: - """Per-contact detail loaded on demand by the contacts UI modals. - - Returns the recent advert paths (same shape as the old list ``all_paths``) and the raw - advertisement data. Both are excluded from the /api/contacts list payload — they were - ~85% of its size and are only needed when a single contact is opened. - """ - try: - with self._db_connection() as conn: - cursor = conn.cursor() - - # Recent advert paths (most-recent 50). The idx_observed_paths_advert_pk_seen covering - # index serves WHERE public_key=? AND packet_type='advert' ORDER BY last_seen DESC directly. - all_paths: list[dict[str, Any]] = [] - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='observed_paths'") - if cursor.fetchone(): - cursor.execute(""" - SELECT path_hex, path_length, bytes_per_hop, observation_count, last_seen - FROM observed_paths - WHERE packet_type = 'advert' AND public_key = ? - ORDER BY last_seen DESC - LIMIT 50 - """, (public_key,)) - for prow in cursor.fetchall(): - if not prow['path_hex']: - continue - bph = None - if prow['bytes_per_hop'] is not None: - try: - bph = int(prow['bytes_per_hop']) - if bph not in (1, 2, 3): - bph = 1 - except (TypeError, ValueError): - bph = 1 - all_paths.append({ - 'path_hex': prow['path_hex'], - 'path_length': int(prow['path_length']) if prow['path_length'] is not None else 0, - 'bytes_per_hop': bph, - 'observation_count': int(prow['observation_count']) if prow['observation_count'] is not None else 1, - 'last_seen': prow['last_seen'] if prow['last_seen'] is not None else None, - }) - - # Raw advertisement data for the "Advertisement Data" modal. - raw_advert_data = None - raw_advert_data_parsed = None - cursor.execute(""" - SELECT raw_advert_data FROM complete_contact_tracking - WHERE public_key = ? ORDER BY last_heard DESC LIMIT 1 - """, (public_key,)) - rad_row = cursor.fetchone() - if rad_row and rad_row['raw_advert_data']: - raw_advert_data = rad_row['raw_advert_data'] - try: - import json - raw_advert_data_parsed = json.loads(raw_advert_data) - except Exception: - raw_advert_data_parsed = None - - return { - 'all_paths': all_paths, - 'raw_advert_data': raw_advert_data, - 'raw_advert_data_parsed': raw_advert_data_parsed, - } - except Exception as e: - self.logger.error(f"Error getting contact detail: {e}") - return {'error': str(e)} - - def _get_tracking_data( - self, - since='30d', - include_detail=False, - page: int | None = None, - page_size: int | None = None, - search: str = '', - path_bytes: str = '', - device_role: str = '', - hop_filter: str = '', - location_filter: str = '', - starred: str = '', - sort: str = 'last_seen', - direction: str = 'desc', - ): - """Get contact tracking data. since: 24h, 7d, 30d, 90d, or all (heard in that window). - - include_detail=False (the interactive /api/contacts list) omits per-contact ``all_paths`` - and ``raw_advert_data`` to keep the payload small; the UI fetches those on demand via - /api/contact-detail. The interactive route supplies ``page`` and ``page_size`` so path - enrichment is limited to visible contacts. include_detail=True (the export endpoint) - keeps the legacy full-result behavior and full fields. - """ - try: - with self._db_connection() as conn: - cursor = conn.cursor() - - # Get bot location from config - bot_lat = self.config.getfloat('Bot', 'bot_latitude', fallback=None) - bot_lon = self.config.getfloat('Bot', 'bot_longitude', fallback=None) - - # Filter by last_heard (default: last 30 days). last_heard is stored as ISO-text - # datetime in LOCAL time (e.g. '2026-06-16 09:03:49.606966', written by datetime.now()), - # so the cutoff must also be local: datetime('now', 'localtime', ...). Using bare - # datetime('now', ...) computes the cutoff in UTC and shaves the local UTC offset off - # the window (e.g. a "24h" filter only returns ~17h of data in US/Pacific). - datetime_offsets = { - '24h': "'-24 hours'", - '7d': "'-7 days'", - '30d': "'-30 days'", - '90d': "'-90 days'", - } - where_parts = [] - where_params: list[Any] = [] - # A node can have more than one observed advert path. Treat its byte class as - # the widest path encoding seen for it, with the contact's current out-path as a - # fallback for databases that have not retained an observed path yet. This gives - # the list one stable, sortable value instead of placing the same node in several - # byte buckets. Only count rows with a known 1/2/3 encoding so NULL/invalid - # observations do not collapse to "1-byte" and block the out-path fallback. - path_bytes_expression = """COALESCE(( - SELECT MAX(op.bytes_per_hop) - FROM observed_paths op - WHERE op.public_key = c.public_key - AND op.packet_type = 'advert' - AND op.path_hex IS NOT NULL AND op.path_hex != '' - AND op.bytes_per_hop IN (1, 2, 3) - ), CASE WHEN c.out_bytes_per_hop IN (1, 2, 3) - THEN c.out_bytes_per_hop ELSE 0 END)""" - if since in datetime_offsets: - where_parts.append( - f"c.last_heard >= datetime('now', 'localtime', {datetime_offsets[since]})" - ) - - search = (search or '').strip().lower()[:100] - if search and not include_detail: - # Match the former client-side behavior: public keys are prefix-only, - # while names, roles, device types, and locations match anywhere. - escaped = search.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_') - where_parts.append( - "(" - "LOWER(COALESCE(c.public_key, '')) LIKE ? ESCAPE '\\' OR " - "LOWER(COALESCE(c.name, '')) LIKE ? ESCAPE '\\' OR " - "LOWER(COALESCE(c.role, '')) LIKE ? ESCAPE '\\' OR " - "LOWER(COALESCE(c.device_type, '')) LIKE ? ESCAPE '\\' OR " - "LOWER(COALESCE(c.city, '')) LIKE ? ESCAPE '\\' OR " - "LOWER(COALESCE(c.state, '')) LIKE ? ESCAPE '\\' OR " - "LOWER(COALESCE(c.country, '')) LIKE ? ESCAPE '\\'" - ")" - ) - where_params.extend([f'{escaped}%'] + [f'%{escaped}%'] * 6) - - path_bytes = str(path_bytes or '').strip() - if path_bytes in ('1', '2', '3'): - where_parts.append(f'{path_bytes_expression} = ?') - where_params.append(int(path_bytes)) - elif path_bytes == 'unknown': - where_parts.append(f'{path_bytes_expression} = 0') - - device_role = str(device_role or '').strip().lower() - if device_role in ('companion', 'repeater', 'roomserver', 'sensor'): - where_parts.append("LOWER(COALESCE(c.role, '')) = ?") - where_params.append(device_role) - elif device_role == 'other': - where_parts.append("LOWER(COALESCE(c.role, '')) NOT IN ('companion', 'repeater', 'roomserver', 'sensor')") - - if hop_filter in ('0', '1', '2', '3'): - where_parts.append( - 'COALESCE(c.hop_count, 0) = ?' if hop_filter == '0' - else 'COALESCE(c.hop_count, 0) >= ?' - ) - where_params.append(int(hop_filter)) - - has_location_expression = ( - "((c.city IS NOT NULL AND c.city != '') OR " - "(c.state IS NOT NULL AND c.state != '') OR " - "(c.country IS NOT NULL AND c.country != '') OR " - "(c.latitude IS NOT NULL AND c.longitude IS NOT NULL " - "AND c.latitude != 0 AND c.longitude != 0))" - ) - if location_filter == 'known': - where_parts.append(has_location_expression) - elif location_filter == 'unknown': - where_parts.append(f'NOT {has_location_expression}') - - if starred == 'yes': - where_parts.append('COALESCE(c.is_starred, 0) = 1') - elif starred == 'no': - where_parts.append('COALESCE(c.is_starred, 0) = 0') - - where_clause = (' WHERE ' + ' AND '.join(where_parts)) if where_parts else '' - - pagination = None - filtered_stats = None - if page is not None and page_size is not None and not include_detail: - page_size = max(1, min(200, int(page_size))) - page = max(1, int(page)) - cursor.execute( - """ - SELECT - COUNT(*) AS total_items, - SUM(CASE WHEN c.last_heard >= datetime('now', 'localtime', '-24 hours') THEN 1 ELSE 0 END) AS contacts_24h, - SUM(CASE WHEN c.last_heard >= datetime('now', 'localtime', '-7 days') THEN 1 ELSE 0 END) AS contacts_7d, - SUM(CASE WHEN c.first_heard >= datetime('now', 'localtime', '-7 days') - AND LOWER(COALESCE(c.device_type, '')) LIKE '%companion%' THEN 1 ELSE 0 END) AS new_companions, - SUM(CASE WHEN c.first_heard >= datetime('now', 'localtime', '-7 days') - AND LOWER(COALESCE(c.device_type, '')) LIKE '%repeater%' THEN 1 ELSE 0 END) AS new_repeaters, - SUM(CASE WHEN c.first_heard >= datetime('now', 'localtime', '-7 days') - AND (LOWER(COALESCE(c.device_type, '')) LIKE '%room%' - OR LOWER(COALESCE(c.device_type, '')) LIKE '%server%') THEN 1 ELSE 0 END) AS new_room_servers - FROM complete_contact_tracking c - """ + where_clause, - tuple(where_params), - ) - aggregate = cursor.fetchone() - total_items = int(aggregate['total_items'] or 0) - total_pages = max(1, (total_items + page_size - 1) // page_size) - page = min(page, total_pages) - pagination = { - 'page': page, - 'page_size': page_size, - 'total_items': total_items, - 'total_pages': total_pages, - 'has_previous': page > 1, - 'has_next': page < total_pages, - } - filtered_stats = { - 'contacts_24h': int(aggregate['contacts_24h'] or 0), - 'contacts_7d': int(aggregate['contacts_7d'] or 0), - 'contacts_total': total_items, - 'new_companions': int(aggregate['new_companions'] or 0), - 'new_repeaters': int(aggregate['new_repeaters'] or 0), - 'new_room_servers': int(aggregate['new_room_servers'] or 0), - } - - # Fetch contacts directly (no join/group-by). The recent paths per contact are - # loaded in a second query below and assembled in Python. This avoids materializing - # a window-function CTE over all of observed_paths and grouping by every contact - # column (incl. the raw_advert_data blob) on every request. last_advert_timestamp is - # the per-contact value, so it matches the old MAX(...) over a single contact's rows. - detail_cols = "c.raw_advert_data," if include_detail else "" - sort_expressions = { - 'username': "LOWER(COALESCE(c.name, ''))", - 'device_type': "LOWER(COALESCE(c.device_type, ''))", - 'location': ( - "LOWER(CASE " - "WHEN c.city IS NOT NULL AND c.city != '' AND c.state IS NOT NULL AND c.state != '' " - "THEN c.city || ', ' || c.state " - "WHEN c.city IS NOT NULL AND c.city != '' THEN c.city " - "WHEN c.latitude IS NOT NULL AND c.longitude IS NOT NULL " - "AND c.latitude != 0 AND c.longitude != 0 THEN printf('%s, %s', c.latitude, c.longitude) " - "ELSE '' END)" - ), - 'snr': 'COALESCE(c.snr, 0)', - 'hop_count': 'COALESCE(c.hop_count, 0)', - 'path_bytes': path_bytes_expression, - 'first_heard': "COALESCE(c.first_heard, '')", - 'last_seen': "COALESCE(c.last_heard, '')", - 'advert_count': 'COALESCE(c.advert_count, 0)', - } - sort = sort if sort in (*sort_expressions.keys(), 'distance') else 'last_seen' - direction = 'asc' if direction == 'asc' else 'desc' - if sort == 'distance': - if bot_lat is None or bot_lon is None: - sort_expression = '0' - else: - conn.create_function('contacts_distance_km', 2, lambda lat, lon: ( - self._calculate_distance(bot_lat, bot_lon, lat, lon) - if lat is not None and lon is not None else 0 - )) - sort_expression = 'contacts_distance_km(c.latitude, c.longitude)' - else: - sort_expression = sort_expressions[sort] - - query_params = list(where_params) - limit_clause = '' - if pagination is not None: - limit_clause = ' LIMIT ? OFFSET ?' - query_params.extend([page_size, (page - 1) * page_size]) - - cursor.execute(""" - SELECT - c.public_key, c.name, c.role, c.device_type, - c.latitude, c.longitude, c.city, c.state, c.country, - c.snr, c.hop_count, c.first_heard, c.last_heard, - c.advert_count, c.is_currently_tracked, - """ + detail_cols + """ - c.signal_strength, - c.is_starred, c.out_path, c.out_path_len, c.out_bytes_per_hop, - """ + path_bytes_expression + """ AS path_bytes_per_hop, - c.last_advert_timestamp as last_message - FROM complete_contact_tracking c - """ + where_clause + """ - ORDER BY """ + sort_expression + f" {direction.upper()}, c.public_key ASC" + limit_clause, - tuple(query_params), - ) - - main_rows = cursor.fetchall() - - paths_by_key = {} - path_rows = [] - if main_rows: - path_params: list[Any] = [] - page_key_clause = '' - if pagination is not None: - # The interactive list enriches only the visible page. At most 200 keys are - # supplied, staying comfortably below SQLite's parameter limit and turning - # the former all-history window scan into targeted index lookups. - page_keys = [row['public_key'] for row in main_rows] - placeholders = ','.join('?' for _ in page_keys) - page_key_clause = f' AND public_key IN ({placeholders})' - path_params.extend(page_keys) - cursor.execute(""" - WITH recent_paths AS ( - SELECT public_key, path_hex, path_length, bytes_per_hop, - observation_count, last_seen, - ROW_NUMBER() OVER (PARTITION BY public_key ORDER BY last_seen DESC) as rn - FROM observed_paths - WHERE packet_type = 'advert' AND public_key IS NOT NULL - """ + page_key_clause + """ - ) - SELECT public_key, path_hex, path_length, bytes_per_hop, observation_count, last_seen - FROM recent_paths WHERE rn <= 50 - ORDER BY public_key, last_seen DESC - """, tuple(path_params)) - path_rows = cursor.fetchall() - - for prow in path_rows: - if not prow['path_hex']: # Skip empty paths - continue - bph = None - if prow['bytes_per_hop'] is not None: - try: - bph = int(prow['bytes_per_hop']) - if bph not in (1, 2, 3): - bph = 1 - except (TypeError, ValueError): - bph = 1 - paths_by_key.setdefault(prow['public_key'], []).append({ - 'path_hex': prow['path_hex'], - 'path_length': int(prow['path_length']) if prow['path_length'] is not None else 0, - 'bytes_per_hop': bph, - 'observation_count': int(prow['observation_count']) if prow['observation_count'] is not None else 1, - 'last_seen': prow['last_seen'] if prow['last_seen'] is not None else None - }) - - multibyte_hop_chunks = self._get_cached_contact_multibyte_hop_chunks(cursor) - chunk_buckets = self._bucket_hop_chunks(multibyte_hop_chunks) - - tracking = [] - for row in main_rows: - # Calculate distance if both bot and contact have coordinates - distance = None - if (bot_lat is not None and bot_lon is not None and - row['latitude'] is not None and row['longitude'] is not None): - distance = self._calculate_distance(bot_lat, bot_lon, row['latitude'], row['longitude']) - - # Recent paths for this contact (grouped from the second query above). The full - # path objects are NOT sent in the list payload (they were ~70% of its size and - # are only used in the per-contact modal); the UI fetches them on demand via - # /api/contact-detail. The list only needs the count and the badge. - all_paths = paths_by_key.get(row['public_key'], []) - paths_count = len(all_paths) - - # Preserve the legacy total_messages value: it was COUNT(*) over the LEFT-JOINed - # path rows, i.e. the number of paths, or 1 when a contact had no paths. - total_messages = max(1, paths_count) - - path_encoding_badge = self._compute_path_encoding_badge( - row, all_paths, chunk_buckets - ) - - # The badge/tooltip decodes out_path (the "primary" path) using out_bytes_per_hop. - # The contact column can be stale (e.g. left at 1 while the primary path is a 3-byte - # path), which makes a multi-byte path render as twice/three-times as many 1-byte - # hops. Index the encoding on the primary observed path itself, which carries the - # authoritative bytes_per_hop, falling back to the contact column when unmatched. - out_path_val = row['out_path'] if row['out_path'] is not None else '' - out_bytes_per_hop_val = row['out_bytes_per_hop'] if row['out_bytes_per_hop'] is not None else None - if out_path_val: - primary_path = next((p for p in all_paths if p['path_hex'] == out_path_val), None) - if primary_path and primary_path.get('bytes_per_hop') in (1, 2, 3): - out_bytes_per_hop_val = primary_path['bytes_per_hop'] - - entry = { - 'user_id': row['public_key'], - 'username': row['name'], - 'role': row['role'], - 'device_type': row['device_type'], - 'latitude': row['latitude'], - 'longitude': row['longitude'], - 'city': row['city'], - 'state': row['state'], - 'country': row['country'], - 'snr': row['snr'], - 'hop_count': row['hop_count'], - 'first_heard': row['first_heard'], - 'last_seen': row['last_heard'], - 'advert_count': row['advert_count'], - 'is_currently_tracked': row['is_currently_tracked'], - 'signal_strength': row['signal_strength'], - 'total_messages': total_messages, - 'last_message': row['last_message'], - 'distance': distance, - 'is_starred': bool(row['is_starred'] if row['is_starred'] is not None else 0), - 'out_path': out_path_val, - 'out_path_len': row['out_path_len'] if row['out_path_len'] is not None else -1, - 'out_bytes_per_hop': out_bytes_per_hop_val, - 'path_bytes_per_hop': int(row['path_bytes_per_hop'] or 0), - 'paths_count': paths_count, - 'path_encoding_badge': path_encoding_badge, - } - if include_detail: - # Full fidelity for the export endpoint (size-tolerant, infrequent download). - raw_advert_data = row['raw_advert_data'] - raw_advert_data_parsed = None - if raw_advert_data: - try: - import json - raw_advert_data_parsed = json.loads(raw_advert_data) - except Exception: - raw_advert_data_parsed = None - entry['all_paths'] = all_paths - entry['raw_advert_data'] = raw_advert_data - entry['raw_advert_data_parsed'] = raw_advert_data_parsed - tracking.append(entry) - - # Get server statistics for daily tracking using direct database queries - server_stats = {} - try: - # Check if daily_stats table exists - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='daily_stats'") - if cursor.fetchone(): - # 24h: Last 24 hours of advertisements - cursor.execute(""" - SELECT SUM(advert_count) FROM daily_stats - WHERE date >= date('now', 'localtime', '-1 day') - """) - server_stats['advertisements_24h'] = cursor.fetchone()[0] or 0 - - # 7d: Previous 6 days (excluding today) - cursor.execute(""" - SELECT SUM(advert_count) FROM daily_stats - WHERE date >= date('now', 'localtime', '-7 days') AND date < date('now', 'localtime') - """) - server_stats['advertisements_7d'] = cursor.fetchone()[0] or 0 - - # All: Everything - cursor.execute(""" - SELECT SUM(advert_count) FROM daily_stats - """) - server_stats['total_advertisements'] = cursor.fetchone()[0] or 0 - - # Nodes per day statistics - # Calculate today's unique nodes from complete_contact_tracking - # (last_heard in last 24 hours) since daily_stats might not have today's data yet - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM complete_contact_tracking - WHERE last_heard >= datetime('now', 'localtime', '-24 hours') - """) - server_stats['nodes_24h'] = cursor.fetchone()[0] or 0 - - # Get today's unique nodes by role for the stacked chart - cursor.execute(""" - SELECT role, COUNT(DISTINCT public_key) as count - FROM complete_contact_tracking - WHERE last_heard >= datetime('now', 'localtime', '-24 hours') - AND role IS NOT NULL AND role != '' - GROUP BY role - """) - today_by_role = {} - for row in cursor.fetchall(): - role = row[0].lower() if row[0] else 'unknown' - count = row[1] - today_by_role[role] = count - - server_stats['nodes_24h_by_role'] = { - 'companion': today_by_role.get('companion', 0), - 'repeater': today_by_role.get('repeater', 0), - 'roomserver': today_by_role.get('roomserver', 0), - 'sensor': today_by_role.get('sensor', 0), - 'other': sum(v for k, v in today_by_role.items() if k not in ['companion', 'repeater', 'roomserver', 'sensor']) - } - - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date >= date('now', 'localtime', '-7 days') AND date < date('now', 'localtime') - """) - server_stats['nodes_7d'] = cursor.fetchone()[0] or 0 - - # Calculate day-over-day and period-over-period comparisons - # Today vs 7 days ago (single day comparison) - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date = date('now', 'localtime', '-7 days') - """) - result = cursor.fetchone() - server_stats['nodes_7d_ago'] = result[0] if result and result[0] else 0 - - # Last 7 days vs previous 7 days (days 8-14 ago) - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date >= date('now', 'localtime', '-14 days') AND date < date('now', 'localtime', '-7 days') - """) - result = cursor.fetchone() - server_stats['nodes_prev_7d'] = result[0] if result and result[0] else 0 - - # Last 30 days vs previous 30 days (days 31-60 ago) - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date >= date('now', 'localtime', '-60 days') AND date < date('now', 'localtime', '-30 days') - """) - result = cursor.fetchone() - server_stats['nodes_prev_30d'] = result[0] if result and result[0] else 0 - - # Also get current period totals for comparison - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date >= date('now', 'localtime', '-7 days') - """) - server_stats['nodes_7d'] = cursor.fetchone()[0] or 0 - - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - WHERE date >= date('now', 'localtime', '-30 days') - """) - server_stats['nodes_30d'] = cursor.fetchone()[0] or 0 - - cursor.execute(""" - SELECT COUNT(DISTINCT public_key) FROM daily_stats - """) - server_stats['nodes_all'] = cursor.fetchone()[0] or 0 - - # Get daily unique node counts by role for the last 30 days for the stacked graph - # Join daily_stats with complete_contact_tracking to get role information - # This gives us accurate historical daily counts by role - cursor.execute(""" - SELECT ds.date, c.role, COUNT(DISTINCT ds.public_key) as daily_count - FROM daily_stats ds - LEFT JOIN complete_contact_tracking c ON ds.public_key = c.public_key - WHERE ds.date >= date('now', 'localtime', '-30 days') AND ds.date <= date('now', 'localtime') - AND (c.role IS NOT NULL AND c.role != '') - GROUP BY ds.date, c.role - ORDER BY ds.date ASC, c.role ASC - """) - daily_data_by_role = cursor.fetchall() - - # Organize data by date and role - daily_by_role = {} - for row in daily_data_by_role: - date_str = row[0] - role = (row[1] or 'unknown').lower() - count = row[2] - - if date_str not in daily_by_role: - daily_by_role[date_str] = {} - daily_by_role[date_str][role] = count - - # Convert to array format with all roles for each date - server_stats['daily_nodes_30d_by_role'] = [] - for date_str in sorted(daily_by_role.keys()): - roles_data = daily_by_role[date_str] - server_stats['daily_nodes_30d_by_role'].append({ - 'date': date_str, - 'companion': roles_data.get('companion', 0), - 'repeater': roles_data.get('repeater', 0), - 'roomserver': roles_data.get('roomserver', 0), - 'sensor': roles_data.get('sensor', 0), - 'other': sum(v for k, v in roles_data.items() if k not in ['companion', 'repeater', 'roomserver', 'sensor']) - }) - - # Also keep the total count for backward compatibility - cursor.execute(""" - SELECT date, COUNT(DISTINCT public_key) as daily_count - FROM daily_stats - WHERE date >= date('now', 'localtime', '-30 days') AND date <= date('now', 'localtime') - GROUP BY date - ORDER BY date ASC - """) - daily_data = cursor.fetchall() - server_stats['daily_nodes_30d'] = [ - {'date': row[0], 'count': row[1]} - for row in daily_data - ] - - except Exception as e: - self.logger.debug(f"Could not get server stats: {e}") - - result = { - 'tracking_data': tracking, - 'server_stats': server_stats - } - if pagination is not None: - result['pagination'] = pagination - result['filtered_stats'] = filtered_stats - return result - except Exception as e: - self.logger.error(f"Error getting tracking data: {e}") - return {'error': str(e)} - - def _calculate_distance(self, lat1, lon1, lat2, lon2): - """Calculate distance between two points using Haversine formula""" - import math - - # Convert latitude and longitude from degrees to radians - lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2]) - - # Haversine formula - dlat = lat2 - lat1 - dlon = lon2 - lon1 - a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2 - c = 2 * math.asin(math.sqrt(a)) - - # Radius of earth in kilometers - r = 6371 - - return c * r - - def _get_cache_data(self): - """Get cache data""" - try: - with self._db_connection() as conn: - cursor = conn.cursor() - - # Get cache statistics - cursor.execute("SELECT COUNT(*) FROM adverts") - total_adverts = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(*) FROM adverts - WHERE timestamp > datetime('now', '-1 hour') - """) - recent_adverts = cursor.fetchone()[0] - - cursor.execute(""" - SELECT COUNT(DISTINCT user_id) FROM adverts - WHERE timestamp > datetime('now', '-24 hours') - """) - active_users = cursor.fetchone()[0] - - return { - 'total_adverts': total_adverts, - 'recent_adverts_1h': recent_adverts, - 'active_users_24h': active_users, - 'timestamp': time.time() - } - except Exception as e: - self.logger.error(f"Error getting cache data: {e}") - return {'error': str(e)} - def _get_channels(self): """Get all configured channels from database plus additional decode-only channels""" diff --git a/modules/web_viewer/database_admin.py b/modules/web_viewer/database_admin.py new file mode 100644 index 0000000..e23dcad --- /dev/null +++ b/modules/web_viewer/database_admin.py @@ -0,0 +1,726 @@ +"""Database statistics, table info, cache view and optimization for the web viewer.""" + +from __future__ import annotations + +import time +from typing import Any + +from modules.security_utils import validate_sql_identifier +from modules.web_viewer.dashboard_stats import humanize_span + + +class DatabaseAdminMixin: + """Mixed into BotDataViewer.""" + + ALLOWED_TABLES: Any + _bucket_hop_chunks: Any + _clients_lock: Any + _contact_has_multibyte_path_evidence: Any + _db_connection: Any + _get_cached_contact_multibyte_hop_chunks: Any + connected_clients: Any + db_path: Any + logger: Any + + def _get_database_stats(self, top_users_window='all', top_commands_window='all', + top_paths_window='all', top_channels_window='all'): + """Get comprehensive database statistics for dashboard""" + try: + with self._db_connection() as conn: + cursor = conn.cursor() + + # Get all available tables + cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") + tables = [row[0] for row in cursor.fetchall()] + + # Filter tables by ALLOWED_TABLES whitelist for security + tables = [t for t in tables if t in self.ALLOWED_TABLES] + + with self._clients_lock: + client_count = len(self.connected_clients) + + stats = { + 'timestamp': time.time(), + 'connected_clients': client_count, + 'tables': tables + } + + # Contact and tracking statistics + if 'complete_contact_tracking' in tables: + cursor.execute("SELECT COUNT(*) FROM complete_contact_tracking") + stats['total_contacts'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(*) FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-24 hours') + """) + stats['contacts_24h'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(*) FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-7 days') + """) + stats['contacts_7d'] = cursor.fetchone()[0] + + # Contacts heard in 7d with multibyte path evidence. Scope observed_paths to 7d so + # the pie chart matches "last 7 days" (lifetime paths + stale out_bytes_per_hop + # otherwise inflated the percentage). + stats['contacts_7d_multibyte_path'] = 0 + chunk_buckets = self._bucket_hop_chunks(set()) + mb_advert_pks: set[str] = set() + if 'observed_paths' in tables: + try: + chunk_buckets = self._bucket_hop_chunks( + self._get_cached_contact_multibyte_hop_chunks( + cursor, recent_days=7 + ) + ) + # Use date() — julianday(iso8601) often returns NULL for Python isoformat() strings + cursor.execute( + """ + SELECT DISTINCT public_key FROM observed_paths + WHERE packet_type = 'advert' AND public_key IS NOT NULL + AND bytes_per_hop IN (2, 3) + AND date(last_seen) >= date('now', 'localtime', '-7 days') + """ + ) + mb_advert_pks = { + row["public_key"] for row in cursor.fetchall() if row["public_key"] + } + except Exception as e: + self.logger.debug(f"Could not load multibyte path sets for 7d stats: {e}") + try: + cursor.execute( + """ + SELECT public_key, role, out_bytes_per_hop + FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-7 days') + """ + ) + mb_7d = 0 + for row in cursor.fetchall(): + if self._contact_has_multibyte_path_evidence( + row["public_key"], + row["role"], + row["out_bytes_per_hop"], + mb_advert_pks, + chunk_buckets, + ): + mb_7d += 1 + stats['contacts_7d_multibyte_path'] = mb_7d + except Exception as e: + self.logger.debug(f"Could not compute contacts_7d_multibyte_path: {e}") + + cursor.execute(""" + SELECT COUNT(*) FROM complete_contact_tracking + WHERE is_currently_tracked = 1 + """) + stats['tracked_contacts'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT AVG(hop_count) FROM complete_contact_tracking + WHERE hop_count IS NOT NULL + """) + avg_hops = cursor.fetchone()[0] + stats['avg_hop_count'] = round(avg_hops, 1) if avg_hops else 0 + + cursor.execute(""" + SELECT MAX(hop_count) FROM complete_contact_tracking + WHERE hop_count IS NOT NULL + """) + stats['max_hop_count'] = cursor.fetchone()[0] or 0 + + cursor.execute(""" + SELECT COUNT(DISTINCT role) FROM complete_contact_tracking + WHERE role IS NOT NULL + """) + stats['unique_roles'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(DISTINCT device_type) FROM complete_contact_tracking + WHERE device_type IS NOT NULL + """) + stats['unique_device_types'] = cursor.fetchone()[0] + + # Incoming packets: multibyte path share over whatever packet_stream + # actually retains. The key names still say 7d for compatibility, + # but the window is reported honestly alongside them — + # packet_stream is pruned at 3 days, so the old label was never true. + stats['incoming_packets_7d'] = 0 + stats['incoming_packets_7d_multibyte_path'] = 0 + if 'packet_stream' in tables: + try: + cutoff_ts = time.time() - 7 * 86400 + cursor.execute( + """ + SELECT COUNT(*), + SUM(CASE WHEN bytes_per_hop IN (2, 3) THEN 1 ELSE 0 END), + MIN(timestamp) + FROM packet_stream + WHERE type = ? AND timestamp > ? AND route_type_name IS NOT NULL + """, + ("packet", cutoff_ts), + ) + row = cursor.fetchone() + stats['incoming_packets_7d'] = row[0] or 0 + stats['incoming_packets_7d_multibyte_path'] = row[1] or 0 + stats['incoming_packets_from'] = row[2] + stats['incoming_packets_window_label'] = humanize_span( + time.time() - row[2] if row[2] else None + ) + except Exception as e: + self.logger.debug(f"Could not compute incoming packet multibyte stats: {e}") + + # Advertisement statistics using daily tracking table + if 'daily_stats' in tables: + # Total advertisements (all time) + cursor.execute(""" + SELECT SUM(advert_count) FROM daily_stats + """) + total_adverts = cursor.fetchone()[0] + stats['total_advertisements'] = total_adverts or 0 + + # 24h advertisements + cursor.execute(""" + SELECT SUM(advert_count) FROM daily_stats + WHERE date = date('now', 'localtime') + """) + stats['advertisements_24h'] = cursor.fetchone()[0] or 0 + + # 7d advertisements (last 7 days, excluding today) + cursor.execute(""" + SELECT SUM(advert_count) FROM daily_stats + WHERE date >= date('now', 'localtime', '-7 days') AND date < date('now', 'localtime') + """) + stats['advertisements_7d'] = cursor.fetchone()[0] or 0 + + # Nodes per day statistics + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date = date('now', 'localtime') + """) + stats['nodes_24h'] = cursor.fetchone()[0] or 0 + + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date >= date('now', 'localtime', '-6 days') + """) + stats['nodes_7d'] = cursor.fetchone()[0] or 0 + + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + """) + stats['nodes_all'] = cursor.fetchone()[0] or 0 + else: + # Fallback to old method if daily table doesn't exist yet + if 'complete_contact_tracking' in tables: + cursor.execute(""" + SELECT SUM(advert_count) FROM complete_contact_tracking + """) + total_adverts = cursor.fetchone()[0] + stats['total_advertisements'] = total_adverts or 0 + + cursor.execute(""" + SELECT SUM(advert_count) FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-24 hours') + """) + stats['advertisements_24h'] = cursor.fetchone()[0] or 0 + + cursor.execute(""" + SELECT SUM(advert_count) FROM complete_contact_tracking + WHERE last_heard > datetime('now', 'localtime', '-7 days') + """) + stats['advertisements_7d'] = cursor.fetchone()[0] or 0 + + # Repeater contacts (if exists) + if 'repeater_contacts' in tables: + cursor.execute("SELECT COUNT(*) FROM repeater_contacts") + stats['repeater_contacts'] = cursor.fetchone()[0] + + cursor.execute("SELECT COUNT(*) FROM repeater_contacts WHERE is_active = 1") + stats['active_repeater_contacts'] = cursor.fetchone()[0] + + # Cache statistics + cache_tables = [t for t in tables if 'cache' in t] + stats['cache_tables'] = cache_tables + stats['total_cache_entries'] = 0 + stats['active_cache_entries'] = 0 + + for table in cache_tables: + try: + validate_sql_identifier(table) + except ValueError: + self.logger.warning(f"Rejecting invalid table name: {table!r}") + raise + cursor.execute(f"SELECT COUNT(*) FROM {table}") + count = cursor.fetchone()[0] + stats['total_cache_entries'] += count + stats[f'{table}_count'] = count + + # Get active entries (not expired) + cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE expires_at > datetime('now')") + active_count = cursor.fetchone()[0] + stats['active_cache_entries'] += active_count + stats[f'{table}_active'] = active_count + + # Message and command statistics (if stats tables exist) + if 'message_stats' in tables: + cursor.execute("SELECT COUNT(*) FROM message_stats") + stats['total_messages'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(*) FROM message_stats + WHERE timestamp > strftime('%s', 'now', '-24 hours') + """) + stats['messages_24h'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(DISTINCT sender_id) FROM message_stats + WHERE timestamp > strftime('%s', 'now', '-24 hours') + """) + stats['unique_senders_24h'] = cursor.fetchone()[0] + + # Total unique users and channels + cursor.execute("SELECT COUNT(DISTINCT sender_id) FROM message_stats") + stats['unique_users_total'] = cursor.fetchone()[0] + + cursor.execute("SELECT COUNT(DISTINCT channel) FROM message_stats WHERE channel IS NOT NULL") + stats['unique_channels_total'] = cursor.fetchone()[0] + + # Top users (most frequent message senders) - filter by time window + if top_users_window == '24h': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-24 hours')" + elif top_users_window == '7d': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-7 days')" + elif top_users_window == '30d': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-30 days')" + else: # 'all' + time_filter = "" + + query = f""" + SELECT sender_id, COUNT(*) as count + FROM message_stats + {time_filter} + GROUP BY sender_id + ORDER BY count DESC + LIMIT 15 + """ + cursor.execute(query) + stats['top_users'] = [{'user': row[0], 'count': row[1]} for row in cursor.fetchall()] + + # Top channels by message count - filter by time window + if top_channels_window == '24h': + time_filter = "AND timestamp > strftime('%s', 'now', '-24 hours')" + elif top_channels_window == '7d': + time_filter = "AND timestamp > strftime('%s', 'now', '-7 days')" + elif top_channels_window == '30d': + time_filter = "AND timestamp > strftime('%s', 'now', '-30 days')" + else: # 'all' + time_filter = "" + + query = f""" + SELECT channel, COUNT(*) as message_count, COUNT(DISTINCT sender_id) as unique_users + FROM message_stats + WHERE channel IS NOT NULL {time_filter} + GROUP BY channel + ORDER BY message_count DESC + LIMIT 10 + """ + cursor.execute(query) + stats['top_channels'] = [ + {'channel': row[0], 'messages': row[1], 'users': row[2]} + for row in cursor.fetchall() + ] + + if 'command_stats' in tables: + cursor.execute("SELECT COUNT(*) FROM command_stats") + stats['total_commands'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-24 hours') + """) + stats['commands_24h'] = cursor.fetchone()[0] + + # Top commands - filter by time window + if top_commands_window == '24h': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-24 hours')" + elif top_commands_window == '7d': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-7 days')" + elif top_commands_window == '30d': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-30 days')" + else: # 'all' + time_filter = "" + + query = f""" + SELECT command_name, COUNT(*) as count + FROM command_stats + {time_filter} + GROUP BY command_name + ORDER BY count DESC + LIMIT 15 + """ + cursor.execute(query) + stats['top_commands'] = [{'command': row[0], 'count': row[1]} for row in cursor.fetchall()] + + # Bot reply rates (commands that got responses) - calculate for different time windows + # 24 hour reply rate + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-24 hours') AND response_sent = 1 + """) + replied_24h = cursor.fetchone()[0] + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-24 hours') + """) + total_24h = cursor.fetchone()[0] + if total_24h > 0: + stats['bot_reply_rate_24h'] = round((replied_24h / total_24h) * 100, 1) + else: + stats['bot_reply_rate_24h'] = 0 + + # 7 day reply rate + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-7 days') AND response_sent = 1 + """) + replied_7d = cursor.fetchone()[0] + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-7 days') + """) + total_7d = cursor.fetchone()[0] + if total_7d > 0: + stats['bot_reply_rate_7d'] = round((replied_7d / total_7d) * 100, 1) + else: + stats['bot_reply_rate_7d'] = 0 + + # 30 day reply rate + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-30 days') AND response_sent = 1 + """) + replied_30d = cursor.fetchone()[0] + cursor.execute(""" + SELECT COUNT(*) FROM command_stats + WHERE timestamp > strftime('%s', 'now', '-30 days') + """) + total_30d = cursor.fetchone()[0] + if total_30d > 0: + stats['bot_reply_rate_30d'] = round((replied_30d / total_30d) * 100, 1) + else: + stats['bot_reply_rate_30d'] = 0 + + # Path statistics (if path_stats table exists) + if 'path_stats' in tables: + cursor.execute(""" + SELECT sender_id, path_length, path_string, timestamp + FROM path_stats + ORDER BY path_length DESC + LIMIT 1 + """) + longest_path = cursor.fetchone() + if longest_path: + stats['longest_path'] = { + 'user': longest_path[0], + 'path_length': longest_path[1], + 'path_string': longest_path[2], + 'timestamp': longest_path[3] + } + + # Top paths (longest paths) - filter by time window + if top_paths_window == '24h': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-24 hours')" + elif top_paths_window == '7d': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-7 days')" + elif top_paths_window == '30d': + time_filter = "WHERE timestamp > strftime('%s', 'now', '-30 days')" + else: # 'all' + time_filter = "" + + query = f""" + SELECT sender_id, path_length, path_string, timestamp + FROM path_stats + {time_filter} + ORDER BY path_length DESC + LIMIT 5 + """ + cursor.execute(query) + stats['top_paths'] = [ + { + 'user': row[0], + 'path_length': row[1], + 'path_string': row[2], + 'timestamp': row[3] + } + for row in cursor.fetchall() + ] + + # Network health metrics + if 'complete_contact_tracking' in tables: + cursor.execute(""" + SELECT AVG(snr) FROM complete_contact_tracking + WHERE snr IS NOT NULL AND last_heard > datetime('now', 'localtime', '-24 hours') + """) + avg_snr = cursor.fetchone()[0] + stats['avg_snr_24h'] = round(avg_snr, 1) if avg_snr else 0 + + cursor.execute(""" + SELECT AVG(signal_strength) FROM complete_contact_tracking + WHERE signal_strength IS NOT NULL AND last_heard > datetime('now', 'localtime', '-24 hours') + """) + avg_signal = cursor.fetchone()[0] + stats['avg_signal_strength_24h'] = round(avg_signal, 1) if avg_signal else 0 + + # Geographic distribution - only count currently tracked contacts heard in the last 30 days + # Normalize country names to avoid duplicates (e.g., "United States" vs "United States of America") + if 'complete_contact_tracking' in tables: + cursor.execute(""" + SELECT COUNT(DISTINCT + CASE + WHEN country IN ('United States', 'United States of America', 'US', 'USA') + THEN 'United States' + ELSE country + END + ) FROM complete_contact_tracking + WHERE country IS NOT NULL AND country != '' + AND last_heard > datetime('now', 'localtime', '-30 days') + AND is_currently_tracked = 1 + """) + stats['countries'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(DISTINCT state) FROM complete_contact_tracking + WHERE state IS NOT NULL AND state != '' + AND last_heard > datetime('now', 'localtime', '-30 days') + AND is_currently_tracked = 1 + """) + stats['states'] = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(DISTINCT city) FROM complete_contact_tracking + WHERE city IS NOT NULL AND city != '' + AND last_heard > datetime('now', 'localtime', '-30 days') + AND is_currently_tracked = 1 + """) + stats['cities'] = cursor.fetchone()[0] + + return stats + + except Exception as e: + self.logger.error(f"Error getting database stats: {e}") + return {'error': str(e)} + + def _get_database_info(self): + """Get comprehensive database information for database page""" + try: + with self._db_connection() as conn: + cursor = conn.cursor() + + # Get all tables + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") + table_names = [row[0] for row in cursor.fetchall()] + + # Filter tables by ALLOWED_TABLES whitelist for security + table_names = [ + name for name in table_names + if name in self.ALLOWED_TABLES + ] + + # Get table information + tables = [] + total_records = 0 + + for table_name in table_names: + try: + # Get record count + cursor.execute(f"SELECT COUNT(*) FROM {table_name}") + record_count = cursor.fetchone()[0] + total_records += record_count + + # Get table size (approximate) + cursor.execute(f"PRAGMA table_info({table_name})") + columns = cursor.fetchall() + + # Estimate size (rough calculation) + estimated_size = record_count * len(columns) * 50 # Rough estimate + size_str = f"{estimated_size:,} bytes" if estimated_size < 1024 else f"{estimated_size/1024:.1f} KB" + + # Get table description based on name + description = self._get_table_description(table_name) + + tables.append({ + 'name': table_name, + 'record_count': record_count, + 'size': size_str, + 'description': description + }) + + except Exception as e: + self.logger.debug(f"Error getting info for table {table_name}: {e}") + tables.append({ + 'name': table_name, + 'record_count': 0, + 'size': 'Unknown', + 'description': 'Error reading table' + }) + + # Get database file size + import os + try: + db_size_bytes = os.path.getsize(self.db_path) + if db_size_bytes < 1024: + db_size = f"{db_size_bytes} bytes" + elif db_size_bytes < 1024 * 1024: + db_size = f"{db_size_bytes/1024:.1f} KB" + else: + db_size = f"{db_size_bytes/(1024*1024):.1f} MB" + except: + db_size = "Unknown" + + return { + 'total_tables': len(table_names), + 'total_records': total_records, + 'last_updated': time.strftime('%Y-%m-%d %H:%M:%S'), + 'db_size': db_size, + 'tables': tables + } + + except Exception as e: + self.logger.error(f"Error getting database info: {e}") + return { + 'total_tables': 0, + 'total_records': 0, + 'last_updated': 'Error', + 'db_size': 'Unknown', + 'tables': [] + } + + def _is_safe_table_name(self, table_name: str) -> bool: + """Check if table name is in the ALLOWED_TABLES whitelist. + + Args: + table_name: The table name to validate + + Returns: + True if the table is in the allowed whitelist, False otherwise + """ + if not table_name or not isinstance(table_name, str): + return False + return table_name in self.ALLOWED_TABLES + + def _get_table_description(self, table_name): + """Get human-readable description for table""" + descriptions = { + 'packet_stream': 'Real-time packet and command data stream', + 'complete_contact_tracking': 'Contact tracking and device information', + 'repeater_contacts': 'Repeater contact management', + 'message_stats': 'Message statistics and analytics', + 'command_stats': 'Command execution statistics', + 'path_stats': 'Network path statistics', + 'geocoding_cache': 'Geocoding service cache', + 'generic_cache': 'General purpose cache storage' + } + return descriptions.get(table_name, 'Database table') + + def _optimize_database(self): + """Optimize database using VACUUM, ANALYZE, and REINDEX""" + try: + with self._db_connection() as conn: + cursor = conn.cursor() + + # Get initial database size + import os + initial_size = os.path.getsize(self.db_path) + + # Perform VACUUM to reclaim unused space + self.logger.info("Starting database VACUUM...") + cursor.execute("VACUUM") + vacuum_size = os.path.getsize(self.db_path) + vacuum_saved = initial_size - vacuum_size + + # Perform ANALYZE to update table statistics + self.logger.info("Starting database ANALYZE...") + cursor.execute("ANALYZE") + + # Get all tables for REINDEX + cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") + tables = [row[0] for row in cursor.fetchall()] + + # Filter tables by ALLOWED_TABLES whitelist for security + tables = [t for t in tables if t in self.ALLOWED_TABLES] + + # Perform REINDEX on all tables + self.logger.info("Starting database REINDEX...") + reindexed_tables = [] + for table in tables: + try: + cursor.execute(f"REINDEX {table}") + reindexed_tables.append(table) + except Exception as e: + self.logger.debug(f"Could not reindex table {table}: {e}") + + # Get final database size + final_size = os.path.getsize(self.db_path) + total_saved = initial_size - final_size + + # Format size information + def format_size(size_bytes): + if size_bytes < 1024: + return f"{size_bytes} bytes" + elif size_bytes < 1024 * 1024: + return f"{size_bytes/1024:.1f} KB" + else: + return f"{size_bytes/(1024*1024):.1f} MB" + + return { + 'success': True, + 'vacuum_result': f"VACUUM completed - saved {format_size(vacuum_saved)}", + 'analyze_result': f"ANALYZE completed - updated statistics for {len(tables)} tables", + 'reindex_result': f"REINDEX completed - rebuilt indexes for {len(reindexed_tables)} tables", + 'initial_size': format_size(initial_size), + 'final_size': format_size(final_size), + 'total_saved': format_size(total_saved), + 'tables_processed': len(tables), + 'tables_reindexed': len(reindexed_tables) + } + + except Exception as e: + self.logger.error(f"Error optimizing database: {e}") + return { + 'success': False, + 'error': str(e) + } + + def _get_cache_data(self): + """Get cache data""" + try: + with self._db_connection() as conn: + cursor = conn.cursor() + + # Get cache statistics + cursor.execute("SELECT COUNT(*) FROM adverts") + total_adverts = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(*) FROM adverts + WHERE timestamp > datetime('now', '-1 hour') + """) + recent_adverts = cursor.fetchone()[0] + + cursor.execute(""" + SELECT COUNT(DISTINCT user_id) FROM adverts + WHERE timestamp > datetime('now', '-24 hours') + """) + active_users = cursor.fetchone()[0] + + return { + 'total_adverts': total_adverts, + 'recent_adverts_1h': recent_adverts, + 'active_users_24h': active_users, + 'timestamp': time.time() + } + except Exception as e: + self.logger.error(f"Error getting cache data: {e}") + return {'error': str(e)} diff --git a/modules/web_viewer/tracking.py b/modules/web_viewer/tracking.py new file mode 100644 index 0000000..c692832 --- /dev/null +++ b/modules/web_viewer/tracking.py @@ -0,0 +1,631 @@ +"""Contact tracking data for the web viewer: the contacts table, contact detail and distances.""" + +from __future__ import annotations + +from typing import Any + + +class ContactTrackingMixin: + """Mixed into BotDataViewer.""" + + _bucket_hop_chunks: Any + _compute_path_encoding_badge: Any + _db_connection: Any + _get_cached_contact_multibyte_hop_chunks: Any + config: Any + logger: Any + + def _get_tracking_data( + self, + since='30d', + include_detail=False, + page: int | None = None, + page_size: int | None = None, + search: str = '', + path_bytes: str = '', + device_role: str = '', + hop_filter: str = '', + location_filter: str = '', + starred: str = '', + sort: str = 'last_seen', + direction: str = 'desc', + ): + """Get contact tracking data. since: 24h, 7d, 30d, 90d, or all (heard in that window). + + include_detail=False (the interactive /api/contacts list) omits per-contact ``all_paths`` + and ``raw_advert_data`` to keep the payload small; the UI fetches those on demand via + /api/contact-detail. The interactive route supplies ``page`` and ``page_size`` so path + enrichment is limited to visible contacts. include_detail=True (the export endpoint) + keeps the legacy full-result behavior and full fields. + """ + try: + with self._db_connection() as conn: + cursor = conn.cursor() + + # Get bot location from config + bot_lat = self.config.getfloat('Bot', 'bot_latitude', fallback=None) + bot_lon = self.config.getfloat('Bot', 'bot_longitude', fallback=None) + + # Filter by last_heard (default: last 30 days). last_heard is stored as ISO-text + # datetime in LOCAL time (e.g. '2026-06-16 09:03:49.606966', written by datetime.now()), + # so the cutoff must also be local: datetime('now', 'localtime', ...). Using bare + # datetime('now', ...) computes the cutoff in UTC and shaves the local UTC offset off + # the window (e.g. a "24h" filter only returns ~17h of data in US/Pacific). + datetime_offsets = { + '24h': "'-24 hours'", + '7d': "'-7 days'", + '30d': "'-30 days'", + '90d': "'-90 days'", + } + where_parts = [] + where_params: list[Any] = [] + # A node can have more than one observed advert path. Treat its byte class as + # the widest path encoding seen for it, with the contact's current out-path as a + # fallback for databases that have not retained an observed path yet. This gives + # the list one stable, sortable value instead of placing the same node in several + # byte buckets. Only count rows with a known 1/2/3 encoding so NULL/invalid + # observations do not collapse to "1-byte" and block the out-path fallback. + path_bytes_expression = """COALESCE(( + SELECT MAX(op.bytes_per_hop) + FROM observed_paths op + WHERE op.public_key = c.public_key + AND op.packet_type = 'advert' + AND op.path_hex IS NOT NULL AND op.path_hex != '' + AND op.bytes_per_hop IN (1, 2, 3) + ), CASE WHEN c.out_bytes_per_hop IN (1, 2, 3) + THEN c.out_bytes_per_hop ELSE 0 END)""" + if since in datetime_offsets: + where_parts.append( + f"c.last_heard >= datetime('now', 'localtime', {datetime_offsets[since]})" + ) + + search = (search or '').strip().lower()[:100] + if search and not include_detail: + # Match the former client-side behavior: public keys are prefix-only, + # while names, roles, device types, and locations match anywhere. + escaped = search.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_') + where_parts.append( + "(" + "LOWER(COALESCE(c.public_key, '')) LIKE ? ESCAPE '\\' OR " + "LOWER(COALESCE(c.name, '')) LIKE ? ESCAPE '\\' OR " + "LOWER(COALESCE(c.role, '')) LIKE ? ESCAPE '\\' OR " + "LOWER(COALESCE(c.device_type, '')) LIKE ? ESCAPE '\\' OR " + "LOWER(COALESCE(c.city, '')) LIKE ? ESCAPE '\\' OR " + "LOWER(COALESCE(c.state, '')) LIKE ? ESCAPE '\\' OR " + "LOWER(COALESCE(c.country, '')) LIKE ? ESCAPE '\\'" + ")" + ) + where_params.extend([f'{escaped}%'] + [f'%{escaped}%'] * 6) + + path_bytes = str(path_bytes or '').strip() + if path_bytes in ('1', '2', '3'): + where_parts.append(f'{path_bytes_expression} = ?') + where_params.append(int(path_bytes)) + elif path_bytes == 'unknown': + where_parts.append(f'{path_bytes_expression} = 0') + + device_role = str(device_role or '').strip().lower() + if device_role in ('companion', 'repeater', 'roomserver', 'sensor'): + where_parts.append("LOWER(COALESCE(c.role, '')) = ?") + where_params.append(device_role) + elif device_role == 'other': + where_parts.append("LOWER(COALESCE(c.role, '')) NOT IN ('companion', 'repeater', 'roomserver', 'sensor')") + + if hop_filter in ('0', '1', '2', '3'): + where_parts.append( + 'COALESCE(c.hop_count, 0) = ?' if hop_filter == '0' + else 'COALESCE(c.hop_count, 0) >= ?' + ) + where_params.append(int(hop_filter)) + + has_location_expression = ( + "((c.city IS NOT NULL AND c.city != '') OR " + "(c.state IS NOT NULL AND c.state != '') OR " + "(c.country IS NOT NULL AND c.country != '') OR " + "(c.latitude IS NOT NULL AND c.longitude IS NOT NULL " + "AND c.latitude != 0 AND c.longitude != 0))" + ) + if location_filter == 'known': + where_parts.append(has_location_expression) + elif location_filter == 'unknown': + where_parts.append(f'NOT {has_location_expression}') + + if starred == 'yes': + where_parts.append('COALESCE(c.is_starred, 0) = 1') + elif starred == 'no': + where_parts.append('COALESCE(c.is_starred, 0) = 0') + + where_clause = (' WHERE ' + ' AND '.join(where_parts)) if where_parts else '' + + pagination = None + filtered_stats = None + if page is not None and page_size is not None and not include_detail: + page_size = max(1, min(200, int(page_size))) + page = max(1, int(page)) + cursor.execute( + """ + SELECT + COUNT(*) AS total_items, + SUM(CASE WHEN c.last_heard >= datetime('now', 'localtime', '-24 hours') THEN 1 ELSE 0 END) AS contacts_24h, + SUM(CASE WHEN c.last_heard >= datetime('now', 'localtime', '-7 days') THEN 1 ELSE 0 END) AS contacts_7d, + SUM(CASE WHEN c.first_heard >= datetime('now', 'localtime', '-7 days') + AND LOWER(COALESCE(c.device_type, '')) LIKE '%companion%' THEN 1 ELSE 0 END) AS new_companions, + SUM(CASE WHEN c.first_heard >= datetime('now', 'localtime', '-7 days') + AND LOWER(COALESCE(c.device_type, '')) LIKE '%repeater%' THEN 1 ELSE 0 END) AS new_repeaters, + SUM(CASE WHEN c.first_heard >= datetime('now', 'localtime', '-7 days') + AND (LOWER(COALESCE(c.device_type, '')) LIKE '%room%' + OR LOWER(COALESCE(c.device_type, '')) LIKE '%server%') THEN 1 ELSE 0 END) AS new_room_servers + FROM complete_contact_tracking c + """ + where_clause, + tuple(where_params), + ) + aggregate = cursor.fetchone() + total_items = int(aggregate['total_items'] or 0) + total_pages = max(1, (total_items + page_size - 1) // page_size) + page = min(page, total_pages) + pagination = { + 'page': page, + 'page_size': page_size, + 'total_items': total_items, + 'total_pages': total_pages, + 'has_previous': page > 1, + 'has_next': page < total_pages, + } + filtered_stats = { + 'contacts_24h': int(aggregate['contacts_24h'] or 0), + 'contacts_7d': int(aggregate['contacts_7d'] or 0), + 'contacts_total': total_items, + 'new_companions': int(aggregate['new_companions'] or 0), + 'new_repeaters': int(aggregate['new_repeaters'] or 0), + 'new_room_servers': int(aggregate['new_room_servers'] or 0), + } + + # Fetch contacts directly (no join/group-by). The recent paths per contact are + # loaded in a second query below and assembled in Python. This avoids materializing + # a window-function CTE over all of observed_paths and grouping by every contact + # column (incl. the raw_advert_data blob) on every request. last_advert_timestamp is + # the per-contact value, so it matches the old MAX(...) over a single contact's rows. + detail_cols = "c.raw_advert_data," if include_detail else "" + sort_expressions = { + 'username': "LOWER(COALESCE(c.name, ''))", + 'device_type': "LOWER(COALESCE(c.device_type, ''))", + 'location': ( + "LOWER(CASE " + "WHEN c.city IS NOT NULL AND c.city != '' AND c.state IS NOT NULL AND c.state != '' " + "THEN c.city || ', ' || c.state " + "WHEN c.city IS NOT NULL AND c.city != '' THEN c.city " + "WHEN c.latitude IS NOT NULL AND c.longitude IS NOT NULL " + "AND c.latitude != 0 AND c.longitude != 0 THEN printf('%s, %s', c.latitude, c.longitude) " + "ELSE '' END)" + ), + 'snr': 'COALESCE(c.snr, 0)', + 'hop_count': 'COALESCE(c.hop_count, 0)', + 'path_bytes': path_bytes_expression, + 'first_heard': "COALESCE(c.first_heard, '')", + 'last_seen': "COALESCE(c.last_heard, '')", + 'advert_count': 'COALESCE(c.advert_count, 0)', + } + sort = sort if sort in (*sort_expressions.keys(), 'distance') else 'last_seen' + direction = 'asc' if direction == 'asc' else 'desc' + if sort == 'distance': + if bot_lat is None or bot_lon is None: + sort_expression = '0' + else: + conn.create_function('contacts_distance_km', 2, lambda lat, lon: ( + self._calculate_distance(bot_lat, bot_lon, lat, lon) + if lat is not None and lon is not None else 0 + )) + sort_expression = 'contacts_distance_km(c.latitude, c.longitude)' + else: + sort_expression = sort_expressions[sort] + + query_params = list(where_params) + limit_clause = '' + if pagination is not None: + limit_clause = ' LIMIT ? OFFSET ?' + query_params.extend([page_size, (page - 1) * page_size]) + + cursor.execute(""" + SELECT + c.public_key, c.name, c.role, c.device_type, + c.latitude, c.longitude, c.city, c.state, c.country, + c.snr, c.hop_count, c.first_heard, c.last_heard, + c.advert_count, c.is_currently_tracked, + """ + detail_cols + """ + c.signal_strength, + c.is_starred, c.out_path, c.out_path_len, c.out_bytes_per_hop, + """ + path_bytes_expression + """ AS path_bytes_per_hop, + c.last_advert_timestamp as last_message + FROM complete_contact_tracking c + """ + where_clause + """ + ORDER BY """ + sort_expression + f" {direction.upper()}, c.public_key ASC" + limit_clause, + tuple(query_params), + ) + + main_rows = cursor.fetchall() + + paths_by_key = {} + path_rows = [] + if main_rows: + path_params: list[Any] = [] + page_key_clause = '' + if pagination is not None: + # The interactive list enriches only the visible page. At most 200 keys are + # supplied, staying comfortably below SQLite's parameter limit and turning + # the former all-history window scan into targeted index lookups. + page_keys = [row['public_key'] for row in main_rows] + placeholders = ','.join('?' for _ in page_keys) + page_key_clause = f' AND public_key IN ({placeholders})' + path_params.extend(page_keys) + cursor.execute(""" + WITH recent_paths AS ( + SELECT public_key, path_hex, path_length, bytes_per_hop, + observation_count, last_seen, + ROW_NUMBER() OVER (PARTITION BY public_key ORDER BY last_seen DESC) as rn + FROM observed_paths + WHERE packet_type = 'advert' AND public_key IS NOT NULL + """ + page_key_clause + """ + ) + SELECT public_key, path_hex, path_length, bytes_per_hop, observation_count, last_seen + FROM recent_paths WHERE rn <= 50 + ORDER BY public_key, last_seen DESC + """, tuple(path_params)) + path_rows = cursor.fetchall() + + for prow in path_rows: + if not prow['path_hex']: # Skip empty paths + continue + bph = None + if prow['bytes_per_hop'] is not None: + try: + bph = int(prow['bytes_per_hop']) + if bph not in (1, 2, 3): + bph = 1 + except (TypeError, ValueError): + bph = 1 + paths_by_key.setdefault(prow['public_key'], []).append({ + 'path_hex': prow['path_hex'], + 'path_length': int(prow['path_length']) if prow['path_length'] is not None else 0, + 'bytes_per_hop': bph, + 'observation_count': int(prow['observation_count']) if prow['observation_count'] is not None else 1, + 'last_seen': prow['last_seen'] if prow['last_seen'] is not None else None + }) + + multibyte_hop_chunks = self._get_cached_contact_multibyte_hop_chunks(cursor) + chunk_buckets = self._bucket_hop_chunks(multibyte_hop_chunks) + + tracking = [] + for row in main_rows: + # Calculate distance if both bot and contact have coordinates + distance = None + if (bot_lat is not None and bot_lon is not None and + row['latitude'] is not None and row['longitude'] is not None): + distance = self._calculate_distance(bot_lat, bot_lon, row['latitude'], row['longitude']) + + # Recent paths for this contact (grouped from the second query above). The full + # path objects are NOT sent in the list payload (they were ~70% of its size and + # are only used in the per-contact modal); the UI fetches them on demand via + # /api/contact-detail. The list only needs the count and the badge. + all_paths = paths_by_key.get(row['public_key'], []) + paths_count = len(all_paths) + + # Preserve the legacy total_messages value: it was COUNT(*) over the LEFT-JOINed + # path rows, i.e. the number of paths, or 1 when a contact had no paths. + total_messages = max(1, paths_count) + + path_encoding_badge = self._compute_path_encoding_badge( + row, all_paths, chunk_buckets + ) + + # The badge/tooltip decodes out_path (the "primary" path) using out_bytes_per_hop. + # The contact column can be stale (e.g. left at 1 while the primary path is a 3-byte + # path), which makes a multi-byte path render as twice/three-times as many 1-byte + # hops. Index the encoding on the primary observed path itself, which carries the + # authoritative bytes_per_hop, falling back to the contact column when unmatched. + out_path_val = row['out_path'] if row['out_path'] is not None else '' + out_bytes_per_hop_val = row['out_bytes_per_hop'] if row['out_bytes_per_hop'] is not None else None + if out_path_val: + primary_path = next((p for p in all_paths if p['path_hex'] == out_path_val), None) + if primary_path and primary_path.get('bytes_per_hop') in (1, 2, 3): + out_bytes_per_hop_val = primary_path['bytes_per_hop'] + + entry = { + 'user_id': row['public_key'], + 'username': row['name'], + 'role': row['role'], + 'device_type': row['device_type'], + 'latitude': row['latitude'], + 'longitude': row['longitude'], + 'city': row['city'], + 'state': row['state'], + 'country': row['country'], + 'snr': row['snr'], + 'hop_count': row['hop_count'], + 'first_heard': row['first_heard'], + 'last_seen': row['last_heard'], + 'advert_count': row['advert_count'], + 'is_currently_tracked': row['is_currently_tracked'], + 'signal_strength': row['signal_strength'], + 'total_messages': total_messages, + 'last_message': row['last_message'], + 'distance': distance, + 'is_starred': bool(row['is_starred'] if row['is_starred'] is not None else 0), + 'out_path': out_path_val, + 'out_path_len': row['out_path_len'] if row['out_path_len'] is not None else -1, + 'out_bytes_per_hop': out_bytes_per_hop_val, + 'path_bytes_per_hop': int(row['path_bytes_per_hop'] or 0), + 'paths_count': paths_count, + 'path_encoding_badge': path_encoding_badge, + } + if include_detail: + # Full fidelity for the export endpoint (size-tolerant, infrequent download). + raw_advert_data = row['raw_advert_data'] + raw_advert_data_parsed = None + if raw_advert_data: + try: + import json + raw_advert_data_parsed = json.loads(raw_advert_data) + except Exception: + raw_advert_data_parsed = None + entry['all_paths'] = all_paths + entry['raw_advert_data'] = raw_advert_data + entry['raw_advert_data_parsed'] = raw_advert_data_parsed + tracking.append(entry) + + # Get server statistics for daily tracking using direct database queries + server_stats = {} + try: + # Check if daily_stats table exists + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='daily_stats'") + if cursor.fetchone(): + # 24h: Last 24 hours of advertisements + cursor.execute(""" + SELECT SUM(advert_count) FROM daily_stats + WHERE date >= date('now', 'localtime', '-1 day') + """) + server_stats['advertisements_24h'] = cursor.fetchone()[0] or 0 + + # 7d: Previous 6 days (excluding today) + cursor.execute(""" + SELECT SUM(advert_count) FROM daily_stats + WHERE date >= date('now', 'localtime', '-7 days') AND date < date('now', 'localtime') + """) + server_stats['advertisements_7d'] = cursor.fetchone()[0] or 0 + + # All: Everything + cursor.execute(""" + SELECT SUM(advert_count) FROM daily_stats + """) + server_stats['total_advertisements'] = cursor.fetchone()[0] or 0 + + # Nodes per day statistics + # Calculate today's unique nodes from complete_contact_tracking + # (last_heard in last 24 hours) since daily_stats might not have today's data yet + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM complete_contact_tracking + WHERE last_heard >= datetime('now', 'localtime', '-24 hours') + """) + server_stats['nodes_24h'] = cursor.fetchone()[0] or 0 + + # Get today's unique nodes by role for the stacked chart + cursor.execute(""" + SELECT role, COUNT(DISTINCT public_key) as count + FROM complete_contact_tracking + WHERE last_heard >= datetime('now', 'localtime', '-24 hours') + AND role IS NOT NULL AND role != '' + GROUP BY role + """) + today_by_role = {} + for row in cursor.fetchall(): + role = row[0].lower() if row[0] else 'unknown' + count = row[1] + today_by_role[role] = count + + server_stats['nodes_24h_by_role'] = { + 'companion': today_by_role.get('companion', 0), + 'repeater': today_by_role.get('repeater', 0), + 'roomserver': today_by_role.get('roomserver', 0), + 'sensor': today_by_role.get('sensor', 0), + 'other': sum(v for k, v in today_by_role.items() if k not in ['companion', 'repeater', 'roomserver', 'sensor']) + } + + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date >= date('now', 'localtime', '-7 days') AND date < date('now', 'localtime') + """) + server_stats['nodes_7d'] = cursor.fetchone()[0] or 0 + + # Calculate day-over-day and period-over-period comparisons + # Today vs 7 days ago (single day comparison) + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date = date('now', 'localtime', '-7 days') + """) + result = cursor.fetchone() + server_stats['nodes_7d_ago'] = result[0] if result and result[0] else 0 + + # Last 7 days vs previous 7 days (days 8-14 ago) + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date >= date('now', 'localtime', '-14 days') AND date < date('now', 'localtime', '-7 days') + """) + result = cursor.fetchone() + server_stats['nodes_prev_7d'] = result[0] if result and result[0] else 0 + + # Last 30 days vs previous 30 days (days 31-60 ago) + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date >= date('now', 'localtime', '-60 days') AND date < date('now', 'localtime', '-30 days') + """) + result = cursor.fetchone() + server_stats['nodes_prev_30d'] = result[0] if result and result[0] else 0 + + # Also get current period totals for comparison + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date >= date('now', 'localtime', '-7 days') + """) + server_stats['nodes_7d'] = cursor.fetchone()[0] or 0 + + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + WHERE date >= date('now', 'localtime', '-30 days') + """) + server_stats['nodes_30d'] = cursor.fetchone()[0] or 0 + + cursor.execute(""" + SELECT COUNT(DISTINCT public_key) FROM daily_stats + """) + server_stats['nodes_all'] = cursor.fetchone()[0] or 0 + + # Get daily unique node counts by role for the last 30 days for the stacked graph + # Join daily_stats with complete_contact_tracking to get role information + # This gives us accurate historical daily counts by role + cursor.execute(""" + SELECT ds.date, c.role, COUNT(DISTINCT ds.public_key) as daily_count + FROM daily_stats ds + LEFT JOIN complete_contact_tracking c ON ds.public_key = c.public_key + WHERE ds.date >= date('now', 'localtime', '-30 days') AND ds.date <= date('now', 'localtime') + AND (c.role IS NOT NULL AND c.role != '') + GROUP BY ds.date, c.role + ORDER BY ds.date ASC, c.role ASC + """) + daily_data_by_role = cursor.fetchall() + + # Organize data by date and role + daily_by_role = {} + for row in daily_data_by_role: + date_str = row[0] + role = (row[1] or 'unknown').lower() + count = row[2] + + if date_str not in daily_by_role: + daily_by_role[date_str] = {} + daily_by_role[date_str][role] = count + + # Convert to array format with all roles for each date + server_stats['daily_nodes_30d_by_role'] = [] + for date_str in sorted(daily_by_role.keys()): + roles_data = daily_by_role[date_str] + server_stats['daily_nodes_30d_by_role'].append({ + 'date': date_str, + 'companion': roles_data.get('companion', 0), + 'repeater': roles_data.get('repeater', 0), + 'roomserver': roles_data.get('roomserver', 0), + 'sensor': roles_data.get('sensor', 0), + 'other': sum(v for k, v in roles_data.items() if k not in ['companion', 'repeater', 'roomserver', 'sensor']) + }) + + # Also keep the total count for backward compatibility + cursor.execute(""" + SELECT date, COUNT(DISTINCT public_key) as daily_count + FROM daily_stats + WHERE date >= date('now', 'localtime', '-30 days') AND date <= date('now', 'localtime') + GROUP BY date + ORDER BY date ASC + """) + daily_data = cursor.fetchall() + server_stats['daily_nodes_30d'] = [ + {'date': row[0], 'count': row[1]} + for row in daily_data + ] + + except Exception as e: + self.logger.debug(f"Could not get server stats: {e}") + + result = { + 'tracking_data': tracking, + 'server_stats': server_stats + } + if pagination is not None: + result['pagination'] = pagination + result['filtered_stats'] = filtered_stats + return result + except Exception as e: + self.logger.error(f"Error getting tracking data: {e}") + return {'error': str(e)} + + def _get_contact_detail(self, public_key: str) -> dict: + """Per-contact detail loaded on demand by the contacts UI modals. + + Returns the recent advert paths (same shape as the old list ``all_paths``) and the raw + advertisement data. Both are excluded from the /api/contacts list payload — they were + ~85% of its size and are only needed when a single contact is opened. + """ + try: + with self._db_connection() as conn: + cursor = conn.cursor() + + # Recent advert paths (most-recent 50). The idx_observed_paths_advert_pk_seen covering + # index serves WHERE public_key=? AND packet_type='advert' ORDER BY last_seen DESC directly. + all_paths: list[dict[str, Any]] = [] + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='observed_paths'") + if cursor.fetchone(): + cursor.execute(""" + SELECT path_hex, path_length, bytes_per_hop, observation_count, last_seen + FROM observed_paths + WHERE packet_type = 'advert' AND public_key = ? + ORDER BY last_seen DESC + LIMIT 50 + """, (public_key,)) + for prow in cursor.fetchall(): + if not prow['path_hex']: + continue + bph = None + if prow['bytes_per_hop'] is not None: + try: + bph = int(prow['bytes_per_hop']) + if bph not in (1, 2, 3): + bph = 1 + except (TypeError, ValueError): + bph = 1 + all_paths.append({ + 'path_hex': prow['path_hex'], + 'path_length': int(prow['path_length']) if prow['path_length'] is not None else 0, + 'bytes_per_hop': bph, + 'observation_count': int(prow['observation_count']) if prow['observation_count'] is not None else 1, + 'last_seen': prow['last_seen'] if prow['last_seen'] is not None else None, + }) + + # Raw advertisement data for the "Advertisement Data" modal. + raw_advert_data = None + raw_advert_data_parsed = None + cursor.execute(""" + SELECT raw_advert_data FROM complete_contact_tracking + WHERE public_key = ? ORDER BY last_heard DESC LIMIT 1 + """, (public_key,)) + rad_row = cursor.fetchone() + if rad_row and rad_row['raw_advert_data']: + raw_advert_data = rad_row['raw_advert_data'] + try: + import json + raw_advert_data_parsed = json.loads(raw_advert_data) + except Exception: + raw_advert_data_parsed = None + + return { + 'all_paths': all_paths, + 'raw_advert_data': raw_advert_data, + 'raw_advert_data_parsed': raw_advert_data_parsed, + } + except Exception as e: + self.logger.error(f"Error getting contact detail: {e}") + return {'error': str(e)} + + def _calculate_distance(self, lat1, lon1, lat2, lon2): + """Calculate distance between two points using Haversine formula""" + import math + + # Convert latitude and longitude from degrees to radians + lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2]) + + # Haversine formula + dlat = lat2 - lat1 + dlon = lon2 - lon1 + a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2 + c = 2 * math.asin(math.sqrt(a)) + + # Radius of earth in kilometers + r = 6371 + + return c * r diff --git a/pyproject.toml b/pyproject.toml index 138b036..99c0895 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -197,6 +197,7 @@ module = [ "modules.service_plugins.packet_capture_service", "modules.service_plugins.weather_service", "modules.web_viewer.app", + "modules.web_viewer.tracking", # moved from web_viewer.app with its existing type errors "modules.web_viewer.integration", ] ignore_errors = true