File size: 3,431 Bytes
5a98598 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
import os
import gradio as gr
import pandas as pd
from datetime import datetime
import time
from count_ip_data import count_files_per_ip
import threading
# Define the path for storing the data
DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
STATS_FILE = os.path.join(DATA_DIR, 'battle_stats.csv')
LAST_UPDATE_FILE = os.path.join(DATA_DIR, 'last_update.txt')
# Ensure data directory exists
os.makedirs(DATA_DIR, exist_ok=True)
def save_stats(df, current_time):
"""Save statistics and last update time to files"""
df.to_csv(STATS_FILE, index=False)
with open(LAST_UPDATE_FILE, 'w') as f:
f.write(current_time)
def load_stats():
"""Load statistics and last update time from files"""
try:
df = pd.read_csv(STATS_FILE)
with open(LAST_UPDATE_FILE, 'r') as f:
last_update = f.read().strip()
return df, last_update
except (FileNotFoundError, pd.errors.EmptyDataError):
return pd.DataFrame(columns=['IP Address', 'Battle Count']), ""
def update_stats():
"""Get the latest battle statistics"""
smb_url = os.getenv("SMB_URL")
if not smb_url:
return pd.DataFrame(columns=['IP Address', 'Battle Count']), ""
ip_counts = count_files_per_ip(smb_url)
# Convert to DataFrame for better display
df = pd.DataFrame(list(ip_counts.items()), columns=['IP Address', 'Battle Count'])
df = df.sort_values('Battle Count', ascending=False)
# Get current time
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# Save the updated stats
save_stats(df, current_time)
return df, current_time
def auto_update(state):
"""Background task to update stats every hour"""
while state['running']:
state['stats'], state['last_update'] = update_stats()
time.sleep(3600) # Sleep for 1 hour
def create_ui():
state = {'running': True}
# Try to load existing stats first
state['stats'], state['last_update'] = load_stats()
# If no existing stats or they're empty, update them
if state['stats'].empty:
state['stats'], state['last_update'] = update_stats()
# Start background update thread
update_thread = threading.Thread(target=auto_update, args=(state,))
update_thread.daemon = True
update_thread.start()
def get_current_stats():
return state['stats']
def get_last_update():
return state['last_update']
def manual_refresh():
state['stats'], state['last_update'] = update_stats()
return state['stats'], state['last_update']
with gr.Blocks(title="Battle Count Statistics") as app:
gr.Markdown("# Battle Count Statistics")
gr.Markdown("Displays the count of valid battles per IP address. Updates automatically every hour.")
with gr.Row():
last_update = gr.Textbox(
value=get_last_update,
label="Last Updated",
interactive=False
)
with gr.Row():
output = gr.DataFrame(
value=get_current_stats,
interactive=False,
wrap=True,
)
# refresh_btn = gr.Button("Refresh Now")
# refresh_btn.click(fn=manual_refresh, outputs=[output, last_update])
return app
if __name__ == "__main__":
app = create_ui()
app.launch() |