# retoor <retoor@molodetz.nl>
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