HASH e221e34c728b
DATE 2025-08-29
SUBJECT nix: simplify team leaderboard logic
FILES 1 CHANGED
HASH e221e34c728b
DATE 2025-08-29
SUBJECT nix: simplify team leaderboard logic
FILES 1 CHANGED
┌─ nix/apps/team-leaderboard-generator.nix ──────────────────────────────────┐│ diff --git a/nix/apps/team-leaderboard-generator.nix b/nix/apps/team-leaderboard-g ││ enerator.nix ││ index e508ef3..3f77514 100644 ││ --- a/nix/apps/team-leaderboard-generator.nix ││ +++ b/nix/apps/team-leaderboard-generator.nix ││ @@ -109,157 +109,99 @@ ││ return team_info ││ ││ def deduplicate_overlapping_teams(teams, all_members_data): ││ - # merge teams that share most of their best times ││ + # Simple rule: No two teams can share ANY runs - if they do, they're th ││ e same team ││ remaining_teams = teams.copy() ││ deduplicated = [] ││ ││ while remaining_teams: ││ current_team = remaining_teams.pop(0) ││ - current_best_times = current_team["best_runs_per_dungeon"] ││ - ││ - # Find teams that share most best times with current team ││ - similar_teams = [current_team] ││ - non_similar = [] ││ + current_run_ids = set() ││ + ││ + # Create unique identifiers for all runs in current team ││ + for run in current_team["all_runs"]: ││ + member_names_sorted = tuple(sorted(run["member_names"])) ││ + run_id = (run["duration"], run["completed_timestamp"], member_n ││ ames_sorted) ││ + current_run_ids.add(run_id) ││ + ││ + # Find teams that share ANY runs with current team ││ + same_teams = [current_team] ││ + non_overlapping = [] ││ ││ for other_team in remaining_teams: ││ - other_best_times = other_team["best_runs_per_dungeon"] ││ - ││ - # Count how many dungeons have identical best times ││ - matching_dungeons = 0 ││ - total_dungeons = min(len(current_best_times), len(other_best_ti ││ mes)) ││ - ││ - for dungeon_slug in current_best_times: ││ - if dungeon_slug in other_best_times: ││ - if current_best_times[dungeon_slug]["duration"] == othe ││ r_best_times[dungeon_slug]["duration"]: ││ - matching_dungeons += 1 ││ - ││ - # If they share 6+ identical times out of 9 dungeons, consider ││ them same team ││ - match_percentage = matching_dungeons / total_dungeons if total_ ││ dungeons > 0 else 0 ││ - if matching_dungeons >= 6 and match_percentage >= 0.6: ││ - similar_teams.append(other_team) ││ + other_run_ids = set() ││ + ││ + # Create unique identifiers for all runs in other team ││ + for run in other_team["all_runs"]: ││ + member_names_sorted = tuple(sorted(run["member_names"])) ││ + run_id = (run["duration"], run["completed_timestamp"], memb ││ er_names_sorted) ││ + other_run_ids.add(run_id) ││ + ││ + # If they share ANY runs, they're the same underlying team ││ + if current_run_ids.intersection(other_run_ids): ││ + same_teams.append(other_team) ││ else: ││ - non_similar.append(other_team) ││ + non_overlapping.append(other_team) ││ ││ - remaining_teams = non_similar ││ + remaining_teams = non_overlapping ││ ││ - if len(similar_teams) == 1: ││ - # No similar teams found ││ + if len(same_teams) == 1: ││ + # No overlapping teams found, keep as is ││ deduplicated.append(current_team) ││ else: ││ - # Merge similar teams (same underlying team with different 3-pl ││ ayer core perspectives) ││ - # Collect all members and data ││ - all_core_members = set() ││ - all_extended_members = set() ││ + # Multiple teams share runs - merge them into one team ││ + # Use the team with the best combined time as the representativ ││ e ││ + best_team = min(same_teams, key=lambda t: t["combined_best_time ││ "]) ││ + ││ + # Merge all data from overlapping teams ││ + # BUT only include players who appear in the BEST runs ││ all_runs = [] ││ regions_played = set() ││ total_runs = 0 ││ - best_combined_time = float('inf') ││ - best_team_runs = None ││ - ││ - for team in similar_teams: ││ - # Collect core members ││ - core_ids = list(map(int, team["team_signature"].split("-")) ││ ) ││ - all_core_members.update(core_ids) ││ - ││ - # Collect extended roster ││ - for member in team["extended_roster"]: ││ - all_extended_members.add(member["name"] + "@" + member[ ││ "realm_slug"]) ││ + seen_runs = set() ││ ││ - # Collect runs and other data ││ - all_runs.extend(team["all_runs"]) ││ + for team in same_teams: ││ + # Collect all unique runs ││ regions_played.update(team["regions_played"]) ││ total_runs += team["total_runs"] ││ + ││ + for run in team["all_runs"]: ││ + member_names_sorted = tuple(sorted(run["member_names"]) ││ ) ││ + run_id = (run["duration"], run["completed_timestamp"], ││ member_names_sorted) ││ + if run_id not in seen_runs: ││ + seen_runs.add(run_id) ││ + all_runs.append(run) ││ + ││ + # Build extended roster ONLY from players in the merged team's ││ best runs ││ + all_extended_roster_ids = set() ││ + for dungeon_slug, run_data in best_team["best_runs_per_dungeon" ││ ].items(): ││ + # Find players who participated in this best run ││ + for run in all_runs: ││ + if (run["duration"] == run_data["duration"] and ││ + run["completed_timestamp"] == run_data["completed_t ││ imestamp"]): ││ + all_extended_roster_ids.update(run["member_ids"]) ││ + break ││ ││ - # Use the best performance among similar teams ││ - if team["combined_best_time"] < best_combined_time: ││ - best_combined_time = team["combined_best_time"] ││ - best_team_runs = team["best_runs_per_dungeon"] ││ - ││ - # Create merged team - limit core to 3 most consistent players ││ across best runs ││ - # Count participation in best runs across all similar teams ││ - player_best_participation = defaultdict(int) ││ - for team in similar_teams: ││ - for run_data in team["best_runs_per_dungeon"].values(): ││ - # Find actual run to count participation ││ - for dungeon_runs in [team["all_runs"]]: # Use all_runs ││ from team ││ - for run in dungeon_runs: ││ - if run["duration"] == run_data["duration"] and ││ run["completed_timestamp"] == run_data["completed_timestamp"]: ││ - for member_id in run["member_ids"]: ││ - player_best_participation[member_id] += ││ 1 ││ - break ││ - ││ - # Select top 3 most consistent players as merged core ││ - top_players = sorted(player_best_participation.items(), key=lam ││ bda x: x[1], reverse=True)[:3] ││ - merged_core_ids = sorted([player_id for player_id, count in top ││ _players]) ││ - merged_core_info = get_team_info(merged_core_ids, all_members_d ││ ata) ││ - ││ - # Create extended roster from players who appear in the merged ││ team's best runs ││ - merged_extended_roster_ids = set() ││ - for run_data in best_team_runs.values(): ││ - # Find players who participated in this best run across all ││ similar teams ││ - for team in similar_teams: ││ - for run in team["all_runs"]: ││ - if run["duration"] == run_data["duration"] and run[ ││ "completed_timestamp"] == run_data["completed_timestamp"]: ││ - merged_extended_roster_ids.update(run["member_i ││ ds"]) ││ - break ││ - ││ + # Create merged extended roster ││ merged_extended_roster = [] ││ - for member_id in sorted(merged_extended_roster_ids): ││ - if member_id in all_members_data: ││ - member_data = all_members_data[member_id] ││ + for player_id in sorted(all_extended_roster_ids): ││ + if player_id in all_members_data: ││ + member_data = all_members_data[player_id] ││ roster_member = { ││ "name": member_data["name"], ││ "realm_slug": member_data["realm_slug"] ││ } ││ - # Add spec info if available (use first spec if multipl ││ e) ││ if member_data.get("specs"): ││ roster_member["spec_id"] = member_data["specs"][0] ││ merged_extended_roster.append(roster_member) ││ ││ - merged_sig = create_team_signature(merged_core_ids) ││ - ││ - merged_team = { ││ - "team_signature": merged_sig, ││ - "core_members": merged_core_info, ││ - "extended_roster": merged_extended_roster, ││ - "dungeons_completed": similar_teams[0]["dungeons_completed" ││ ], ││ - "total_runs": total_runs // len(similar_teams), ││ - "combined_best_time": best_combined_time, ││ - "average_best_time": best_combined_time / similar_teams[0][ ││ "dungeons_completed"], ││ - "regions_played": list(regions_played), ││ - "best_runs_per_dungeon": best_team_runs, ││ - "all_runs": sorted(all_runs, key=lambda x: x["duration"]) ││ - } ││ + # Use best performing team as base and update with merged data ││ + merged_team = best_team.copy() ││ + merged_team["extended_roster"] = merged_extended_roster ││ + merged_team["total_runs"] = len(all_runs) # Use actual total u ││ nique runs ││ + merged_team["regions_played"] = list(regions_played) ││ + merged_team["all_runs"] = sorted(all_runs, key=lambda x: x["dur ││ ation"]) ││ ││ - # Validate that merged core players appear in most of their bes ││ t runs ││ - core_participation_count = 0 ││ - total_best_runs = len(best_team_runs) ││ - failed_runs = [] ││ - ││ - for dungeon_slug, run_data in best_team_runs.items(): ││ - # Find actual run to check core participation ││ - run_found = False ││ - for team in similar_teams: ││ - for run in team["all_runs"]: ││ - if run["duration"] == run_data["duration"] and run[ ││ "completed_timestamp"] == run_data["completed_timestamp"]: ││ - # Count how many core members participated in t ││ his run ││ - core_in_run = sum(1 for core_id in merged_core_ ││ ids if core_id in run["member_ids"]) ││ - if core_in_run >= 2: # At least 2 of 3 core me ││ mbers ││ - core_participation_count += 1 ││ - else: ││ - failed_runs.append(f"{dungeon_slug}: {core_ ││ in_run}/3 core members") ││ - run_found = True ││ - break ││ - if run_found: ││ - break ││ - ││ - core_participation_rate = core_participation_count / total_best ││ _runs if total_best_runs > 0 else 0 ││ - if core_participation_rate >= 0.85 and len(failed_runs) <= 1: ││ # Allow max 1 failed run ││ - deduplicated.append(merged_team) ││ - else: ││ - print(f"Warning: Rejecting merged team {merged_sig} - core ││ players only appear in {core_participation_rate:.1%} of best runs. Failed runs: {f ││ ailed_runs}") ││ - # Add individual teams instead of the problematic merged te ││ am ││ - deduplicated.extend(similar_teams) ││ + deduplicated.append(merged_team) ││ ││ return deduplicated ││ ││ @@ -270,8 +212,8 @@ ││ print("Loading global rankings...") ││ global_rankings = load_global_rankings() ││ ││ - # first pass: collect all runs and group by extended rosters ││ - roster_runs = defaultdict(lambda: defaultdict(list)) ││ + # Collect all runs and track player data ││ + all_runs = [] ││ available_dungeons = set() ││ all_members_data = {} ││ ││ @@ -285,9 +227,7 @@ ││ ││ print(f"Found {len(leaderboard_files)} leaderboard files to analyze.") ││ ││ - # First pass: collect all runs and identify unique extended rosters ││ - extended_rosters = {} # roster_sig -> set of all player_ids who have r ││ un together ││ - ││ + # Collect all runs ││ for file_path in leaderboard_files: ││ path = Path(file_path) ││ parts = path.parts ││ @@ -347,58 +287,50 @@ ││ "member_names": [get_player_name(m) for m in members] ││ } ││ ││ - # Track extended rosters - players who run together consistentl ││ y ││ - # Use stricter criteria to prevent artificial team merging ││ - found_roster = None ││ - current_players = set(member_ids) ││ + all_runs.append(run_data) ││ ││ - # Check if this run matches any existing extended roster ││ - for roster_sig, roster_players in extended_rosters.items(): ││ - overlap = len(current_players.intersection(roster_players)) ││ - overlap_percentage = overlap / len(current_players.union(ro ││ ster_players)) ││ - ││ - # More balanced criteria: Allow extended rosters but preven ││ t mega-merging ││ - # Require 3+ overlap but with minimum 35% similarity to pre ││ vent distant connections ││ - if overlap >= 3 and overlap_percentage >= 0.35: ││ - found_roster = roster_sig ││ - # Add new players to the extended roster ││ - extended_rosters[roster_sig].update(current_players) ││ - break ││ - ││ - # If no matching roster found, create new one ││ - if found_roster is None: ││ - roster_sig = create_team_signature(sorted(member_ids)) ││ - extended_rosters[roster_sig] = current_players.copy() ││ - found_roster = roster_sig ││ - ││ - # Store run for this extended roster ││ - roster_runs[found_roster][dungeon_slug].append(run_data) ││ - ││ - print(f"Identified {len(extended_rosters)} unique extended rosters.") ││ + print(f"Collected {len(all_runs)} total runs across {len(available_dung ││ eons)} dungeons") ││ print(f"Available dungeons: {sorted(available_dungeons)}") ││ ││ - # Second pass: For each extended roster, identify best 3-player core an ││ d performance ││ - print("Analyzing extended rosters to identify consistent 3-player cores ││ ...") ││ + # Generate all possible 3-player cores from runs and track their dungeo ││ n coverage ││ + print("Analyzing 3-player cores for complete dungeon coverage...") ││ + core_runs = defaultdict(lambda: defaultdict(list)) # core_sig -> dunge ││ on -> [runs] ││ + ││ + for run_data in all_runs: ││ + member_ids = run_data["member_ids"] ││ + dungeon_slug = run_data["dungeon_slug"] ││ + ││ + # Generate all possible 3-player combinations from this 5-player ru ││ n ││ + core_combinations = generate_team_cores(run_data["members"]) ││ + ││ + for core_combo in core_combinations: ││ + core_sig = create_team_signature(core_combo) ││ + core_runs[core_sig][dungeon_slug].append(run_data) ││ + ││ + print(f"Generated {len(core_runs)} unique 3-player cores from all runs" ││ ) ││ + ││ + # Filter cores that have complete dungeon coverage ││ qualified_teams = [] ││ - rosters_analyzed = 0 ││ - rosters_with_complete_coverage = 0 ││ + cores_with_complete_coverage = 0 ││ ││ - for roster_sig, dungeon_data in roster_runs.items(): ││ - rosters_analyzed += 1 ││ + for core_sig, dungeon_data in core_runs.items(): ││ dungeons_completed = set(dungeon_data.keys()) ││ - ││ - # REQUIREMENT: Roster must have runs in ALL available dungeons ││ + ││ + # REQUIREMENT: Core must have runs in ALL available dungeons ││ if dungeons_completed != available_dungeons: ││ continue ││ + ││ + cores_with_complete_coverage += 1 ││ + ││ + # Get core player IDs ││ + core_ids = sorted([int(x) for x in core_sig.split("-")]) ││ ││ - rosters_with_complete_coverage += 1 ││ - ││ - # Find best run for each dungeon and track player participation in ││ best runs ││ + # Find best run for each dungeon containing this core ││ best_runs_per_dungeon = {} ││ - player_participation_in_best = defaultdict(int) ││ - ││ + total_runs = 0 ││ + ││ for dungeon_slug, runs in dungeon_data.items(): ││ - # Deduplicate runs within this dungeon (same run across differe ││ nt realms) ││ + # Deduplicate runs within this dungeon ││ unique_runs = [] ││ seen_dungeon_runs = set() ││ for run in runs: ││ @@ -408,56 +340,57 @@ ││ seen_dungeon_runs.add(run_id) ││ unique_runs.append(run) ││ ││ - # Sort unique runs by duration to get best time for this roster ││ in this dungeon ││ + total_runs += len(unique_runs) ││ + ││ + # Sort by duration to get best time ││ sorted_runs = sorted(unique_runs, key=lambda x: x["duration"]) ││ best_run = sorted_runs[0] ││ ││ - # Look up global ranking for this run ││ + # Look up global ranking ││ global_ranking = lookup_global_ranking(best_run, global_ranking ││ s) ││ ││ best_runs_per_dungeon[dungeon_slug] = { ││ "duration": best_run["duration"], ││ "dungeon_name": best_run["dungeon_name"], ││ - "ranking": global_ranking, # Use global ranking instead of ││ realm ranking ││ + "ranking": global_ranking, ││ "completed_timestamp": best_run["completed_timestamp"], ││ "region": best_run["region"], ││ "realm_slug": best_run["realm_slug"], ││ "members": best_run["member_names"] ││ } ││ ││ - # Track which players appear in the best runs (for core identif ││ ication) ││ - for player_id in best_run["member_ids"]: ││ - player_participation_in_best[player_id] += 1 ││ - ││ - # Identify the 3-player core: players who appear in the most best r ││ uns ││ - total_dungeons = len(dungeons_completed) ││ - participation_sorted = sorted(player_participation_in_best.items(), ││ key=lambda x: x[1], reverse=True) ││ - ││ - # Take top 3 players who appear in the most best runs as the core ││ - if len(participation_sorted) >= 3: ││ - core_ids = sorted([pid for pid, count in participation_sorted[: ││ 3]]) ││ - else: ││ - continue # Not enough consistent players ││ - ││ - # Get extended roster (only players who appear in the best runs per ││ dungeon) ││ - extended_roster_ids = set() ││ - for run_data in best_runs_per_dungeon.values(): ││ - # Find the actual run to get member_ids ││ - for dungeon_slug, runs in dungeon_data.items(): ││ - for run in runs: ││ - if run["duration"] == run_data["duration"] and run["com ││ pleted_timestamp"] == run_data["completed_timestamp"]: ││ - extended_roster_ids.update(run["member_ids"]) ││ - break ││ - ││ - # REQUIREMENT: Roster must have minimum number of total runs ││ - total_runs = sum(len(runs) for runs in dungeon_data.values()) ││ + # REQUIREMENT: Must have minimum number of total runs ││ if total_runs < MIN_TEAM_RUNS: ││ continue ││ ││ + # Build extended roster ONLY from players who appear in best runs ││ + extended_roster_ids = set() ││ + all_team_runs = [] ││ + seen_runs = set() ││ + regions_played = set() ││ + ││ + # First, collect ONLY the players from best runs per dungeon ││ + for dungeon_slug, run_data in best_runs_per_dungeon.items(): ││ + # Find the actual best run to get member_ids ││ + for run in dungeon_data[dungeon_slug]: ││ + if run["duration"] == run_data["duration"] and run["complet ││ ed_timestamp"] == run_data["completed_timestamp"]: ││ + extended_roster_ids.update(run["member_ids"]) ││ + break ││ + ││ + # Then collect all runs for this core (for statistics, not for rost ││ er) ││ + for run in core_runs[core_sig].values(): ││ + for single_run in run: ││ + regions_played.add(single_run["region"]) ││ + member_names_sorted = tuple(sorted(single_run["member_names ││ "])) ││ + run_id = (single_run["duration"], single_run["completed_tim ││ estamp"], member_names_sorted) ││ + if run_id not in seen_runs: ││ + seen_runs.add(run_id) ││ + all_team_runs.append(single_run) ││ + ││ # Create core team info ││ core_info = get_team_info(core_ids, all_members_data) ││ ││ - # Create extended roster info (only players who contributed to best ││ times) ││ + # Create extended roster ││ extended_roster = [] ││ for player_id in sorted(extended_roster_ids): ││ if player_id in all_members_data: ││ @@ -466,31 +399,13 @@ ││ "name": member_data["name"], ││ "realm_slug": member_data["realm_slug"] ││ } ││ - # Add spec info if available (use first spec if multiple) ││ if member_data.get("specs"): ││ roster_member["spec_id"] = member_data["specs"][0] ││ extended_roster.append(roster_member) ││ ││ - # Calculate combined best time across ALL dungeons ││ + # Calculate combined best time ││ combined_best_time = sum(run["duration"] for run in best_runs_per_d ││ ungeon.values()) ││ ││ - # Calculate team statistics ││ - regions_played = set() ││ - all_team_runs = [] ││ - seen_runs = set() # Track unique runs by (duration, timestamp) ││ - for runs in dungeon_data.values(): ││ - for run in runs: ││ - regions_played.add(run["region"]) ││ - # Create unique identifier for run deduplication (include m ││ ember names for cross-realm duplicates) ││ - member_names_sorted = tuple(sorted(run["member_names"])) ││ - run_id = (run["duration"], run["completed_timestamp"], memb ││ er_names_sorted) ││ - if run_id not in seen_runs: ││ - seen_runs.add(run_id) ││ - all_team_runs.append(run) ││ - ││ - # Use core signature for uniqueness ││ - core_sig = create_team_signature(core_ids) ││ - ││ qualified_teams.append({ ││ "team_signature": core_sig, ││ "core_members": core_info, ││ @@ -505,8 +420,8 @@ ││ }) ││ ││ print(f"Analysis results:") ││ - print(f" Extended rosters analyzed: {rosters_analyzed}") ││ - print(f" Rosters with complete coverage: {rosters_with_complete_covera ││ ge}") ││ + print(f" 3-player cores analyzed: {len(core_runs)}") ││ + print(f" Cores with complete coverage: {cores_with_complete_coverage}" ││ ) ││ print(f" Final qualifying teams: {len(qualified_teams)}") ││ ││ return qualified_teams, all_members_data │└────────────────────────────────────────────────────────────────────────────────────┘
┌─ ...eam-leaderboard-generator.nix ───┐│ diff --git a/nix/apps/team-leaderboard-gener ││ ator.nix b/nix/apps/team-leaderboard-generat ││ or.nix ││ index e508ef3..3f77514 100644 ││ --- a/nix/apps/team-leaderboard-generator.ni ││ x ││ +++ b/nix/apps/team-leaderboard-generator.ni ││ x ││ @@ -109,157 +109,99 @@ ││ return team_info ││ ││ def deduplicate_overlapping_teams(tea ││ ms, all_members_data): ││ - # merge teams that share most of ││ their best times ││ + # Simple rule: No two teams can s ││ hare ANY runs - if they do, they're the same ││ team ││ remaining_teams = teams.copy() ││ deduplicated = [] ││ ││ while remaining_teams: ││ current_team = remaining_team ││ s.pop(0) ││ - current_best_times = current_ ││ team["best_runs_per_dungeon"] ││ - ││ - # Find teams that share most ││ best times with current team ││ - similar_teams = [current_team ││ ] ││ - non_similar = [] ││ + current_run_ids = set() ││ + ││ + # Create unique identifiers f ││ or all runs in current team ││ + for run in current_team["all_ ││ runs"]: ││ + member_names_sorted = tup ││ le(sorted(run["member_names"])) ││ + run_id = (run["duration"] ││ , run["completed_timestamp"], member_names_s ││ orted) ││ + current_run_ids.add(run_i ││ d) ││ + ││ + # Find teams that share ANY r ││ uns with current team ││ + same_teams = [current_team] ││ + non_overlapping = [] ││ ││ for other_team in remaining_t ││ eams: ││ - other_best_times = other_ ││ team["best_runs_per_dungeon"] ││ - ││ - # Count how many dungeons ││ have identical best times ││ - matching_dungeons = 0 ││ - total_dungeons = min(len( ││ current_best_times), len(other_best_times)) ││ - ││ - for dungeon_slug in curre ││ nt_best_times: ││ - if dungeon_slug in ot ││ her_best_times: ││ - if current_best_t ││ imes[dungeon_slug]["duration"] == other_best ││ _times[dungeon_slug]["duration"]: ││ - matching_dung ││ eons += 1 ││ - ││ - # If they share 6+ identi ││ cal times out of 9 dungeons, consider them s ││ ame team ││ - match_percentage = matchi ││ ng_dungeons / total_dungeons if total_dungeo ││ ns > 0 else 0 ││ - if matching_dungeons >= 6 ││ and match_percentage >= 0.6: ││ - similar_teams.append( ││ other_team) ││ + other_run_ids = set() ││ + ││ + # Create unique identifie ││ rs for all runs in other team ││ + for run in other_team["al ││ l_runs"]: ││ + member_names_sorted = ││ tuple(sorted(run["member_names"])) ││ + run_id = (run["durati ││ on"], run["completed_timestamp"], member_nam ││ es_sorted) ││ + other_run_ids.add(run ││ _id) ││ + ││ + # If they share ANY runs, ││ they're the same underlying team ││ + if current_run_ids.inters ││ ection(other_run_ids): ││ + same_teams.append(oth ││ er_team) ││ else: ││ - non_similar.append(ot ││ her_team) ││ + non_overlapping.appen ││ d(other_team) ││ ││ - remaining_teams = non_similar ││ + remaining_teams = non_overlap ││ ping ││ ││ - if len(similar_teams) == 1: ││ - # No similar teams found ││ + if len(same_teams) == 1: ││ + # No overlapping teams fo ││ und, keep as is ││ deduplicated.append(curre ││ nt_team) ││ else: ││ - # Merge similar teams (sa ││ me underlying team with different 3-player c ││ ore perspectives) ││ - # Collect all members and ││ data ││ - all_core_members = set() ││ - all_extended_members = se ││ t() ││ + # Multiple teams share ru ││ ns - merge them into one team ││ + # Use the team with the b ││ est combined time as the representative ││ + best_team = min(same_team ││ s, key=lambda t: t["combined_best_time"]) ││ + ││ + # Merge all data from ove ││ rlapping teams ││ + # BUT only include player ││ s who appear in the BEST runs ││ all_runs = [] ││ regions_played = set() ││ total_runs = 0 ││ - best_combined_time = floa ││ t('inf') ││ - best_team_runs = None ││ - ││ - for team in similar_teams ││ : ││ - # Collect core member ││ s ││ - core_ids = list(map(i ││ nt, team["team_signature"].split("-"))) ││ - all_core_members.upda ││ te(core_ids) ││ - ││ - # Collect extended ro ││ ster ││ - for member in team["e ││ xtended_roster"]: ││ - all_extended_memb ││ ers.add(member["name"] + "@" + member["realm ││ _slug"]) ││ + seen_runs = set() ││ ││ - # Collect runs and ot ││ her data ││ - all_runs.extend(team[ ││ "all_runs"]) ││ + for team in same_teams: ││ + # Collect all unique ││ runs ││ regions_played.update ││ (team["regions_played"]) ││ total_runs += team["t ││ otal_runs"] ││ + ││ + for run in team["all_ ││ runs"]: ││ + member_names_sort ││ ed = tuple(sorted(run["member_names"])) ││ + run_id = (run["du ││ ration"], run["completed_timestamp"], member ││ _names_sorted) ││ + if run_id not in ││ seen_runs: ││ + seen_runs.add ││ (run_id) ││ + all_runs.appe ││ nd(run) ││ + ││ + # Build extended roster O ││ NLY from players in the merged team's best r ││ uns ││ + all_extended_roster_ids = ││ set() ││ + for dungeon_slug, run_dat ││ a in best_team["best_runs_per_dungeon"].item ││ s(): ││ + # Find players who pa ││ rticipated in this best run ││ + for run in all_runs: ││ + if (run["duration ││ "] == run_data["duration"] and ││ + run["complete ││ d_timestamp"] == run_data["completed_timesta ││ mp"]): ││ + all_extended_ ││ roster_ids.update(run["member_ids"]) ││ + break ││ ││ - # Use the best perfor ││ mance among similar teams ││ - if team["combined_bes ││ t_time"] < best_combined_time: ││ - best_combined_tim ││ e = team["combined_best_time"] ││ - best_team_runs = ││ team["best_runs_per_dungeon"] ││ - ││ - # Create merged team - li ││ mit core to 3 most consistent players across ││ best runs ││ - # Count participation in ││ best runs across all similar teams ││ - player_best_participation ││ = defaultdict(int) ││ - for team in similar_teams ││ : ││ - for run_data in team[ ││ "best_runs_per_dungeon"].values(): ││ - # Find actual run ││ to count participation ││ - for dungeon_runs ││ in [team["all_runs"]]: # Use all_runs from ││ team ││ - for run in du ││ ngeon_runs: ││ - if run["d ││ uration"] == run_data["duration"] and run["c ││ ompleted_timestamp"] == run_data["completed_ ││ timestamp"]: ││ - for m ││ ember_id in run["member_ids"]: ││ - p ││ layer_best_participation[member_id] += 1 ││ - break ││ - ││ - # Select top 3 most consi ││ stent players as merged core ││ - top_players = sorted(play ││ er_best_participation.items(), key=lambda x: ││ x[1], reverse=True)[:3] ││ - merged_core_ids = sorted( ││ [player_id for player_id, count in top_playe ││ rs]) ││ - merged_core_info = get_te ││ am_info(merged_core_ids, all_members_data) ││ - ││ - # Create extended roster ││ from players who appear in the merged team's ││ best runs ││ - merged_extended_roster_id ││ s = set() ││ - for run_data in best_team ││ _runs.values(): ││ - # Find players who pa ││ rticipated in this best run across all simil ││ ar teams ││ - for team in similar_t ││ eams: ││ - for run in team[" ││ all_runs"]: ││ - if run["durat ││ ion"] == run_data["duration"] and run["compl ││ eted_timestamp"] == run_data["completed_time ││ stamp"]: ││ - merged_ex ││ tended_roster_ids.update(run["member_ids"]) ││ - break ││ - ││ + # Create merged extended ││ roster ││ merged_extended_roster = ││ [] ││ - for member_id in sorted(m ││ erged_extended_roster_ids): ││ - if member_id in all_m ││ embers_data: ││ - member_data = all ││ _members_data[member_id] ││ + for player_id in sorted(a ││ ll_extended_roster_ids): ││ + if player_id in all_m ││ embers_data: ││ + member_data = all ││ _members_data[player_id] ││ roster_member = { ││ "name": membe ││ r_data["name"], ││ "realm_slug": ││ member_data["realm_slug"] ││ } ││ - # Add spec info i ││ f available (use first spec if multiple) ││ if member_data.ge ││ t("specs"): ││ roster_member ││ ["spec_id"] = member_data["specs"][0] ││ merged_extended_r ││ oster.append(roster_member) ││ ││ - merged_sig = create_team_ ││ signature(merged_core_ids) ││ - ││ - merged_team = { ││ - "team_signature": mer ││ ged_sig, ││ - "core_members": merge ││ d_core_info, ││ - "extended_roster": me ││ rged_extended_roster, ││ - "dungeons_completed": ││ similar_teams[0]["dungeons_completed"], ││ - "total_runs": total_r ││ uns // len(similar_teams), ││ - "combined_best_time": ││ best_combined_time, ││ - "average_best_time": ││ best_combined_time / similar_teams[0]["dunge ││ ons_completed"], ││ - "regions_played": lis ││ t(regions_played), ││ - "best_runs_per_dungeo ││ n": best_team_runs, ││ - "all_runs": sorted(al ││ l_runs, key=lambda x: x["duration"]) ││ - } ││ + # Use best performing tea ││ m as base and update with merged data ││ + merged_team = best_team.c ││ opy() ││ + merged_team["extended_ros ││ ter"] = merged_extended_roster ││ + merged_team["total_runs"] ││ = len(all_runs) # Use actual total unique ││ runs ││ + merged_team["regions_play ││ ed"] = list(regions_played) ││ + merged_team["all_runs"] = ││ sorted(all_runs, key=lambda x: x["duration" ││ ]) ││ ││ - # Validate that merged co ││ re players appear in most of their best runs ││ - core_participation_count ││ = 0 ││ - total_best_runs = len(bes ││ t_team_runs) ││ - failed_runs = [] ││ - ││ - for dungeon_slug, run_dat ││ a in best_team_runs.items(): ││ - # Find actual run to ││ check core participation ││ - run_found = False ││ - for team in similar_t ││ eams: ││ - for run in team[" ││ all_runs"]: ││ - if run["durat ││ ion"] == run_data["duration"] and run["compl ││ eted_timestamp"] == run_data["completed_time ││ stamp"]: ││ - # Count h ││ ow many core members participated in this ru ││ n ││ - core_in_r ││ un = sum(1 for core_id in merged_core_ids if ││ core_id in run["member_ids"]) ││ - if core_i ││ n_run >= 2: # At least 2 of 3 core members ││ - core_ ││ participation_count += 1 ││ - else: ││ - faile ││ d_runs.append(f"{dungeon_slug}: {core_in_run ││ }/3 core members") ││ - run_found ││ = True ││ - break ││ - if run_found: ││ - break ││ - ││ - core_participation_rate = ││ core_participation_count / total_best_runs ││ if total_best_runs > 0 else 0 ││ - if core_participation_rat ││ e >= 0.85 and len(failed_runs) <= 1: # Allo ││ w max 1 failed run ││ - deduplicated.append(m ││ erged_team) ││ - else: ││ - print(f"Warning: Reje ││ cting merged team {merged_sig} - core player ││ s only appear in {core_participation_rate:.1 ││ %} of best runs. Failed runs: {failed_runs}" ││ ) ││ - # Add individual team ││ s instead of the problematic merged team ││ - deduplicated.extend(s ││ imilar_teams) ││ + deduplicated.append(merge ││ d_team) ││ ││ return deduplicated ││ ││ @@ -270,8 +212,8 @@ ││ print("Loading global rankings... ││ ") ││ global_rankings = load_global_ran ││ kings() ││ ││ - # first pass: collect all runs an ││ d group by extended rosters ││ - roster_runs = defaultdict(lambda: ││ defaultdict(list)) ││ + # Collect all runs and track play ││ er data ││ + all_runs = [] ││ available_dungeons = set() ││ all_members_data = {} ││ ││ @@ -285,9 +227,7 @@ ││ ││ print(f"Found {len(leaderboard_fi ││ les)} leaderboard files to analyze.") ││ ││ - # First pass: collect all runs an ││ d identify unique extended rosters ││ - extended_rosters = {} # roster_s ││ ig -> set of all player_ids who have run tog ││ ether ││ - ││ + # Collect all runs ││ for file_path in leaderboard_file ││ s: ││ path = Path(file_path) ││ parts = path.parts ││ @@ -347,58 +287,50 @@ ││ "member_names": [get_ ││ player_name(m) for m in members] ││ } ││ ││ - # Track extended rosters ││ - players who run together consistently ││ - # Use stricter criteria t ││ o prevent artificial team merging ││ - found_roster = None ││ - current_players = set(mem ││ ber_ids) ││ + all_runs.append(run_data) ││ ││ - # Check if this run match ││ es any existing extended roster ││ - for roster_sig, roster_pl ││ ayers in extended_rosters.items(): ││ - overlap = len(current ││ _players.intersection(roster_players)) ││ - overlap_percentage = ││ overlap / len(current_players.union(roster_p ││ layers)) ││ - ││ - # More balanced crite ││ ria: Allow extended rosters but prevent mega ││ -merging ││ - # Require 3+ overlap ││ but with minimum 35% similarity to prevent d ││ istant connections ││ - if overlap >= 3 and o ││ verlap_percentage >= 0.35: ││ - found_roster = ro ││ ster_sig ││ - # Add new players ││ to the extended roster ││ - extended_rosters[ ││ roster_sig].update(current_players) ││ - break ││ - ││ - # If no matching roster f ││ ound, create new one ││ - if found_roster is None: ││ - roster_sig = create_t ││ eam_signature(sorted(member_ids)) ││ - extended_rosters[rost ││ er_sig] = current_players.copy() ││ - found_roster = roster ││ _sig ││ - ││ - # Store run for this exte ││ nded roster ││ - roster_runs[found_roster] ││ [dungeon_slug].append(run_data) ││ - ││ - print(f"Identified {len(extended_ ││ rosters)} unique extended rosters.") ││ + print(f"Collected {len(all_runs)} ││ total runs across {len(available_dungeons)} ││ dungeons") ││ print(f"Available dungeons: {sort ││ ed(available_dungeons)}") ││ ││ - # Second pass: For each extended ││ roster, identify best 3-player core and perf ││ ormance ││ - print("Analyzing extended rosters ││ to identify consistent 3-player cores...") ││ + # Generate all possible 3-player ││ cores from runs and track their dungeon cove ││ rage ││ + print("Analyzing 3-player cores f ││ or complete dungeon coverage...") ││ + core_runs = defaultdict(lambda: d ││ efaultdict(list)) # core_sig -> dungeon -> ││ [runs] ││ + ││ + for run_data in all_runs: ││ + member_ids = run_data["member ││ _ids"] ││ + dungeon_slug = run_data["dung ││ eon_slug"] ││ + ││ + # Generate all possible 3-pla ││ yer combinations from this 5-player run ││ + core_combinations = generate_ ││ team_cores(run_data["members"]) ││ + ││ + for core_combo in core_combin ││ ations: ││ + core_sig = create_team_si ││ gnature(core_combo) ││ + core_runs[core_sig][dunge ││ on_slug].append(run_data) ││ + ││ + print(f"Generated {len(core_runs) ││ } unique 3-player cores from all runs") ││ + ││ + # Filter cores that have complete ││ dungeon coverage ││ qualified_teams = [] ││ - rosters_analyzed = 0 ││ - rosters_with_complete_coverage = ││ 0 ││ + cores_with_complete_coverage = 0 ││ ││ - for roster_sig, dungeon_data in r ││ oster_runs.items(): ││ - rosters_analyzed += 1 ││ + for core_sig, dungeon_data in cor ││ e_runs.items(): ││ dungeons_completed = set(dung ││ eon_data.keys()) ││ - ││ - # REQUIREMENT: Roster must ha ││ ve runs in ALL available dungeons ││ + ││ + # REQUIREMENT: Core must have ││ runs in ALL available dungeons ││ if dungeons_completed != avai ││ lable_dungeons: ││ continue ││ + ││ + cores_with_complete_coverage ││ += 1 ││ + ││ + # Get core player IDs ││ + core_ids = sorted([int(x) for ││ x in core_sig.split("-")]) ││ ││ - rosters_with_complete_coverag ││ e += 1 ││ - ││ - # Find best run for each dung ││ eon and track player participation in best r ││ uns ││ + # Find best run for each dung ││ eon containing this core ││ best_runs_per_dungeon = {} ││ - player_participation_in_best ││ = defaultdict(int) ││ - ││ + total_runs = 0 ││ + ││ for dungeon_slug, runs in dun ││ geon_data.items(): ││ - # Deduplicate runs within ││ this dungeon (same run across different rea ││ lms) ││ + # Deduplicate runs within ││ this dungeon ││ unique_runs = [] ││ seen_dungeon_runs = set() ││ for run in runs: ││ @@ -408,56 +340,57 @@ ││ seen_dungeon_runs ││ .add(run_id) ││ unique_runs.appen ││ d(run) ││ ││ - # Sort unique runs by dur ││ ation to get best time for this roster in th ││ is dungeon ││ + total_runs += len(unique_ ││ runs) ││ + ││ + # Sort by duration to get ││ best time ││ sorted_runs = sorted(uniq ││ ue_runs, key=lambda x: x["duration"]) ││ best_run = sorted_runs[0] ││ ││ - # Look up global ranking ││ for this run ││ + # Look up global ranking ││ global_ranking = lookup_g ││ lobal_ranking(best_run, global_rankings) ││ ││ best_runs_per_dungeon[dun ││ geon_slug] = { ││ "duration": best_run[ ││ "duration"], ││ "dungeon_name": best_ ││ run["dungeon_name"], ││ - "ranking": global_ran ││ king, # Use global ranking instead of realm ││ ranking ││ + "ranking": global_ran ││ king, ││ "completed_timestamp" ││ : best_run["completed_timestamp"], ││ "region": best_run["r ││ egion"], ││ "realm_slug": best_ru ││ n["realm_slug"], ││ "members": best_run[" ││ member_names"] ││ } ││ ││ - # Track which players app ││ ear in the best runs (for core identificatio ││ n) ││ - for player_id in best_run ││ ["member_ids"]: ││ - player_participation_ ││ in_best[player_id] += 1 ││ - ││ - # Identify the 3-player core: ││ players who appear in the most best runs ││ - total_dungeons = len(dungeons ││ _completed) ││ - participation_sorted = sorted ││ (player_participation_in_best.items(), key=l ││ ambda x: x[1], reverse=True) ││ - ││ - # Take top 3 players who appe ││ ar in the most best runs as the core ││ - if len(participation_sorted) ││ >= 3: ││ - core_ids = sorted([pid fo ││ r pid, count in participation_sorted[:3]]) ││ - else: ││ - continue # Not enough co ││ nsistent players ││ - ││ - # Get extended roster (only p ││ layers who appear in the best runs per dunge ││ on) ││ - extended_roster_ids = set() ││ - for run_data in best_runs_per ││ _dungeon.values(): ││ - # Find the actual run to ││ get member_ids ││ - for dungeon_slug, runs in ││ dungeon_data.items(): ││ - for run in runs: ││ - if run["duration" ││ ] == run_data["duration"] and run["completed ││ _timestamp"] == run_data["completed_timestam ││ p"]: ││ - extended_rost ││ er_ids.update(run["member_ids"]) ││ - break ││ - ││ - # REQUIREMENT: Roster must ha ││ ve minimum number of total runs ││ - total_runs = sum(len(runs) fo ││ r runs in dungeon_data.values()) ││ + # REQUIREMENT: Must have mini ││ mum number of total runs ││ if total_runs < MIN_TEAM_RUNS ││ : ││ continue ││ ││ + # Build extended roster ONLY ││ from players who appear in best runs ││ + extended_roster_ids = set() ││ + all_team_runs = [] ││ + seen_runs = set() ││ + regions_played = set() ││ + ││ + # First, collect ONLY the pla ││ yers from best runs per dungeon ││ + for dungeon_slug, run_data in ││ best_runs_per_dungeon.items(): ││ + # Find the actual best ru ││ n to get member_ids ││ + for run in dungeon_data[d ││ ungeon_slug]: ││ + if run["duration"] == ││ run_data["duration"] and run["completed_tim ││ estamp"] == run_data["completed_timestamp"]: ││ + extended_roster_i ││ ds.update(run["member_ids"]) ││ + break ││ + ││ + # Then collect all runs for t ││ his core (for statistics, not for roster) ││ + for run in core_runs[core_sig ││ ].values(): ││ + for single_run in run: ││ + regions_played.add(si ││ ngle_run["region"]) ││ + member_names_sorted = ││ tuple(sorted(single_run["member_names"])) ││ + run_id = (single_run[ ││ "duration"], single_run["completed_timestamp ││ "], member_names_sorted) ││ + if run_id not in seen ││ _runs: ││ + seen_runs.add(run ││ _id) ││ + all_team_runs.app ││ end(single_run) ││ + ││ # Create core team info ││ core_info = get_team_info(cor ││ e_ids, all_members_data) ││ ││ - # Create extended roster info ││ (only players who contributed to best times ││ ) ││ + # Create extended roster ││ extended_roster = [] ││ for player_id in sorted(exten ││ ded_roster_ids): ││ if player_id in all_membe ││ rs_data: ││ @@ -466,31 +399,13 @@ ││ "name": member_da ││ ta["name"], ││ "realm_slug": mem ││ ber_data["realm_slug"] ││ } ││ - # Add spec info if av ││ ailable (use first spec if multiple) ││ if member_data.get("s ││ pecs"): ││ roster_member["sp ││ ec_id"] = member_data["specs"][0] ││ extended_roster.appen ││ d(roster_member) ││ ││ - # Calculate combined best tim ││ e across ALL dungeons ││ + # Calculate combined best tim ││ e ││ combined_best_time = sum(run[ ││ "duration"] for run in best_runs_per_dungeon ││ .values()) ││ ││ - # Calculate team statistics ││ - regions_played = set() ││ - all_team_runs = [] ││ - seen_runs = set() # Track un ││ ique runs by (duration, timestamp) ││ - for runs in dungeon_data.valu ││ es(): ││ - for run in runs: ││ - regions_played.add(ru ││ n["region"]) ││ - # Create unique ident ││ ifier for run deduplication (include member ││ names for cross-realm duplicates) ││ - member_names_sorted = ││ tuple(sorted(run["member_names"])) ││ - run_id = (run["durati ││ on"], run["completed_timestamp"], member_nam ││ es_sorted) ││ - if run_id not in seen ││ _runs: ││ - seen_runs.add(run ││ _id) ││ - all_team_runs.app ││ end(run) ││ - ││ - # Use core signature for uniq ││ ueness ││ - core_sig = create_team_signat ││ ure(core_ids) ││ - ││ qualified_teams.append({ ││ "team_signature": core_si ││ g, ││ "core_members": core_info ││ , ││ @@ -505,8 +420,8 @@ ││ }) ││ ││ print(f"Analysis results:") ││ - print(f" Extended rosters analyz ││ ed: {rosters_analyzed}") ││ - print(f" Rosters with complete c ││ overage: {rosters_with_complete_coverage}") ││ + print(f" 3-player cores analyzed ││ : {len(core_runs)}") ││ + print(f" Cores with complete cov ││ erage: {cores_with_complete_coverage}") ││ print(f" Final qualifying teams: ││ {len(qualified_teams)}") ││ ││ return qualified_teams, all_membe ││ rs_data │└──────────────────────────────────────────────┘