""" End-of-Season Migration Pipeline (Luân Hồi Chuyển Sinh) for FreeExile. Orchestrates the atomic, non-destructive migration of expiring seasonal leagues to Standard: 1. Freezes trade and marketplace activity. 2. Cancels active listings on Universal Consignment Vault and refunds items. 3. Migrates shared stash tabs into Standard as 'Remove-Only' tabs. 4. Updates character realm affiliation from league_id to 'standard'. 5. Snapshots and persists the final Season Leaderboard (Hall of Fame). 6. Initializes and activates the new Season with clean slate economy. """ import time import uuid from typing import Dict, List, Optional, Any from dataclasses import dataclass, field from world.season_league_manager import ( LeagueType, LeagueStatus, SeasonLeagueManager, SeasonCharacter, StashTab, SeasonStashItem, ) from trade.consignment_vault import UniversalConsignmentVault @dataclass(slots=True) class HallOfFameEntry: rank: int character_id: str character_name: str account_id: str martial_class: str level: int snapshot_timestamp_ms: int @dataclass(slots=True) class MigrationReport: success: bool expiring_league_id: str characters_migrated: int vault_listings_canceled: int remove_only_tabs_created: int leaderboard_snapshot_entries: int next_season_id: Optional[str] = None migration_duration_ms: float = 0.0 errors: List[str] = field(default_factory=list) class SeasonMigrationPipeline: """ Automated Migration Orchestrator for Seasonal Big Updates. Guarantees zero-duplication, atomic settlement, and zero item loss. """ def __init__( self, league_manager: SeasonLeagueManager, vault: Optional[UniversalConsignmentVault] = None ) -> None: self.league_manager: SeasonLeagueManager = league_manager self.vault: Optional[UniversalConsignmentVault] = vault # league_id -> List[HallOfFameEntry] self.archived_leaderboards: Dict[str, List[HallOfFameEntry]] = {} def execute_season_migration( self, expiring_league_id: str, next_season_id: Optional[str] = None, next_season_name: Optional[str] = None, next_season_version: str = "v1.1.0", next_season_mechanics: Optional[Dict[str, str]] = None ) -> MigrationReport: start_time = time.time() errors: List[str] = [] expiring_league = self.league_manager.get_league(expiring_league_id) if not expiring_league: raise KeyError(f"Expiring league '{expiring_league_id}' does not exist") if expiring_league.league_type == LeagueType.STANDARD: raise ValueError("Standard realm cannot be expired or migrated") # ========================================================================= # PHASE 1: FREEZE LEAGUE # ========================================================================= self.league_manager.set_league_status(expiring_league_id, LeagueStatus.FREEZING) # ========================================================================= # PHASE 2: CANCEL & REFUND CONSIGNMENT VAULT LISTINGS # ========================================================================= canceled_vault_count = 0 if self.vault: canceled_vault_count = self._refund_vault_listings(expiring_league_id) # ========================================================================= # PHASE 3: ENTER MIGRATING STATE # ========================================================================= self.league_manager.set_league_status(expiring_league_id, LeagueStatus.MIGRATING) # ========================================================================= # PHASE 4: STASH CONSOLIDATION AS 'REMOVE-ONLY' TABS IN STANDARD # ========================================================================= remove_only_tabs_count = self._migrate_stashes_to_standard(expiring_league_id, expiring_league.display_name) # ========================================================================= # PHASE 5: CHARACTER REALM REASSIGNMENT TO STANDARD # ========================================================================= characters_migrated = self._migrate_characters_to_standard(expiring_league_id) # ========================================================================= # PHASE 6: LEADERBOARD SNAPSHOT & HALL OF FAME # ========================================================================= ladder_entries = self._archive_leaderboard(expiring_league_id) # ========================================================================= # PHASE 7: ARCHIVE EXPIRING LEAGUE # ========================================================================= self.league_manager.set_league_status(expiring_league_id, LeagueStatus.ARCHIVED) # ========================================================================= # PHASE 8: INITIALIZE & ACTIVATE NEW SEASON (IF SPECIFIED) # ========================================================================= if next_season_id: display_name = next_season_name or f"Mùa Giải Mới: {next_season_id}" self.league_manager.create_league( league_id=next_season_id, display_name=display_name, league_type=LeagueType.CHALLENGE, status=LeagueStatus.ACTIVE, version_requirement=next_season_version, mechanic_flags=next_season_mechanics or {} ) duration_ms = (time.time() - start_time) * 1000.0 return MigrationReport( success=len(errors) == 0, expiring_league_id=expiring_league_id, characters_migrated=characters_migrated, vault_listings_canceled=canceled_vault_count, remove_only_tabs_created=remove_only_tabs_count, leaderboard_snapshot_entries=len(ladder_entries), next_season_id=next_season_id, migration_duration_ms=duration_ms, errors=errors ) def _refund_vault_listings(self, expiring_league_id: str) -> int: """ Finds all active listings, marks them closed, and refunds items to the seller's Standard stash or temporary migration holding. """ if not self.vault: return 0 canceled_count = 0 for listing in list(self.vault.listings.values()): if not listing.is_sold: listing.is_sold = True canceled_count += 1 # Clean from buyout engine if present if hasattr(self.vault, "buyout_engine") and self.vault.buyout_engine: seller_stash = self.vault.buyout_engine.accounts.get(listing.seller_id) if seller_stash and listing.item.item_uuid in seller_stash.items: del seller_stash.items[listing.item.item_uuid] # Deposit back as stash item in Standard std_key = (listing.seller_id, "standard") if std_key not in self.league_manager.stashes: self.league_manager.stashes[std_key] = [ StashTab( tab_id=f"stash_{uuid.uuid4().hex[:8]}", account_id=listing.seller_id, league_id="standard", name="Kho Chuẩn Standard", is_remove_only=False ) ] target_tab = self.league_manager.stashes[std_key][0] refund_item = SeasonStashItem( item_uuid=listing.item.item_uuid, item_name=listing.item.item_name, item_level=listing.item.item_level, rarity=listing.item.rarity.value if hasattr(listing.item.rarity, "value") else str(listing.item.rarity) ) target_tab.items[refund_item.item_uuid] = refund_item return canceled_count def _migrate_stashes_to_standard(self, expiring_league_id: str, league_display_name: str) -> int: created_tabs_count = 0 # Iterate over all account stashes matching expiring_league_id for (account_id, league_id), tabs in list(self.league_manager.stashes.items()): if league_id != expiring_league_id: continue std_key = (account_id, "standard") if std_key not in self.league_manager.stashes: self.league_manager.stashes[std_key] = [ StashTab( tab_id=f"stash_{uuid.uuid4().hex[:8]}", account_id=account_id, league_id="standard", name="Kho Chuẩn Standard", is_remove_only=False ) ] for tab in tabs: if tab.is_empty(): continue # Create Remove-Only tab in Standard migrated_tab = StashTab( tab_id=f"ro_{uuid.uuid4().hex[:8]}", account_id=account_id, league_id="standard", name=f"[Chỉ Rút - {league_display_name}] {tab.name}", is_remove_only=True, color_hex="#b45309", capacity=tab.capacity, items=dict(tab.items) # Clone items ) self.league_manager.stashes[std_key].append(migrated_tab) created_tabs_count += 1 # Remove old league stash entries del self.league_manager.stashes[(account_id, league_id)] return created_tabs_count def _migrate_characters_to_standard(self, expiring_league_id: str) -> int: migrated_count = 0 std_league = self.league_manager.get_league("standard") for char in self.league_manager.characters.values(): if char.league_id == expiring_league_id: char.league_id = "standard" migrated_count += 1 if std_league: std_league.active_characters_count += 1 expiring_league = self.league_manager.get_league(expiring_league_id) if expiring_league: expiring_league.active_characters_count = 0 return migrated_count def _archive_leaderboard(self, league_id: str) -> List[HallOfFameEntry]: league_chars = [c for c in self.league_manager.characters.values() if c.league_id == league_id or c.league_id == "standard"] # Sort descending by level, then creation timestamp league_chars.sort(key=lambda c: (c.level, -c.created_at_ms), reverse=True) now_ms = int(time.time() * 1000) entries: List[HallOfFameEntry] = [] for rank, char in enumerate(league_chars[:100], start=1): entries.append( HallOfFameEntry( rank=rank, character_id=char.character_id, character_name=char.name, account_id=char.account_id, martial_class=char.martial_class, level=char.level, snapshot_timestamp_ms=now_ms ) ) self.archived_leaderboards[league_id] = entries return entries