# retoor from __future__ import annotations from devplacepy.database import get_table, get_users_by_uids from devplacepy.utils import generate_uid from .. import economy from .common import GameError, _farms, _iso, _lvl, _now, _update_farm def _eras(): return get_table("game_eras") def _era_results(): return get_table("game_era_results") def active_era() -> dict | None: return _eras().find_one(active=1) def active_era_name() -> str | None: era = active_era() return era["name"] if era else None def start_era(name: str, duration_days: int) -> dict: if active_era(): raise GameError("An Era is already running.") now = _now() ends_at = now.replace(microsecond=0) from datetime import timedelta ends_at = ends_at + timedelta(days=max(1, duration_days)) last = list(_eras().find(order_by=["-era_number"])) era_number = (int(last[0]["era_number"]) + 1) if last else 1 row = { "uid": generate_uid(), "era_number": era_number, "name": name, "started_at": _iso(now), "ends_at": _iso(ends_at), "active": 1, "created_at": _iso(now), } _eras().insert(row) now_iso = _iso(now) for farm in _farms().find(): _update_farm( farm["uid"], {"era_coins": 0, "era_harvests": 0, "era_joined_at": now_iso}, ) return row def end_era() -> dict: era = active_era() if not era: raise GameError("No Era is currently running.") farms = [f for f in _farms().find() if _lvl(f, "era_coins") or _lvl(f, "era_harvests")] ranked = sorted( farms, key=lambda f: economy.era_score( _lvl(f, "era_coins"), _lvl(f, "era_harvests"), _lvl(f, "prestige") ), reverse=True, ) now = _iso(_now()) from .cosmetics import _cosmetics era_cosmetics = [c for c in economy.COSMETICS if c.era_key == era["name"]] for rank, farm in enumerate(ranked, start=1): stars_award = economy.era_reward_stars(rank) reward_cosmetic_key = "" if stars_award and era_cosmetics: reward_cosmetic_key = era_cosmetics[(rank - 1) % len(era_cosmetics)].key if not _cosmetics().find_one( user_uid=farm["user_uid"], cosmetic_key=reward_cosmetic_key ): _cosmetics().insert( { "uid": generate_uid(), "user_uid": farm["user_uid"], "cosmetic_key": reward_cosmetic_key, "purchased_at": now, "created_at": now, } ) reset_fields = {"era_coins": 0, "era_harvests": 0} if stars_award: reset_fields["stars"] = _lvl(farm, "stars") + stars_award _update_farm(farm["uid"], reset_fields) _era_results().insert( { "uid": generate_uid(), "era_number": era["era_number"], "user_uid": farm["user_uid"], "rank": rank, "era_score": economy.era_score( _lvl(farm, "era_coins"), _lvl(farm, "era_harvests"), _lvl(farm, "prestige") ), "era_coins_final": _lvl(farm, "era_coins"), "reward_stars": stars_award, "reward_cosmetic_key": reward_cosmetic_key, "created_at": now, } ) _eras().update({"uid": era["uid"], "active": 0}, ["uid"]) return {"era_number": era["era_number"], "participants": len(ranked)} def leaderboard_era(limit: int = 25) -> list[dict]: era = active_era() if not era: return [] farms = sorted( _farms().find(), key=lambda f: economy.era_score( _lvl(f, "era_coins"), _lvl(f, "era_harvests"), _lvl(f, "prestige") ), reverse=True, )[:limit] if not farms: return [] users = get_users_by_uids([f["user_uid"] for f in farms]) entries = [] for rank, farm in enumerate(farms, start=1): owner = users.get(farm["user_uid"]) if not owner: continue entries.append( { "rank": rank, "username": owner.get("username", ""), "level": int(farm.get("level", 1)), "xp": int(farm.get("xp", 0)), "coins": int(farm.get("coins", 0)), "total_harvests": int(farm.get("total_harvests", 0)), "prestige": int(farm.get("prestige") or 0), "score": economy.era_score( _lvl(farm, "era_coins"), _lvl(farm, "era_harvests"), _lvl(farm, "prestige") ), "title": economy.cosmetic_title_name(farm.get("active_title") or ""), } ) return entries