496 lines
20 KiB
Python
496 lines
20 KiB
Python
import os
|
|
import json
|
|
import requests
|
|
import argparse
|
|
from collections import defaultdict
|
|
from tqdm import tqdm # For progress bars
|
|
import time # For delays between API requests
|
|
import csv # For writing CSV
|
|
|
|
# Cache file name
|
|
CACHE_FILE = "uuid_cache.json"
|
|
|
|
# Delay in seconds between API requests
|
|
API_DELAY = 1.0 # 1 second delay to avoid rate limiting
|
|
|
|
# Function to load cache
|
|
def load_cache(world_path):
|
|
cache_path = os.path.join(world_path, CACHE_FILE)
|
|
if os.path.exists(cache_path):
|
|
print(f"Loading UUID cache from {cache_path}")
|
|
with open(cache_path, 'r') as f:
|
|
cache = json.load(f)
|
|
print(f"Loaded {len(cache)} cached usernames")
|
|
return cache
|
|
else:
|
|
print("No UUID cache found, starting fresh")
|
|
return {}
|
|
|
|
# Function to save cache
|
|
def save_cache(world_path, cache):
|
|
cache_path = os.path.join(world_path, CACHE_FILE)
|
|
print(f"Saving UUID cache to {cache_path}")
|
|
with open(cache_path, 'w') as f:
|
|
json.dump(cache, f, indent=4)
|
|
print("Cache saved")
|
|
|
|
# Function to convert UUID to Minecraft username using Mojang API or cache
|
|
def uuid_to_username(uuid, cache, world_path):
|
|
if uuid in cache:
|
|
print(f"Using cached username for UUID {uuid}: {cache[uuid]}")
|
|
return cache[uuid]
|
|
|
|
print(f"Converting uncached UUID {uuid} to username...")
|
|
time.sleep(API_DELAY) # Delay before making API request
|
|
try:
|
|
url = f"https://api.mojang.com/user/profile/{uuid.replace('-', '')}"
|
|
print(f"Requesting Mojang API: {url}")
|
|
response = requests.get(url)
|
|
if response.status_code == 200:
|
|
username = response.json()['name']
|
|
print(f"Successfully got username: {username}")
|
|
cache[uuid] = username
|
|
save_cache(world_path, cache) # Save immediately after adding
|
|
return username
|
|
else:
|
|
fallback = f"Unknown_{uuid[:8]}"
|
|
print(f"API request failed (status {response.status_code}), using fallback: {fallback}")
|
|
cache[uuid] = fallback
|
|
save_cache(world_path, cache) # Save immediately after adding
|
|
return fallback
|
|
except Exception as e:
|
|
fallback = f"Unknown_{uuid[:8]}"
|
|
print(f"Exception during API request: {e}, using fallback: {fallback}")
|
|
cache[uuid] = fallback
|
|
save_cache(world_path, cache) # Save immediately after adding
|
|
return fallback
|
|
|
|
def get_stat_value(stats, path):
|
|
"""Get stat value by path like 'mined' for total, 'mined:stone', 'custom:play_time'"""
|
|
print(f"Getting stat value for path '{path}'...")
|
|
if ':' not in path:
|
|
cat = f"minecraft:{path}"
|
|
if cat in stats and isinstance(stats[cat], dict):
|
|
total = sum(stats[cat].values())
|
|
print(f"Total for category '{path}': {total}")
|
|
return total
|
|
print(f"No data for category '{path}'")
|
|
return 0
|
|
else:
|
|
cat, item = path.split(':', 1)
|
|
cat_key = f"minecraft:{cat}"
|
|
item_key = f"minecraft:{item}"
|
|
if cat_key in stats and isinstance(stats[cat_key], dict) and item_key in stats[cat_key]:
|
|
value = stats[cat_key][item_key]
|
|
print(f"Value for '{path}': {value}")
|
|
return value
|
|
print(f"No data for '{path}'")
|
|
return 0
|
|
|
|
def get_value(stats, info):
|
|
"""Get formatted value for predefined categories"""
|
|
print(f"Getting formatted value for category...")
|
|
if 'extractor' in info:
|
|
value = info['extractor'](stats)
|
|
print(f"Extractor value: {value}")
|
|
return value
|
|
path = info['path']
|
|
raw = get_stat_value(stats, path)
|
|
div = info.get('div', 1)
|
|
formatted = raw / div
|
|
print(f"Formatted value ({raw} / {div}): {formatted}")
|
|
return formatted
|
|
|
|
def get_categories(player_stats):
|
|
print("Collecting available categories from stats...")
|
|
cats = set()
|
|
total_players = len(player_stats)
|
|
progress_bar = tqdm(total=total_players, desc="Processing categories", unit="player")
|
|
for uuid, stats in player_stats.items():
|
|
print(f"Processing categories for UUID {uuid}")
|
|
for cat_full in stats:
|
|
if cat_full.startswith('minecraft:'):
|
|
cat = cat_full[10:] # remove 'minecraft:'
|
|
cats.add(cat)
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
sorted_cats = sorted(cats)
|
|
print(f"Found categories: {sorted_cats}")
|
|
return sorted_cats
|
|
|
|
def get_items(cat, player_stats):
|
|
print(f"Collecting unique items in category '{cat}'...")
|
|
items = set()
|
|
total_players = len(player_stats)
|
|
progress_bar = tqdm(total=total_players, desc=f"Processing items in '{cat}'", unit="player")
|
|
for uuid, stats in player_stats.items():
|
|
print(f"Processing items for UUID {uuid} in '{cat}'")
|
|
cat_key = f'minecraft:{cat}'
|
|
if cat_key in stats and isinstance(stats[cat_key], dict):
|
|
for item_full in stats[cat_key]:
|
|
item = item_full[10:] # remove 'minecraft:'
|
|
items.add(item)
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
sorted_items = sorted(items)
|
|
print(f"Found {len(sorted_items)} unique items in '{cat}'")
|
|
return sorted_items
|
|
|
|
# Formatters for categories: (divisor, unit)
|
|
formatters = {
|
|
'custom:play_time': (72000, 'hours'), # 20 ticks/sec * 3600 sec/hour
|
|
'custom:total_world_time': (72000, 'hours'),
|
|
'custom:damage_dealt': (10, 'hearts'),
|
|
'custom:damage_taken': (10, 'hearts'),
|
|
'custom:walk_one_cm': (100000, 'km'),
|
|
'custom:sprint_one_cm': (100000, 'km'),
|
|
'custom:fly_one_cm': (100000, 'km'),
|
|
'mined': (1, 'blocks'),
|
|
'crafted': (1, 'items'),
|
|
'used': (1, 'uses'),
|
|
'killed': (1, 'kills'),
|
|
'killed_by': (1, 'deaths'),
|
|
'custom:mob_kills': (1, 'kills'),
|
|
'custom:player_kills': (1, 'kills'),
|
|
'custom:deaths': (1, ''),
|
|
'picked_up': (1, 'items'),
|
|
'dropped': (1, 'items'),
|
|
'broken': (1, 'items'),
|
|
# Add more if needed
|
|
}
|
|
|
|
# Predefined categories (for default mode)
|
|
categories = {
|
|
'play_time': {
|
|
'path': 'custom:play_time',
|
|
'div': 72000,
|
|
'desc': 'Top players by play time (hours)'
|
|
},
|
|
'blocks_mined': {
|
|
'path': 'mined',
|
|
'div': 1,
|
|
'desc': 'Top players by blocks mined'
|
|
},
|
|
'mobs_killed': {
|
|
'path': 'custom:mob_kills',
|
|
'div': 1,
|
|
'desc': 'Top players by mobs killed'
|
|
},
|
|
'damage_dealt': {
|
|
'path': 'custom:damage_dealt',
|
|
'div': 10,
|
|
'desc': 'Top players by damage dealt (hearts)'
|
|
},
|
|
'distance_traveled': {
|
|
'extractor': lambda s: sum([
|
|
s.get('minecraft:custom', {}).get('minecraft:walk_one_cm', 0),
|
|
s.get('minecraft:custom', {}).get('minecraft:sprint_one_cm', 0),
|
|
s.get('minecraft:custom', {}).get('minecraft:fly_one_cm', 0)
|
|
]) / 100000,
|
|
'desc': 'Top players by distance traveled (km)'
|
|
}
|
|
}
|
|
|
|
def write_top_to_file(world_path, path, sorted_players, div, unit, precision, cache):
|
|
filename = f"top_{path.replace(':', '_')}.txt"
|
|
output_path = os.path.join(world_path, filename)
|
|
print(f"Writing top for '{path}' to file: {output_path}")
|
|
with open(output_path, 'w', encoding='utf-8') as f:
|
|
f.write(f"Top players by '{path}':\n")
|
|
for i, (uuid, value) in enumerate(sorted_players, 1):
|
|
username = uuid_to_username(uuid, cache, world_path) # Pass world_path
|
|
fvalue = value / div
|
|
line = f"{i}. {username}: {fvalue:.{precision}f}"
|
|
if unit:
|
|
line += f" {unit}"
|
|
f.write(line + "\n")
|
|
print(f"Written {len(sorted_players)} players to {output_path}")
|
|
|
|
def main(world_path, args):
|
|
stats_dir = os.path.join(world_path, 'stats')
|
|
print(f"Checking stats directory: {stats_dir}")
|
|
|
|
if not os.path.exists(stats_dir):
|
|
print(f"Stats directory not found: {stats_dir}")
|
|
return
|
|
|
|
# Load UUID cache
|
|
uuid_cache = load_cache(world_path)
|
|
|
|
player_stats = {}
|
|
print("Starting to read player stats files...")
|
|
|
|
# Get list of JSON files
|
|
json_files = [f for f in os.listdir(stats_dir) if f.endswith('.json')]
|
|
total_files = len(json_files)
|
|
print(f"Found {total_files} player stats files")
|
|
progress_bar = tqdm(total=total_files, desc="Loading player stats", unit="file")
|
|
|
|
# Read all JSON files in stats directory, store by UUID
|
|
for filename in json_files:
|
|
uuid = filename[:-5]
|
|
filepath = os.path.join(stats_dir, filename)
|
|
print(f"Processing file: {filepath} (UUID: {uuid})")
|
|
|
|
try:
|
|
with open(filepath, 'r') as f:
|
|
data = json.load(f)
|
|
raw_stats = data.get('stats', {})
|
|
print(f"Successfully loaded stats for UUID {uuid}")
|
|
player_stats[uuid] = raw_stats
|
|
print(f"Added UUID {uuid} to stats")
|
|
except json.JSONDecodeError as e:
|
|
print(f"Invalid JSON in {filepath}: {e}")
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
|
|
if not player_stats:
|
|
print("No player stats found.")
|
|
return
|
|
|
|
print(f"Loaded stats for {len(player_stats)} players")
|
|
|
|
if args.list_categories:
|
|
print("Listing available categories...")
|
|
cats = get_categories(player_stats)
|
|
content = "Available categories (use --stat 'cat' for total or --stat 'cat:item' for specific):\n"
|
|
for c in cats:
|
|
content += f" {c}\n"
|
|
content += "\nExamples:\n"
|
|
content += " --stat mined # total blocks mined\n"
|
|
content += " --stat mined:stone # stone mined\n"
|
|
content += " --stat killed:creeper # creepers killed\n"
|
|
content += " --stat custom:play_time # play time\n"
|
|
print(content)
|
|
output_path = os.path.join(world_path, "categories.txt")
|
|
with open(output_path, 'w') as f:
|
|
f.write(content)
|
|
print(f"Written categories to {output_path}")
|
|
save_cache(world_path, uuid_cache)
|
|
return
|
|
|
|
if args.list_items:
|
|
cat = args.list_items
|
|
print(f"Listing items in category '{cat}'...")
|
|
items_list = get_items(cat, player_stats)
|
|
content = f"Unique items in '{cat}' (sorted alphabetically):\n"
|
|
for item in items_list:
|
|
content += f" {item}\n"
|
|
print(content)
|
|
output_path = os.path.join(world_path, f"items_{cat}.txt")
|
|
with open(output_path, 'w') as f:
|
|
f.write(content)
|
|
print(f"Written items to {output_path}")
|
|
save_cache(world_path, uuid_cache)
|
|
return
|
|
|
|
top_n = args.top if args.top > 0 else len(player_stats) # If --top=0, all players
|
|
|
|
if args.all_categories:
|
|
print("Generating tops for all categories and items...")
|
|
cats = get_categories(player_stats)
|
|
|
|
# Calculate total tasks
|
|
total_tasks = 0
|
|
for category in cats:
|
|
items = get_items(category, player_stats) # Pre-fetch items to count
|
|
total_tasks += len(items)
|
|
if category != 'custom':
|
|
total_tasks += 1 # For total
|
|
print(f"Total tasks to process: {total_tasks}")
|
|
|
|
global_progress = tqdm(total=total_tasks, desc="Overall progress", unit="task", position=1, leave=True)
|
|
|
|
all_tops = [] # Collect all data
|
|
|
|
for category in cats:
|
|
print(f"Processing category '{category}'...")
|
|
|
|
# Generate total for category
|
|
if category != 'custom':
|
|
print(f"Generating total top for '{category}'...")
|
|
player_values = {}
|
|
total_players = len(player_stats)
|
|
progress_bar = tqdm(total=total_players, desc=f"Calculating total '{category}' values", unit="player", position=0, leave=False)
|
|
for uuid, stats in player_stats.items():
|
|
player_values[uuid] = get_stat_value(stats, category)
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
|
|
sorted_players = sorted(
|
|
player_values.items(),
|
|
key=lambda item: item[1],
|
|
reverse=True
|
|
)
|
|
print(f"Sorted {len(sorted_players)} players for total '{category}'")
|
|
|
|
div, unit = formatters.get(category, (1, ''))
|
|
precision = 2 if div > 1 else 0
|
|
|
|
for i, (uuid, value) in enumerate(sorted_players[:top_n], 1):
|
|
username = uuid_to_username(uuid, uuid_cache, world_path)
|
|
fvalue = round(value / div, precision)
|
|
all_tops.append({
|
|
'stat': category,
|
|
'rank': i,
|
|
'username': username,
|
|
'uuid': uuid,
|
|
'raw_value': value,
|
|
'formatted_value': fvalue,
|
|
'unit': unit
|
|
})
|
|
global_progress.update(1)
|
|
|
|
# Get unique items for the category
|
|
items = get_items(category, player_stats)
|
|
|
|
# Generate top for each item
|
|
for item in items:
|
|
path = f"{category}:{item}"
|
|
print(f"Generating top for '{path}'...")
|
|
|
|
player_values = {}
|
|
total_players = len(player_stats)
|
|
progress_bar = tqdm(total=total_players, desc=f"Calculating '{path}' values", unit="player", position=0, leave=False)
|
|
for uuid, stats in player_stats.items():
|
|
player_values[uuid] = get_stat_value(stats, path)
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
|
|
sorted_players = sorted(
|
|
player_values.items(),
|
|
key=lambda item: item[1],
|
|
reverse=True
|
|
)
|
|
print(f"Sorted {len(sorted_players)} players for '{path}'")
|
|
|
|
div, unit = formatters.get(category, (1, '')) # Use category formatter for items
|
|
precision = 2 if div > 1 else 0
|
|
|
|
for i, (uuid, value) in enumerate(sorted_players[:top_n], 1):
|
|
username = uuid_to_username(uuid, uuid_cache, world_path)
|
|
fvalue = round(value / div, precision)
|
|
all_tops.append({
|
|
'stat': path,
|
|
'rank': i,
|
|
'username': username,
|
|
'uuid': uuid,
|
|
'raw_value': value,
|
|
'formatted_value': fvalue,
|
|
'unit': unit
|
|
})
|
|
global_progress.update(1)
|
|
|
|
global_progress.close()
|
|
|
|
# Write all to single CSV
|
|
output_path = os.path.join(world_path, "all_tops.csv")
|
|
print(f"Writing all tops to CSV: {output_path}")
|
|
with open(output_path, 'w', newline='', encoding='utf-8') as csvfile:
|
|
fieldnames = ['stat', 'rank', 'username', 'uuid', 'raw_value', 'formatted_value', 'unit']
|
|
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
for row in all_tops:
|
|
writer.writerow(row)
|
|
print(f"Written {len(all_tops)} rows to {output_path}")
|
|
|
|
save_cache(world_path, uuid_cache)
|
|
return
|
|
|
|
if args.stat:
|
|
path = args.stat
|
|
print(f"Generating top for stat '{path}'...")
|
|
|
|
player_values = {}
|
|
total_players = len(player_stats)
|
|
progress_bar = tqdm(total=total_players, desc="Calculating stat values", unit="player")
|
|
for uuid, stats in player_stats.items():
|
|
player_values[uuid] = get_stat_value(stats, path)
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
|
|
sorted_players = sorted(
|
|
player_values.items(),
|
|
key=lambda item: item[1],
|
|
reverse=True
|
|
)
|
|
print(f"Sorted {len(sorted_players)} players")
|
|
div, unit = formatters.get(path, (1, ''))
|
|
precision = 2 if div > 1 else 0
|
|
print(f"Using formatter: divisor={div}, unit='{unit}', precision={precision}")
|
|
content = f"Top players by '{path}':\n"
|
|
|
|
output_progress = tqdm(total=min(top_n, len(sorted_players)), desc="Converting and outputting top players", unit="player")
|
|
for i, (uuid, value) in enumerate(sorted_players[:top_n], 1):
|
|
username = uuid_to_username(uuid, uuid_cache, world_path)
|
|
fvalue = value / div
|
|
line = f"{i}. {username}: {fvalue:.{precision}f}"
|
|
if unit:
|
|
line += f" {unit}"
|
|
content += line + "\n"
|
|
print(line)
|
|
output_progress.update(1)
|
|
output_progress.close()
|
|
output_path = os.path.join(world_path, f"top_{path.replace(':', '_')}.txt")
|
|
with open(output_path, 'w') as f:
|
|
f.write(content)
|
|
print(f"Written to {output_path}")
|
|
save_cache(world_path, uuid_cache)
|
|
return
|
|
|
|
# Default: predefined tops
|
|
print("Generating predefined tops...")
|
|
for cat_key, cat_info in categories.items():
|
|
print(f"Processing predefined category '{cat_key}': {cat_info['desc']}")
|
|
|
|
player_values = {}
|
|
total_players = len(player_stats)
|
|
progress_bar = tqdm(total=total_players, desc=f"Calculating '{cat_key}' values", unit="player")
|
|
for uuid, stats in player_stats.items():
|
|
player_values[uuid] = get_value(stats, cat_info)
|
|
progress_bar.update(1)
|
|
progress_bar.close()
|
|
|
|
sorted_players = sorted(
|
|
player_values.items(),
|
|
key=lambda item: item[1],
|
|
reverse=True
|
|
)
|
|
print(f"Sorted {len(sorted_players)} players for '{cat_key}'")
|
|
content = f"{cat_info['desc']}:\n"
|
|
|
|
output_progress = tqdm(total=min(top_n, len(sorted_players)), desc="Converting and outputting top players", unit="player")
|
|
for i, (uuid, value) in enumerate(sorted_players[:top_n], 1):
|
|
username = uuid_to_username(uuid, uuid_cache, world_path)
|
|
line = f"{i}. {username}: {value:.2f}"
|
|
content += line + "\n"
|
|
print(line)
|
|
output_progress.update(1)
|
|
output_progress.close()
|
|
output_path = os.path.join(world_path, f"top_{cat_key}.txt")
|
|
with open(output_path, 'w') as f:
|
|
f.write(content)
|
|
print(f"Written to {output_path}")
|
|
|
|
save_cache(world_path, uuid_cache)
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting script...")
|
|
parser = argparse.ArgumentParser(description="Generate Minecraft player tops from world stats")
|
|
parser.add_argument("world_path", help="Path to the Minecraft world directory")
|
|
parser.add_argument("--stat", help="Stat path e.g. 'mined:stone', 'killed:creeper', 'mined' for total")
|
|
parser.add_argument("--list-categories", action="store_true", help="List available categories")
|
|
parser.add_argument("--list-items", help="List unique items in a category e.g. 'mined'")
|
|
parser.add_argument("--all-categories", action="store_true", help="Generate tops for all categories and items (all players)")
|
|
parser.add_argument("--top", type=int, default=0, help="Number of top players to show (0 for all, default all for --all-categories)")
|
|
args = parser.parse_args()
|
|
print(f"Parsed arguments: {vars(args)}")
|
|
|
|
# Mutual exclusivity check
|
|
active_modes = [args.stat, args.list_categories, args.list_items, args.all_categories]
|
|
if sum(bool(x) for x in active_modes) > 1:
|
|
print("Use only one of --stat, --list-categories, --list-items, --all-categories")
|
|
exit(1)
|
|
|
|
main(args.world_path, args)
|
|
print("Script execution completed.") |