|
from snek.system.service import BaseService
|
|
import sqlite3
|
|
|
|
class StatisticsService(BaseService):
|
|
|
|
def database(self):
|
|
db_path = self.app.db_path.split("///")[-1]
|
|
print(db_path)
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
# Existing analysis code...
|
|
def get_table_columns(table_name):
|
|
cursor.execute(f"PRAGMA table_info({table_name})")
|
|
return cursor.fetchall()
|
|
|
|
tables = [
|
|
'http_access', 'user', 'channel', 'channel_member', 'broadcast',
|
|
'channel_message', 'notification', 'repository', 'test', 'drive',
|
|
'user_property', 'a', 'channel_attachment', 'push_registration'
|
|
]
|
|
|
|
for table in tables:
|
|
print(f"\n--- Statistics for table: {table} ---")
|
|
columns = get_table_columns(table)
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table}")
|
|
total_rows = cursor.fetchone()[0]
|
|
print(f"Total rows: {total_rows}")
|
|
|
|
for col in columns:
|
|
cid, name, col_type, notnull, dflt_value, pk = col
|
|
col_type_upper = col_type.upper()
|
|
|
|
cursor.execute(f"SELECT COUNT(DISTINCT '{name}') FROM {table}")
|
|
distinct_count = cursor.fetchone()[0]
|
|
print(f"\nColumn: {name} ({col_type})")
|
|
print(f"Distinct values: {distinct_count}")
|
|
|
|
if 'INT' in col_type_upper or 'BIGINT' in col_type_upper or 'FLOAT' in col_type_upper:
|
|
cursor.execute(f"SELECT MIN('{name}'), MAX('{name}'), AVG('{name}') FROM {table} WHERE '{name}' IS NOT NULL")
|
|
min_val, max_val, avg_val = cursor.fetchone()
|
|
print(f"Min: {min_val}, Max: {max_val}, Avg: {avg_val}")
|
|
|
|
elif 'TEXT' in col_type_upper and ('date' in name.lower() or 'time' in name.lower() or 'created' in name.lower() or 'updated' in name.lower() or 'on' in name.lower()):
|
|
cursor.execute(f"SELECT MIN({name}), MAX({name}) FROM {table} WHERE {name} IS NOT NULL")
|
|
min_date, max_date = cursor.fetchone()
|
|
print(f"Earliest: {min_date}, Latest: {max_date}")
|
|
|
|
elif 'TEXT' in col_type_upper:
|
|
cursor.execute(f"SELECT LENGTH({name}) FROM {table} WHERE {name} IS NOT NULL")
|
|
lengths = [len_row[0] for len_row in cursor.fetchall()]
|
|
if lengths:
|
|
avg_length = sum(lengths) / len(lengths)
|
|
max_length = max(lengths)
|
|
min_length = min(lengths)
|
|
print(f"Avg length: {avg_length:.2f}, Max length: {max_length}, Min length: {min_length}")
|
|
else:
|
|
print("No data to compute length statistics.")
|
|
|
|
# New statistics functions
|
|
def get_time_series_stats(table_name, date_column):
|
|
cursor.execute(f"SELECT strftime('%Y-%m-%d', {date_column}) AS day, COUNT(*) FROM {table_name} GROUP BY day")
|
|
return cursor.fetchall()
|
|
|
|
def get_count_created(table_name, date_column):
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
|
|
return cursor.fetchone()[0]
|
|
|
|
def get_channels_per_user():
|
|
cursor.execute("SELECT user_uid, COUNT(*) AS channel_count FROM channel_member GROUP BY user_uid ORDER BY channel_count DESC")
|
|
return cursor.fetchall()
|
|
|
|
def get_online_users():
|
|
cursor.execute("SELECT COUNT(*) FROM user WHERE last_ping >= datetime('now', '-5 minutes')")
|
|
return cursor.fetchone()[0]
|
|
|
|
# Example usage of new functions
|
|
messages_per_day = get_time_series_stats('channel_message', 'created_at')
|
|
users_created = get_count_created('user', 'created_at')
|
|
channels_created = get_count_created('channel', 'created_at')
|
|
channels_per_user = get_channels_per_user()
|
|
online_users = get_online_users()
|
|
|
|
# Print or store these stats as needed
|
|
print("\nMessages per day:", messages_per_day)
|
|
print("Total users created:", users_created)
|
|
print("Total channels created:", channels_created)
|
|
print("Channels per user (top):", channels_per_user[:10])
|
|
print("Currently online users:", online_users)
|
|
|
|
conn.close()
|