From 4618089cea52e9d0b01b40fed50be23622e99d89 Mon Sep 17 00:00:00 2001 From: retoor Date: Sat, 4 Jul 2026 19:16:33 +0000 Subject: [PATCH] refactor: split game store.py into subpackage (ref.md 3.11) --- devplacepy/services/game/store.py | 948 -------------------- devplacepy/services/game/store/__init__.py | 62 ++ devplacepy/services/game/store/actions.py | 447 +++++++++ devplacepy/services/game/store/common.py | 84 ++ devplacepy/services/game/store/farm.py | 125 +++ devplacepy/services/game/store/quests.py | 56 ++ devplacepy/services/game/store/serialize.py | 300 +++++++ 7 files changed, 1074 insertions(+), 948 deletions(-) delete mode 100644 devplacepy/services/game/store.py create mode 100644 devplacepy/services/game/store/__init__.py create mode 100644 devplacepy/services/game/store/actions.py create mode 100644 devplacepy/services/game/store/common.py create mode 100644 devplacepy/services/game/store/farm.py create mode 100644 devplacepy/services/game/store/quests.py create mode 100644 devplacepy/services/game/store/serialize.py diff --git a/devplacepy/services/game/store.py b/devplacepy/services/game/store.py deleted file mode 100644 index 5ad0d5e4..00000000 --- a/devplacepy/services/game/store.py +++ /dev/null @@ -1,948 +0,0 @@ -# retoor - -from __future__ import annotations - -import json -from datetime import datetime, timedelta, timezone - -from devplacepy.database import get_table -from devplacepy.utils import generate_uid - -from . import economy - - -class GameError(Exception): - pass - - -def _now() -> datetime: - return datetime.now(timezone.utc) - - -def _iso(value: datetime) -> str: - return value.isoformat() - - -def _parse(value: str) -> datetime | None: - if not value: - return None - try: - return datetime.fromisoformat(value) - except ValueError: - return None - - -def _today() -> str: - return _now().date().isoformat() - - -def _parse_date(value: str): - parsed = _parse(value) - return parsed.date() if parsed else None - - -def _farms(): - return get_table("game_farms") - - -def _plots(): - return get_table("game_plots") - - -def _quests(): - return get_table("game_quests") - - -def _steals(): - return get_table("game_steals") - - -def last_steal_at(thief_uid: str, owner_uid: str) -> datetime | None: - latest = None - for row in _steals().find(thief_uid=thief_uid, owner_uid=owner_uid): - stamp = _parse(row.get("stolen_at", "")) - if stamp and (latest is None or stamp > latest): - latest = stamp - return latest - - -def steal_cooldown_remaining(thief_uid: str, owner_uid: str, now: datetime) -> int: - latest = last_steal_at(thief_uid, owner_uid) - if not latest: - return 0 - elapsed = (now - latest).total_seconds() - return max(0, int(economy.STEAL_COOLDOWN_SECONDS - elapsed)) - - -def _lvl(farm: dict, key: str) -> int: - return int(farm.get(key) or 0) - - -PERK_COLUMN = {perk.key: f"perk_{perk.key}" for perk in economy.PERKS} - - -def get_farm(user_uid: str) -> dict | None: - return _farms().find_one(user_uid=user_uid) - - -def ensure_farm(user_uid: str) -> dict: - farm = get_farm(user_uid) - if farm: - return farm - now = _iso(_now()) - uid = generate_uid() - _farms().insert( - { - "uid": uid, - "user_uid": user_uid, - "coins": economy.STARTING_COINS, - "xp": 0, - "level": 1, - "ci_tier": 1, - "plot_count": economy.STARTING_PLOTS, - "total_harvests": 0, - "created_at": now, - "updated_at": now, - } - ) - for slot_index in range(economy.STARTING_PLOTS): - _create_plot(uid, user_uid, slot_index, now) - return get_farm(user_uid) - - -def _create_plot(farm_uid: str, user_uid: str, slot_index: int, now: str) -> None: - _plots().insert( - { - "uid": generate_uid(), - "farm_uid": farm_uid, - "user_uid": user_uid, - "slot_index": slot_index, - "crop_key": "", - "planted_at": "", - "ready_at": "", - "watered_by": "[]", - "created_at": now, - "updated_at": now, - } - ) - - -def get_plots(farm_uid: str) -> list[dict]: - rows = list(_plots().find(farm_uid=farm_uid)) - rows.sort(key=lambda row: row.get("slot_index", 0)) - return rows - - -def _plot_at(farm_uid: str, slot_index: int) -> dict | None: - return _plots().find_one(farm_uid=farm_uid, slot_index=slot_index) - - -def _watered_by(plot: dict) -> list[str]: - try: - value = json.loads(plot.get("watered_by") or "[]") - return value if isinstance(value, list) else [] - except (ValueError, TypeError): - return [] - - -def _plot_state(plot: dict, now: datetime) -> str: - if not plot.get("crop_key"): - return "empty" - ready_at = _parse(plot.get("ready_at", "")) - if ready_at and now >= ready_at: - return "ready" - return "growing" - - -def serialize_plot( - plot: dict, - *, - viewer_uid: str, - owner_uid: str, - now: datetime, - yield_level: int = 0, - prestige: int = 0, - growth_level: int = 0, - ci_tier: int = 1, - legacy_mult_level: int = 0, - legacy_speed_level: int = 0, - defense_level: int = 0, - steal_locked_until: int = 0, -) -> dict: - state = _plot_state(plot, now) - crop = economy.crop_for(plot.get("crop_key", "")) - ready_at = _parse(plot.get("ready_at", "")) - remaining = 0 - if state == "growing" and ready_at: - remaining = max(0, int((ready_at - now).total_seconds())) - watered = _watered_by(plot) - is_owner = viewer_uid == owner_uid - golden = economy.is_golden(plot.get("uid", ""), plot.get("planted_at", "")) - can_water = ( - state == "growing" - and not is_owner - and bool(viewer_uid) - and viewer_uid not in watered - and len(watered) < economy.MAX_WATERS_PER_PLOT - ) - grace = economy.effective_steal_grace(defense_level) - protected = bool(ready_at) and now < ready_at + timedelta(seconds=grace) - eligible = ( - state == "ready" - and not is_owner - and bool(viewer_uid) - and bool(crop) - and ready_at is not None - ) - can_steal = eligible and not protected and steal_locked_until <= 0 - steal_reason = "" - if eligible and protected: - steal_reason = "protected" - elif eligible and steal_locked_until > 0: - steal_reason = "cooldown" - steal_coins = ( - economy.steal_reward_coins( - crop, yield_level, prestige, legacy_mult_level, defense_level - ) - if can_steal - else 0 - ) - fertilize_cost = 0 - if state == "growing" and is_owner and crop: - full_grow = economy.grow_seconds_for( - crop, ci_tier, growth_level, legacy_speed_level - ) - reduce_by = int(remaining * economy.FERTILIZE_FRACTION) - eff_reward = economy.effective_reward_coins( - crop, yield_level, prestige, legacy_mult_level - ) - fertilize_cost = economy.fertilize_click_cost(eff_reward, reduce_by, full_grow) - return { - "slot": plot.get("slot_index", 0), - "state": state, - "crop_key": plot.get("crop_key", ""), - "crop_name": crop.name if crop else "", - "crop_icon": crop.icon if crop else "", - "reward_coins": crop.reward_coins if crop else 0, - "reward_xp": crop.reward_xp if crop else 0, - "ready_at": plot.get("ready_at", ""), - "remaining_seconds": remaining, - "watered_count": len(watered), - "max_waters": economy.MAX_WATERS_PER_PLOT, - "can_water": can_water, - "can_steal": can_steal, - "steal_coins": steal_coins, - "steal_cooldown_seconds": steal_locked_until if steal_reason == "cooldown" else 0, - "steal_reason": steal_reason, - "is_golden": golden and (is_owner or can_steal), - "fertilize_cost": fertilize_cost, - } - - -def serialize_farm( - farm: dict, *, viewer: dict | None, owner: dict, now: datetime | None = None -) -> dict: - now = now or _now() - viewer_uid = viewer["uid"] if viewer else "" - owner_uid = owner["uid"] - is_owner = viewer_uid == owner_uid - if is_owner and _lvl(farm, "legacy_autoharvest") > 0: - farm = _auto_harvest(farm, owner_uid, now) - prestige = _lvl(farm, "prestige") - growth_level = _lvl(farm, "perk_growth") - discount_level = _lvl(farm, "perk_discount") - yield_level = _lvl(farm, "perk_yield") - legacy_mult_level = _lvl(farm, "legacy_multiplier") - legacy_speed_level = _lvl(farm, "legacy_speed") - defense_level = _lvl(farm, "legacy_defense") - ci_tier = int(farm.get("ci_tier", 1)) - steal_locked_until = ( - steal_cooldown_remaining(viewer_uid, owner_uid, now) - if viewer_uid and not is_owner - else 0 - ) - plots = get_plots(farm["uid"]) - serialized_plots = [ - serialize_plot( - plot, - viewer_uid=viewer_uid, - owner_uid=owner_uid, - now=now, - yield_level=yield_level, - prestige=prestige, - growth_level=growth_level, - ci_tier=ci_tier, - legacy_mult_level=legacy_mult_level, - legacy_speed_level=legacy_speed_level, - defense_level=defense_level, - steal_locked_until=steal_locked_until, - ) - for plot in plots - ] - progress = economy.level_progress(int(farm.get("xp", 0))) - level = progress["level"] - next_tier = economy.next_ci_tier(ci_tier) - ci_entry = economy.CI_BY_TIER.get(ci_tier) - streak = _lvl(farm, "streak") - daily_available = is_owner and _daily_available(farm, now) - return { - "owner_username": owner.get("username", ""), - "owner_uid": owner_uid, - "is_owner": is_owner, - "coins": int(farm.get("coins", 0)), - "xp": int(farm.get("xp", 0)), - "level": level, - "level_into": progress["into_level"], - "level_span": progress["span"], - "level_is_max": progress["is_max"], - "total_harvests": int(farm.get("total_harvests", 0)), - "plot_count": int(farm.get("plot_count", economy.STARTING_PLOTS)), - "max_plots": economy.MAX_PLOTS, - "next_plot_cost": ( - economy.plot_cost(int(farm.get("plot_count", economy.STARTING_PLOTS))) - if int(farm.get("plot_count", economy.STARTING_PLOTS)) < economy.MAX_PLOTS - else 0 - ), - "ci_tier": ci_tier, - "ci_label": ci_entry.label if ci_entry else "", - "ci_speed": economy.ci_speed(ci_tier), - "ci_next_tier": next_tier.tier if next_tier else 0, - "ci_next_label": next_tier.label if next_tier else "", - "ci_next_cost": next_tier.upgrade_cost if next_tier else 0, - "plots": serialized_plots, - "crops": [ - economy.crop_payload( - crop, - ci_tier, - level, - growth_level, - discount_level, - yield_level, - prestige, - legacy_mult_level, - legacy_speed_level, - ) - for crop in economy.CROPS - ], - "prestige": prestige, - "prestige_multiplier": economy.prestige_multiplier(prestige), - "prestige_min_level": economy.PRESTIGE_MIN_LEVEL, - "prestige_available": is_owner and level >= economy.PRESTIGE_MIN_LEVEL, - "streak": streak, - "daily_available": daily_available, - "daily_reward": economy.daily_reward(streak + 1 if daily_available else streak), - "perks": _serialize_perks(farm) if is_owner else [], - "quests": _serialize_quests(owner_uid, now) if is_owner else [], - "stars": _lvl(farm, "stars"), - "legacy": _serialize_legacy(farm) if is_owner else [], - "steal_cooldown_seconds": steal_locked_until, - } - - -def _update_farm(farm_uid: str, fields: dict) -> None: - fields = {**fields, "uid": farm_uid, "updated_at": _iso(_now())} - _farms().update(fields, ["uid"]) - - -def plant(user: dict, slot: int, crop_key: str) -> dict: - farm = ensure_farm(user["uid"]) - crop = economy.crop_for(crop_key) - if not crop: - raise GameError("Unknown crop type.") - progress = economy.level_progress(int(farm.get("xp", 0))) - if crop.min_level > progress["level"]: - raise GameError(f"{crop.name} unlocks at level {crop.min_level}.") - plot = _plot_at(farm["uid"], slot) - if not plot: - raise GameError("That plot does not exist.") - if plot.get("crop_key"): - raise GameError("That plot is already in use.") - coins = int(farm.get("coins", 0)) - cost = economy.effective_plant_cost(crop, _lvl(farm, "perk_discount")) - if coins < cost: - raise GameError("Not enough coins to plant that.") - now = _now() - ci_tier = int(farm.get("ci_tier", 1)) - grow = economy.grow_seconds_for( - crop, ci_tier, _lvl(farm, "perk_growth"), _lvl(farm, "legacy_speed") - ) - ready_at = now + timedelta(seconds=grow) - _plots().update( - { - "uid": plot["uid"], - "crop_key": crop.key, - "planted_at": _iso(now), - "ready_at": _iso(ready_at), - "watered_by": "[]", - "updated_at": _iso(now), - }, - ["uid"], - ) - _update_farm(farm["uid"], {"coins": coins - cost}) - advance_quests(user["uid"], "plant", 1) - return {"slot": slot, "crop": crop.key, "spent": cost} - - -def harvest(user: dict, slot: int) -> dict: - farm = ensure_farm(user["uid"]) - plot = _plot_at(farm["uid"], slot) - if not plot or not plot.get("crop_key"): - raise GameError("That plot is empty.") - now = _now() - if _plot_state(plot, now) != "ready": - raise GameError("That build is still running.") - crop = economy.crop_for(plot.get("crop_key", "")) - if not crop: - raise GameError("Unknown crop type.") - _plots().update( - { - "uid": plot["uid"], - "crop_key": "", - "planted_at": "", - "ready_at": "", - "watered_by": "[]", - "updated_at": _iso(now), - }, - ["uid"], - ) - prestige = _lvl(farm, "prestige") - golden = economy.is_golden(plot.get("uid", ""), plot.get("planted_at", "")) - coins_gain = economy.effective_reward_coins( - crop, _lvl(farm, "perk_yield"), prestige, _lvl(farm, "legacy_multiplier") - ) - if golden: - coins_gain *= economy.GOLDEN_MULTIPLIER - xp_gain = economy.effective_reward_xp(crop, _lvl(farm, "perk_xp")) - new_xp = int(farm.get("xp", 0)) + xp_gain - _update_farm( - farm["uid"], - { - "coins": int(farm.get("coins", 0)) + coins_gain, - "xp": new_xp, - "level": economy.level_for_xp(new_xp), - "total_harvests": int(farm.get("total_harvests", 0)) + 1, - }, - ) - advance_quests(user["uid"], "harvest", 1) - advance_quests(user["uid"], "earn", coins_gain) - return { - "slot": slot, - "crop": crop.key, - "coins": coins_gain, - "xp": xp_gain, - "golden": golden, - } - - -def buy_plot(user: dict) -> dict: - farm = ensure_farm(user["uid"]) - plot_count = int(farm.get("plot_count", economy.STARTING_PLOTS)) - if plot_count >= economy.MAX_PLOTS: - raise GameError("All plots are already unlocked.") - cost = economy.plot_cost(plot_count) - coins = int(farm.get("coins", 0)) - if coins < cost: - raise GameError("Not enough coins for a new plot.") - _create_plot(farm["uid"], user["uid"], plot_count, _iso(_now())) - _update_farm(farm["uid"], {"coins": coins - cost, "plot_count": plot_count + 1}) - return {"plot_count": plot_count + 1, "spent": cost} - - -def upgrade_ci(user: dict) -> dict: - farm = ensure_farm(user["uid"]) - ci_tier = int(farm.get("ci_tier", 1)) - next_tier = economy.next_ci_tier(ci_tier) - if not next_tier: - raise GameError("CI is already at the top tier.") - coins = int(farm.get("coins", 0)) - if coins < next_tier.upgrade_cost: - raise GameError("Not enough coins to upgrade CI.") - _update_farm( - farm["uid"], - {"coins": coins - next_tier.upgrade_cost, "ci_tier": next_tier.tier}, - ) - return {"ci_tier": next_tier.tier, "spent": next_tier.upgrade_cost} - - -def water(visitor: dict, owner: dict, slot: int) -> dict: - if visitor["uid"] == owner["uid"]: - raise GameError("You cannot water your own build.") - farm = ensure_farm(owner["uid"]) - plot = _plot_at(farm["uid"], slot) - if not plot or not plot.get("crop_key"): - raise GameError("That plot is empty.") - now = _now() - if _plot_state(plot, now) != "growing": - raise GameError("That build is not running.") - watered = _watered_by(plot) - if visitor["uid"] in watered: - raise GameError("You already watered this build.") - if len(watered) >= economy.MAX_WATERS_PER_PLOT: - raise GameError("This build has been watered enough.") - crop = economy.crop_for(plot.get("crop_key", "")) - ready_at = _parse(plot.get("ready_at", "")) or now - bonus = economy.water_bonus_seconds( - crop, int(farm.get("ci_tier", 1)), _lvl(farm, "perk_growth"), _lvl(farm, "legacy_speed") - ) - new_ready = max(now, ready_at - timedelta(seconds=bonus)) - watered.append(visitor["uid"]) - _plots().update( - { - "uid": plot["uid"], - "ready_at": _iso(new_ready), - "watered_by": json.dumps(watered), - "updated_at": _iso(now), - }, - ["uid"], - ) - visitor_farm = ensure_farm(visitor["uid"]) - new_xp = int(visitor_farm.get("xp", 0)) + economy.WATER_REWARD_XP - _update_farm( - visitor_farm["uid"], - { - "coins": int(visitor_farm.get("coins", 0)) + economy.WATER_REWARD_COINS, - "xp": new_xp, - "level": economy.level_for_xp(new_xp), - }, - ) - advance_quests(visitor["uid"], "water", 1) - return { - "slot": slot, - "bonus_seconds": bonus, - "reward_coins": economy.WATER_REWARD_COINS, - "reward_xp": economy.WATER_REWARD_XP, - } - - -def steal(thief: dict, owner: dict, slot: int) -> dict: - if thief["uid"] == owner["uid"]: - raise GameError("You cannot steal from your own farm.") - farm = ensure_farm(owner["uid"]) - plot = _plot_at(farm["uid"], slot) - if not plot or not plot.get("crop_key"): - raise GameError("That plot is empty.") - now = _now() - if _plot_state(plot, now) != "ready": - raise GameError("That build is not ready.") - crop = economy.crop_for(plot.get("crop_key", "")) - if not crop: - raise GameError("Unknown crop type.") - defense_level = _lvl(farm, "legacy_defense") - ready_at = _parse(plot.get("ready_at", "")) or now - grace = economy.effective_steal_grace(defense_level) - if now < ready_at + timedelta(seconds=grace): - raise GameError("That harvest is still protected.") - cooldown = steal_cooldown_remaining(thief["uid"], owner["uid"], now) - if cooldown > 0: - minutes = max(1, (cooldown + 59) // 60) - raise GameError( - f"You can only raid {owner.get('username', 'this farmer')} " - f"once an hour. Try again in {minutes} min." - ) - _plots().update( - { - "uid": plot["uid"], - "crop_key": "", - "planted_at": "", - "ready_at": "", - "watered_by": "[]", - "updated_at": _iso(now), - }, - ["uid"], - ) - coins_gain = economy.steal_reward_coins( - crop, - _lvl(farm, "perk_yield"), - _lvl(farm, "prestige"), - _lvl(farm, "legacy_multiplier"), - defense_level, - ) - thief_farm = ensure_farm(thief["uid"]) - _update_farm( - thief_farm["uid"], - {"coins": int(thief_farm.get("coins", 0)) + coins_gain}, - ) - _steals().insert( - { - "uid": generate_uid(), - "thief_uid": thief["uid"], - "owner_uid": owner["uid"], - "slot_index": slot, - "crop_key": crop.key, - "coins": coins_gain, - "stolen_at": _iso(now), - "created_at": _iso(now), - } - ) - return { - "slot": slot, - "crop": crop.key, - "coins": coins_gain, - "owner_uid": owner["uid"], - } - - -def leaderboard(limit: int = 25) -> list[dict]: - farms = sorted( - _farms().find(), - key=economy.farm_score, - reverse=True, - )[:limit] - if not farms: - return [] - from devplacepy.database import get_users_by_uids - - users = get_users_by_uids([farm["user_uid"] for farm 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.farm_score(farm), - } - ) - return entries - - -def _daily_available(farm: dict, now: datetime) -> bool: - last = _parse_date(farm.get("last_daily_at") or "") - return last != now.date() - - -def _serialize_perks(farm: dict) -> list[dict]: - perks = [] - for perk in economy.PERKS: - level = _lvl(farm, PERK_COLUMN[perk.key]) - maxed = level >= perk.max_level - perks.append( - { - "key": perk.key, - "name": perk.name, - "icon": perk.icon, - "description": perk.description, - "level": level, - "max_level": perk.max_level, - "cost": 0 if maxed else economy.perk_cost(perk, level), - "maxed": maxed, - "effect": economy.perk_value_text(perk, level), - } - ) - return perks - - -def _serialize_legacy(farm: dict) -> list[dict]: - upgrades = [] - for up in economy.LEGACY_UPGRADES: - level = _lvl(farm, f"legacy_{up.key}") - maxed = level >= up.max_level - upgrades.append( - { - "key": up.key, - "name": up.name, - "icon": up.icon, - "description": up.description, - "level": level, - "max_level": up.max_level, - "cost": 0 if maxed else economy.legacy_cost(up, level), - "maxed": maxed, - "effect": economy.legacy_value_text(up, level), - } - ) - return upgrades - - -def _auto_harvest(farm: dict, owner_uid: str, now: datetime) -> dict: - yield_level = _lvl(farm, "perk_yield") - xp_level = _lvl(farm, "perk_xp") - prestige = _lvl(farm, "prestige") - mult_level = _lvl(farm, "legacy_multiplier") - coins_gain = 0 - xp_gain = 0 - harvested = 0 - for plot in get_plots(farm["uid"]): - if _plot_state(plot, now) != "ready": - continue - crop = economy.crop_for(plot.get("crop_key", "")) - if not crop: - continue - golden = economy.is_golden(plot.get("uid", ""), plot.get("planted_at", "")) - _plots().update( - { - "uid": plot["uid"], - "crop_key": "", - "planted_at": "", - "ready_at": "", - "watered_by": "[]", - "updated_at": _iso(now), - }, - ["uid"], - ) - coins = economy.effective_reward_coins(crop, yield_level, prestige, mult_level) - if golden: - coins *= economy.GOLDEN_MULTIPLIER - coins_gain += coins - xp_gain += economy.effective_reward_xp(crop, xp_level) - harvested += 1 - if not harvested: - return farm - new_xp = int(farm.get("xp", 0)) + xp_gain - _update_farm( - farm["uid"], - { - "coins": int(farm.get("coins", 0)) + coins_gain, - "xp": new_xp, - "level": economy.level_for_xp(new_xp), - "total_harvests": int(farm.get("total_harvests", 0)) + harvested, - }, - ) - advance_quests(owner_uid, "harvest", harvested) - advance_quests(owner_uid, "earn", coins_gain) - return get_farm(owner_uid) or farm - - -def ensure_quests(farm: dict, day: str) -> None: - if _quests().find_one(farm_uid=farm["uid"], day=day): - return - now = _iso(_now()) - for index, quest in enumerate(economy.daily_quests(farm["user_uid"], day)): - _quests().insert( - { - "uid": generate_uid(), - "farm_uid": farm["uid"], - "user_uid": farm["user_uid"], - "day": day, - "slot_index": index, - "kind": quest["kind"], - "label": quest["label"], - "goal": quest["goal"], - "progress": 0, - "reward_coins": quest["reward_coins"], - "reward_xp": quest["reward_xp"], - "claimed": 0, - "created_at": now, - "updated_at": now, - } - ) - - -def _serialize_quests(user_uid: str, now: datetime) -> list[dict]: - farm = get_farm(user_uid) - if not farm: - return [] - day = now.date().isoformat() - ensure_quests(farm, day) - rows = sorted( - _quests().find(farm_uid=farm["uid"], day=day), - key=lambda row: row.get("slot_index", 0), - ) - quests = [] - for row in rows: - goal = int(row.get("goal") or 0) - progress = min(goal, int(row.get("progress") or 0)) - claimed = bool(row.get("claimed")) - quests.append( - { - "kind": row.get("kind", ""), - "label": row.get("label", ""), - "goal": goal, - "progress": progress, - "reward_coins": int(row.get("reward_coins") or 0), - "reward_xp": int(row.get("reward_xp") or 0), - "claimed": claimed, - "can_claim": (not claimed) and goal > 0 and progress >= goal, - } - ) - return quests - - -def advance_quests(user_uid: str, kind: str, amount: int) -> None: - if amount <= 0: - return - try: - farm = get_farm(user_uid) - if not farm: - return - day = _today() - ensure_quests(farm, day) - for row in _quests().find(farm_uid=farm["uid"], day=day, kind=kind): - if row.get("claimed"): - continue - goal = int(row.get("goal") or 0) - progress = min(goal, int(row.get("progress") or 0) + amount) - _quests().update( - {"uid": row["uid"], "progress": progress, "updated_at": _iso(_now())}, - ["uid"], - ) - except Exception: - return - - -def claim_quest(user: dict, kind: str) -> dict: - farm = ensure_farm(user["uid"]) - day = _today() - ensure_quests(farm, day) - row = _quests().find_one(farm_uid=farm["uid"], day=day, kind=kind) - if not row: - raise GameError("No such quest today.") - if row.get("claimed"): - raise GameError("Quest already claimed.") - goal = int(row.get("goal") or 0) - if int(row.get("progress") or 0) < goal: - raise GameError("Quest not complete yet.") - reward_coins = int(row.get("reward_coins") or 0) - reward_xp = int(row.get("reward_xp") or 0) - _quests().update( - {"uid": row["uid"], "claimed": 1, "updated_at": _iso(_now())}, ["uid"] - ) - new_xp = int(farm.get("xp", 0)) + reward_xp - _update_farm( - farm["uid"], - { - "coins": int(farm.get("coins", 0)) + reward_coins, - "xp": new_xp, - "level": economy.level_for_xp(new_xp), - }, - ) - return {"kind": kind, "reward_coins": reward_coins, "reward_xp": reward_xp} - - -def claim_daily(user: dict) -> dict: - farm = ensure_farm(user["uid"]) - now = _now() - if not _daily_available(farm, now): - raise GameError("Daily bonus already claimed today.") - last = _parse_date(farm.get("last_daily_at") or "") - streak = _lvl(farm, "streak") + 1 if last == now.date() - timedelta(days=1) else 1 - reward = economy.daily_reward(streak) - _update_farm( - farm["uid"], - { - "coins": int(farm.get("coins", 0)) + reward, - "streak": streak, - "last_daily_at": _iso(now), - }, - ) - return {"reward": reward, "streak": streak} - - -def upgrade_perk(user: dict, perk_key: str) -> dict: - perk = economy.perk_for(perk_key) - if not perk: - raise GameError("Unknown perk.") - farm = ensure_farm(user["uid"]) - column = PERK_COLUMN[perk.key] - level = _lvl(farm, column) - if level >= perk.max_level: - raise GameError("That perk is maxed out.") - cost = economy.perk_cost(perk, level) - if int(farm.get("coins", 0)) < cost: - raise GameError("Not enough coins for that upgrade.") - _update_farm(farm["uid"], {"coins": int(farm.get("coins", 0)) - cost, column: level + 1}) - return {"perk": perk.key, "level": level + 1, "spent": cost} - - -def upgrade_legacy(user: dict, key: str) -> dict: - upgrade = economy.legacy_for(key) - if not upgrade: - raise GameError("Unknown legacy upgrade.") - farm = ensure_farm(user["uid"]) - column = f"legacy_{upgrade.key}" - level = _lvl(farm, column) - if level >= upgrade.max_level: - raise GameError("That legacy upgrade is maxed out.") - cost = economy.legacy_cost(upgrade, level) - if _lvl(farm, "stars") < cost: - raise GameError("Not enough stars for that legacy upgrade.") - _update_farm( - farm["uid"], {"stars": _lvl(farm, "stars") - cost, column: level + 1} - ) - return {"key": upgrade.key, "level": level + 1, "spent": cost} - - -def prestige(user: dict) -> dict: - farm = ensure_farm(user["uid"]) - level = economy.level_for_xp(int(farm.get("xp", 0))) - if level < economy.PRESTIGE_MIN_LEVEL: - raise GameError( - f"Reach level {economy.PRESTIGE_MIN_LEVEL} to refactor (prestige)." - ) - new_prestige = _lvl(farm, "prestige") + 1 - stars_award = economy.stars_for_refactor(level, _lvl(farm, "prestige")) - base_plots = economy.prestige_base_plots(_lvl(farm, "legacy_plots")) - now = _iso(_now()) - kept_slots = set() - for plot in get_plots(farm["uid"]): - if int(plot.get("slot_index", 0)) >= base_plots: - _plots().delete(uid=plot["uid"]) - else: - kept_slots.add(int(plot.get("slot_index", 0))) - _plots().update( - { - "uid": plot["uid"], - "crop_key": "", - "planted_at": "", - "ready_at": "", - "watered_by": "[]", - "updated_at": now, - }, - ["uid"], - ) - for slot_index in range(base_plots): - if slot_index not in kept_slots: - _create_plot(farm["uid"], user["uid"], slot_index, now) - reset = { - "coins": economy.STARTING_COINS, - "xp": 0, - "level": 1, - "ci_tier": 1, - "plot_count": base_plots, - "prestige": new_prestige, - "stars": _lvl(farm, "stars") + stars_award, - } - for perk in economy.PERKS: - reset[PERK_COLUMN[perk.key]] = 0 - _update_farm(farm["uid"], reset) - return {"prestige": new_prestige, "stars_awarded": stars_award} - - -def fertilize(user: dict, slot: int) -> dict: - farm = ensure_farm(user["uid"]) - plot = _plot_at(farm["uid"], slot) - if not plot or not plot.get("crop_key"): - raise GameError("That plot is empty.") - now = _now() - if _plot_state(plot, now) != "growing": - raise GameError("That build is not running.") - crop = economy.crop_for(plot.get("crop_key", "")) - if not crop: - raise GameError("Unknown crop type.") - ready_at = _parse(plot.get("ready_at", "")) or now - remaining = max(1, int((ready_at - now).total_seconds())) - reduce_by = int(remaining * economy.FERTILIZE_FRACTION) - if reduce_by < 1: - raise GameError("That build is almost ready already.") - full_grow = economy.grow_seconds_for( - crop, int(farm.get("ci_tier", 1)), _lvl(farm, "perk_growth"), _lvl(farm, "legacy_speed") - ) - eff_reward = economy.effective_reward_coins( - crop, _lvl(farm, "perk_yield"), _lvl(farm, "prestige"), _lvl(farm, "legacy_multiplier") - ) - cost = economy.fertilize_click_cost(eff_reward, reduce_by, full_grow) - if int(farm.get("coins", 0)) < cost: - raise GameError("Not enough coins to fertilize.") - new_ready = max(now, ready_at - timedelta(seconds=reduce_by)) - _plots().update( - {"uid": plot["uid"], "ready_at": _iso(new_ready), "updated_at": _iso(now)}, - ["uid"], - ) - _update_farm(farm["uid"], {"coins": int(farm.get("coins", 0)) - cost}) - return {"slot": slot, "spent": cost, "reduced_seconds": reduce_by} diff --git a/devplacepy/services/game/store/__init__.py b/devplacepy/services/game/store/__init__.py new file mode 100644 index 00000000..8173be3e --- /dev/null +++ b/devplacepy/services/game/store/__init__.py @@ -0,0 +1,62 @@ +# retoor + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone + +from devplacepy.database import get_table +from devplacepy.utils import generate_uid + +from .. import economy +from .actions import ( + _auto_harvest, + claim_daily, + claim_quest, + fertilize, + harvest, + plant, + prestige, + steal, + upgrade_legacy, + upgrade_perk, + water, +) +from .common import ( + GameError, + PERK_COLUMN, + _farms, + _iso, + _lvl, + _now, + _parse, + _parse_date, + _plots, + _quests, + _steals, + _today, + _update_farm, + last_steal_at, + steal_cooldown_remaining, +) +from .farm import ( + _create_plot, + _plot_at, + buy_plot, + ensure_farm, + get_farm, + get_plots, + leaderboard, + upgrade_ci, +) +from .quests import advance_quests, ensure_quests +from .serialize import ( + _daily_available, + _plot_state, + _serialize_legacy, + _serialize_perks, + _serialize_quests, + _watered_by, + serialize_farm, + serialize_plot, +) diff --git a/devplacepy/services/game/store/actions.py b/devplacepy/services/game/store/actions.py new file mode 100644 index 00000000..0ffee305 --- /dev/null +++ b/devplacepy/services/game/store/actions.py @@ -0,0 +1,447 @@ +# retoor + +from __future__ import annotations + +import json +from datetime import datetime, timedelta + +from devplacepy.utils import generate_uid + +from .. import economy +from .common import ( + GameError, + PERK_COLUMN, + _iso, + _lvl, + _now, + _parse, + _parse_date, + _plots, + _quests, + _steals, + _today, + _update_farm, + steal_cooldown_remaining, +) +from .farm import _create_plot, _plot_at, ensure_farm, get_farm, get_plots +from .quests import advance_quests, ensure_quests +from .serialize import _daily_available, _plot_state, _watered_by + + +def plant(user: dict, slot: int, crop_key: str) -> dict: + farm = ensure_farm(user["uid"]) + crop = economy.crop_for(crop_key) + if not crop: + raise GameError("Unknown crop type.") + progress = economy.level_progress(int(farm.get("xp", 0))) + if crop.min_level > progress["level"]: + raise GameError(f"{crop.name} unlocks at level {crop.min_level}.") + plot = _plot_at(farm["uid"], slot) + if not plot: + raise GameError("That plot does not exist.") + if plot.get("crop_key"): + raise GameError("That plot is already in use.") + coins = int(farm.get("coins", 0)) + cost = economy.effective_plant_cost(crop, _lvl(farm, "perk_discount")) + if coins < cost: + raise GameError("Not enough coins to plant that.") + now = _now() + ci_tier = int(farm.get("ci_tier", 1)) + grow = economy.grow_seconds_for( + crop, ci_tier, _lvl(farm, "perk_growth"), _lvl(farm, "legacy_speed") + ) + ready_at = now + timedelta(seconds=grow) + _plots().update( + { + "uid": plot["uid"], + "crop_key": crop.key, + "planted_at": _iso(now), + "ready_at": _iso(ready_at), + "watered_by": "[]", + "updated_at": _iso(now), + }, + ["uid"], + ) + _update_farm(farm["uid"], {"coins": coins - cost}) + advance_quests(user["uid"], "plant", 1) + return {"slot": slot, "crop": crop.key, "spent": cost} + + +def harvest(user: dict, slot: int) -> dict: + farm = ensure_farm(user["uid"]) + plot = _plot_at(farm["uid"], slot) + if not plot or not plot.get("crop_key"): + raise GameError("That plot is empty.") + now = _now() + if _plot_state(plot, now) != "ready": + raise GameError("That build is still running.") + crop = economy.crop_for(plot.get("crop_key", "")) + if not crop: + raise GameError("Unknown crop type.") + _plots().update( + { + "uid": plot["uid"], + "crop_key": "", + "planted_at": "", + "ready_at": "", + "watered_by": "[]", + "updated_at": _iso(now), + }, + ["uid"], + ) + prestige = _lvl(farm, "prestige") + golden = economy.is_golden(plot.get("uid", ""), plot.get("planted_at", "")) + coins_gain = economy.effective_reward_coins( + crop, _lvl(farm, "perk_yield"), prestige, _lvl(farm, "legacy_multiplier") + ) + if golden: + coins_gain *= economy.GOLDEN_MULTIPLIER + xp_gain = economy.effective_reward_xp(crop, _lvl(farm, "perk_xp")) + new_xp = int(farm.get("xp", 0)) + xp_gain + _update_farm( + farm["uid"], + { + "coins": int(farm.get("coins", 0)) + coins_gain, + "xp": new_xp, + "level": economy.level_for_xp(new_xp), + "total_harvests": int(farm.get("total_harvests", 0)) + 1, + }, + ) + advance_quests(user["uid"], "harvest", 1) + advance_quests(user["uid"], "earn", coins_gain) + return { + "slot": slot, + "crop": crop.key, + "coins": coins_gain, + "xp": xp_gain, + "golden": golden, + } + + +def water(visitor: dict, owner: dict, slot: int) -> dict: + if visitor["uid"] == owner["uid"]: + raise GameError("You cannot water your own build.") + farm = ensure_farm(owner["uid"]) + plot = _plot_at(farm["uid"], slot) + if not plot or not plot.get("crop_key"): + raise GameError("That plot is empty.") + now = _now() + if _plot_state(plot, now) != "growing": + raise GameError("That build is not running.") + watered = _watered_by(plot) + if visitor["uid"] in watered: + raise GameError("You already watered this build.") + if len(watered) >= economy.MAX_WATERS_PER_PLOT: + raise GameError("This build has been watered enough.") + crop = economy.crop_for(plot.get("crop_key", "")) + ready_at = _parse(plot.get("ready_at", "")) or now + bonus = economy.water_bonus_seconds( + crop, int(farm.get("ci_tier", 1)), _lvl(farm, "perk_growth"), _lvl(farm, "legacy_speed") + ) + new_ready = max(now, ready_at - timedelta(seconds=bonus)) + watered.append(visitor["uid"]) + _plots().update( + { + "uid": plot["uid"], + "ready_at": _iso(new_ready), + "watered_by": json.dumps(watered), + "updated_at": _iso(now), + }, + ["uid"], + ) + visitor_farm = ensure_farm(visitor["uid"]) + new_xp = int(visitor_farm.get("xp", 0)) + economy.WATER_REWARD_XP + _update_farm( + visitor_farm["uid"], + { + "coins": int(visitor_farm.get("coins", 0)) + economy.WATER_REWARD_COINS, + "xp": new_xp, + "level": economy.level_for_xp(new_xp), + }, + ) + advance_quests(visitor["uid"], "water", 1) + return { + "slot": slot, + "bonus_seconds": bonus, + "reward_coins": economy.WATER_REWARD_COINS, + "reward_xp": economy.WATER_REWARD_XP, + } + + +def steal(thief: dict, owner: dict, slot: int) -> dict: + if thief["uid"] == owner["uid"]: + raise GameError("You cannot steal from your own farm.") + farm = ensure_farm(owner["uid"]) + plot = _plot_at(farm["uid"], slot) + if not plot or not plot.get("crop_key"): + raise GameError("That plot is empty.") + now = _now() + if _plot_state(plot, now) != "ready": + raise GameError("That build is not ready.") + crop = economy.crop_for(plot.get("crop_key", "")) + if not crop: + raise GameError("Unknown crop type.") + defense_level = _lvl(farm, "legacy_defense") + ready_at = _parse(plot.get("ready_at", "")) or now + grace = economy.effective_steal_grace(defense_level) + if now < ready_at + timedelta(seconds=grace): + raise GameError("That harvest is still protected.") + cooldown = steal_cooldown_remaining(thief["uid"], owner["uid"], now) + if cooldown > 0: + minutes = max(1, (cooldown + 59) // 60) + raise GameError( + f"You can only raid {owner.get('username', 'this farmer')} " + f"once an hour. Try again in {minutes} min." + ) + _plots().update( + { + "uid": plot["uid"], + "crop_key": "", + "planted_at": "", + "ready_at": "", + "watered_by": "[]", + "updated_at": _iso(now), + }, + ["uid"], + ) + coins_gain = economy.steal_reward_coins( + crop, + _lvl(farm, "perk_yield"), + _lvl(farm, "prestige"), + _lvl(farm, "legacy_multiplier"), + defense_level, + ) + thief_farm = ensure_farm(thief["uid"]) + _update_farm( + thief_farm["uid"], + {"coins": int(thief_farm.get("coins", 0)) + coins_gain}, + ) + _steals().insert( + { + "uid": generate_uid(), + "thief_uid": thief["uid"], + "owner_uid": owner["uid"], + "slot_index": slot, + "crop_key": crop.key, + "coins": coins_gain, + "stolen_at": _iso(now), + "created_at": _iso(now), + } + ) + return { + "slot": slot, + "crop": crop.key, + "coins": coins_gain, + "owner_uid": owner["uid"], + } + + +def _auto_harvest(farm: dict, owner_uid: str, now: datetime) -> dict: + yield_level = _lvl(farm, "perk_yield") + xp_level = _lvl(farm, "perk_xp") + prestige = _lvl(farm, "prestige") + mult_level = _lvl(farm, "legacy_multiplier") + coins_gain = 0 + xp_gain = 0 + harvested = 0 + for plot in get_plots(farm["uid"]): + if _plot_state(plot, now) != "ready": + continue + crop = economy.crop_for(plot.get("crop_key", "")) + if not crop: + continue + golden = economy.is_golden(plot.get("uid", ""), plot.get("planted_at", "")) + _plots().update( + { + "uid": plot["uid"], + "crop_key": "", + "planted_at": "", + "ready_at": "", + "watered_by": "[]", + "updated_at": _iso(now), + }, + ["uid"], + ) + coins = economy.effective_reward_coins(crop, yield_level, prestige, mult_level) + if golden: + coins *= economy.GOLDEN_MULTIPLIER + coins_gain += coins + xp_gain += economy.effective_reward_xp(crop, xp_level) + harvested += 1 + if not harvested: + return farm + new_xp = int(farm.get("xp", 0)) + xp_gain + _update_farm( + farm["uid"], + { + "coins": int(farm.get("coins", 0)) + coins_gain, + "xp": new_xp, + "level": economy.level_for_xp(new_xp), + "total_harvests": int(farm.get("total_harvests", 0)) + harvested, + }, + ) + advance_quests(owner_uid, "harvest", harvested) + advance_quests(owner_uid, "earn", coins_gain) + return get_farm(owner_uid) or farm + + +def claim_quest(user: dict, kind: str) -> dict: + farm = ensure_farm(user["uid"]) + day = _today() + ensure_quests(farm, day) + row = _quests().find_one(farm_uid=farm["uid"], day=day, kind=kind) + if not row: + raise GameError("No such quest today.") + if row.get("claimed"): + raise GameError("Quest already claimed.") + goal = int(row.get("goal") or 0) + if int(row.get("progress") or 0) < goal: + raise GameError("Quest not complete yet.") + reward_coins = int(row.get("reward_coins") or 0) + reward_xp = int(row.get("reward_xp") or 0) + _quests().update( + {"uid": row["uid"], "claimed": 1, "updated_at": _iso(_now())}, ["uid"] + ) + new_xp = int(farm.get("xp", 0)) + reward_xp + _update_farm( + farm["uid"], + { + "coins": int(farm.get("coins", 0)) + reward_coins, + "xp": new_xp, + "level": economy.level_for_xp(new_xp), + }, + ) + return {"kind": kind, "reward_coins": reward_coins, "reward_xp": reward_xp} + + +def claim_daily(user: dict) -> dict: + farm = ensure_farm(user["uid"]) + now = _now() + if not _daily_available(farm, now): + raise GameError("Daily bonus already claimed today.") + last = _parse_date(farm.get("last_daily_at") or "") + streak = _lvl(farm, "streak") + 1 if last == now.date() - timedelta(days=1) else 1 + reward = economy.daily_reward(streak) + _update_farm( + farm["uid"], + { + "coins": int(farm.get("coins", 0)) + reward, + "streak": streak, + "last_daily_at": _iso(now), + }, + ) + return {"reward": reward, "streak": streak} + + +def upgrade_perk(user: dict, perk_key: str) -> dict: + perk = economy.perk_for(perk_key) + if not perk: + raise GameError("Unknown perk.") + farm = ensure_farm(user["uid"]) + column = PERK_COLUMN[perk.key] + level = _lvl(farm, column) + if level >= perk.max_level: + raise GameError("That perk is maxed out.") + cost = economy.perk_cost(perk, level) + if int(farm.get("coins", 0)) < cost: + raise GameError("Not enough coins for that upgrade.") + _update_farm(farm["uid"], {"coins": int(farm.get("coins", 0)) - cost, column: level + 1}) + return {"perk": perk.key, "level": level + 1, "spent": cost} + + +def upgrade_legacy(user: dict, key: str) -> dict: + upgrade = economy.legacy_for(key) + if not upgrade: + raise GameError("Unknown legacy upgrade.") + farm = ensure_farm(user["uid"]) + column = f"legacy_{upgrade.key}" + level = _lvl(farm, column) + if level >= upgrade.max_level: + raise GameError("That legacy upgrade is maxed out.") + cost = economy.legacy_cost(upgrade, level) + if _lvl(farm, "stars") < cost: + raise GameError("Not enough stars for that legacy upgrade.") + _update_farm( + farm["uid"], {"stars": _lvl(farm, "stars") - cost, column: level + 1} + ) + return {"key": upgrade.key, "level": level + 1, "spent": cost} + + +def prestige(user: dict) -> dict: + farm = ensure_farm(user["uid"]) + level = economy.level_for_xp(int(farm.get("xp", 0))) + if level < economy.PRESTIGE_MIN_LEVEL: + raise GameError( + f"Reach level {economy.PRESTIGE_MIN_LEVEL} to refactor (prestige)." + ) + new_prestige = _lvl(farm, "prestige") + 1 + stars_award = economy.stars_for_refactor(level, _lvl(farm, "prestige")) + base_plots = economy.prestige_base_plots(_lvl(farm, "legacy_plots")) + now = _iso(_now()) + kept_slots = set() + for plot in get_plots(farm["uid"]): + if int(plot.get("slot_index", 0)) >= base_plots: + _plots().delete(uid=plot["uid"]) + else: + kept_slots.add(int(plot.get("slot_index", 0))) + _plots().update( + { + "uid": plot["uid"], + "crop_key": "", + "planted_at": "", + "ready_at": "", + "watered_by": "[]", + "updated_at": now, + }, + ["uid"], + ) + for slot_index in range(base_plots): + if slot_index not in kept_slots: + _create_plot(farm["uid"], user["uid"], slot_index, now) + reset = { + "coins": economy.STARTING_COINS, + "xp": 0, + "level": 1, + "ci_tier": 1, + "plot_count": base_plots, + "prestige": new_prestige, + "stars": _lvl(farm, "stars") + stars_award, + } + for perk in economy.PERKS: + reset[PERK_COLUMN[perk.key]] = 0 + _update_farm(farm["uid"], reset) + return {"prestige": new_prestige, "stars_awarded": stars_award} + + +def fertilize(user: dict, slot: int) -> dict: + farm = ensure_farm(user["uid"]) + plot = _plot_at(farm["uid"], slot) + if not plot or not plot.get("crop_key"): + raise GameError("That plot is empty.") + now = _now() + if _plot_state(plot, now) != "growing": + raise GameError("That build is not running.") + crop = economy.crop_for(plot.get("crop_key", "")) + if not crop: + raise GameError("Unknown crop type.") + ready_at = _parse(plot.get("ready_at", "")) or now + remaining = max(1, int((ready_at - now).total_seconds())) + reduce_by = int(remaining * economy.FERTILIZE_FRACTION) + if reduce_by < 1: + raise GameError("That build is almost ready already.") + full_grow = economy.grow_seconds_for( + crop, int(farm.get("ci_tier", 1)), _lvl(farm, "perk_growth"), _lvl(farm, "legacy_speed") + ) + eff_reward = economy.effective_reward_coins( + crop, _lvl(farm, "perk_yield"), _lvl(farm, "prestige"), _lvl(farm, "legacy_multiplier") + ) + cost = economy.fertilize_click_cost(eff_reward, reduce_by, full_grow) + if int(farm.get("coins", 0)) < cost: + raise GameError("Not enough coins to fertilize.") + new_ready = max(now, ready_at - timedelta(seconds=reduce_by)) + _plots().update( + {"uid": plot["uid"], "ready_at": _iso(new_ready), "updated_at": _iso(now)}, + ["uid"], + ) + _update_farm(farm["uid"], {"coins": int(farm.get("coins", 0)) - cost}) + return {"slot": slot, "spent": cost, "reduced_seconds": reduce_by} diff --git a/devplacepy/services/game/store/common.py b/devplacepy/services/game/store/common.py new file mode 100644 index 00000000..65222b9a --- /dev/null +++ b/devplacepy/services/game/store/common.py @@ -0,0 +1,84 @@ +# retoor + +from __future__ import annotations + +from datetime import datetime, timezone + +from devplacepy.database import get_table + +from .. import economy + + +class GameError(Exception): + pass + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(value: datetime) -> str: + return value.isoformat() + + +def _parse(value: str) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def _today() -> str: + return _now().date().isoformat() + + +def _parse_date(value: str): + parsed = _parse(value) + return parsed.date() if parsed else None + + +def _farms(): + return get_table("game_farms") + + +def _plots(): + return get_table("game_plots") + + +def _quests(): + return get_table("game_quests") + + +def _steals(): + return get_table("game_steals") + + +def last_steal_at(thief_uid: str, owner_uid: str) -> datetime | None: + latest = None + for row in _steals().find(thief_uid=thief_uid, owner_uid=owner_uid): + stamp = _parse(row.get("stolen_at", "")) + if stamp and (latest is None or stamp > latest): + latest = stamp + return latest + + +def steal_cooldown_remaining(thief_uid: str, owner_uid: str, now: datetime) -> int: + latest = last_steal_at(thief_uid, owner_uid) + if not latest: + return 0 + elapsed = (now - latest).total_seconds() + return max(0, int(economy.STEAL_COOLDOWN_SECONDS - elapsed)) + + +def _lvl(farm: dict, key: str) -> int: + return int(farm.get(key) or 0) + + +PERK_COLUMN = {perk.key: f"perk_{perk.key}" for perk in economy.PERKS} + + +def _update_farm(farm_uid: str, fields: dict) -> None: + fields = {**fields, "uid": farm_uid, "updated_at": _iso(_now())} + _farms().update(fields, ["uid"]) diff --git a/devplacepy/services/game/store/farm.py b/devplacepy/services/game/store/farm.py new file mode 100644 index 00000000..0a23e2d0 --- /dev/null +++ b/devplacepy/services/game/store/farm.py @@ -0,0 +1,125 @@ +# retoor + +from __future__ import annotations + +from devplacepy.utils import generate_uid + +from .. import economy +from .common import GameError, _farms, _iso, _now, _plots, _update_farm + + +def get_farm(user_uid: str) -> dict | None: + return _farms().find_one(user_uid=user_uid) + + +def ensure_farm(user_uid: str) -> dict: + farm = get_farm(user_uid) + if farm: + return farm + now = _iso(_now()) + uid = generate_uid() + _farms().insert( + { + "uid": uid, + "user_uid": user_uid, + "coins": economy.STARTING_COINS, + "xp": 0, + "level": 1, + "ci_tier": 1, + "plot_count": economy.STARTING_PLOTS, + "total_harvests": 0, + "created_at": now, + "updated_at": now, + } + ) + for slot_index in range(economy.STARTING_PLOTS): + _create_plot(uid, user_uid, slot_index, now) + return get_farm(user_uid) + + +def _create_plot(farm_uid: str, user_uid: str, slot_index: int, now: str) -> None: + _plots().insert( + { + "uid": generate_uid(), + "farm_uid": farm_uid, + "user_uid": user_uid, + "slot_index": slot_index, + "crop_key": "", + "planted_at": "", + "ready_at": "", + "watered_by": "[]", + "created_at": now, + "updated_at": now, + } + ) + + +def get_plots(farm_uid: str) -> list[dict]: + rows = list(_plots().find(farm_uid=farm_uid)) + rows.sort(key=lambda row: row.get("slot_index", 0)) + return rows + + +def _plot_at(farm_uid: str, slot_index: int) -> dict | None: + return _plots().find_one(farm_uid=farm_uid, slot_index=slot_index) + + +def buy_plot(user: dict) -> dict: + farm = ensure_farm(user["uid"]) + plot_count = int(farm.get("plot_count", economy.STARTING_PLOTS)) + if plot_count >= economy.MAX_PLOTS: + raise GameError("All plots are already unlocked.") + cost = economy.plot_cost(plot_count) + coins = int(farm.get("coins", 0)) + if coins < cost: + raise GameError("Not enough coins for a new plot.") + _create_plot(farm["uid"], user["uid"], plot_count, _iso(_now())) + _update_farm(farm["uid"], {"coins": coins - cost, "plot_count": plot_count + 1}) + return {"plot_count": plot_count + 1, "spent": cost} + + +def upgrade_ci(user: dict) -> dict: + farm = ensure_farm(user["uid"]) + ci_tier = int(farm.get("ci_tier", 1)) + next_tier = economy.next_ci_tier(ci_tier) + if not next_tier: + raise GameError("CI is already at the top tier.") + coins = int(farm.get("coins", 0)) + if coins < next_tier.upgrade_cost: + raise GameError("Not enough coins to upgrade CI.") + _update_farm( + farm["uid"], + {"coins": coins - next_tier.upgrade_cost, "ci_tier": next_tier.tier}, + ) + return {"ci_tier": next_tier.tier, "spent": next_tier.upgrade_cost} + + +def leaderboard(limit: int = 25) -> list[dict]: + farms = sorted( + _farms().find(), + key=economy.farm_score, + reverse=True, + )[:limit] + if not farms: + return [] + from devplacepy.database import get_users_by_uids + + users = get_users_by_uids([farm["user_uid"] for farm 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.farm_score(farm), + } + ) + return entries diff --git a/devplacepy/services/game/store/quests.py b/devplacepy/services/game/store/quests.py new file mode 100644 index 00000000..38f34724 --- /dev/null +++ b/devplacepy/services/game/store/quests.py @@ -0,0 +1,56 @@ +# retoor + +from __future__ import annotations + +from devplacepy.utils import generate_uid + +from .. import economy +from .common import _iso, _now, _quests, _today +from .farm import get_farm + + +def ensure_quests(farm: dict, day: str) -> None: + if _quests().find_one(farm_uid=farm["uid"], day=day): + return + now = _iso(_now()) + for index, quest in enumerate(economy.daily_quests(farm["user_uid"], day)): + _quests().insert( + { + "uid": generate_uid(), + "farm_uid": farm["uid"], + "user_uid": farm["user_uid"], + "day": day, + "slot_index": index, + "kind": quest["kind"], + "label": quest["label"], + "goal": quest["goal"], + "progress": 0, + "reward_coins": quest["reward_coins"], + "reward_xp": quest["reward_xp"], + "claimed": 0, + "created_at": now, + "updated_at": now, + } + ) + + +def advance_quests(user_uid: str, kind: str, amount: int) -> None: + if amount <= 0: + return + try: + farm = get_farm(user_uid) + if not farm: + return + day = _today() + ensure_quests(farm, day) + for row in _quests().find(farm_uid=farm["uid"], day=day, kind=kind): + if row.get("claimed"): + continue + goal = int(row.get("goal") or 0) + progress = min(goal, int(row.get("progress") or 0) + amount) + _quests().update( + {"uid": row["uid"], "progress": progress, "updated_at": _iso(_now())}, + ["uid"], + ) + except Exception: + return diff --git a/devplacepy/services/game/store/serialize.py b/devplacepy/services/game/store/serialize.py new file mode 100644 index 00000000..b41e4cc1 --- /dev/null +++ b/devplacepy/services/game/store/serialize.py @@ -0,0 +1,300 @@ +# retoor + +from __future__ import annotations + +import json +from datetime import datetime, timedelta + +from .. import economy +from .common import ( + PERK_COLUMN, + _lvl, + _now, + _parse, + _parse_date, + _quests, + steal_cooldown_remaining, +) +from .farm import get_farm, get_plots +from .quests import ensure_quests + + +def _watered_by(plot: dict) -> list[str]: + try: + value = json.loads(plot.get("watered_by") or "[]") + return value if isinstance(value, list) else [] + except (ValueError, TypeError): + return [] + + +def _plot_state(plot: dict, now: datetime) -> str: + if not plot.get("crop_key"): + return "empty" + ready_at = _parse(plot.get("ready_at", "")) + if ready_at and now >= ready_at: + return "ready" + return "growing" + + +def serialize_plot( + plot: dict, + *, + viewer_uid: str, + owner_uid: str, + now: datetime, + yield_level: int = 0, + prestige: int = 0, + growth_level: int = 0, + ci_tier: int = 1, + legacy_mult_level: int = 0, + legacy_speed_level: int = 0, + defense_level: int = 0, + steal_locked_until: int = 0, +) -> dict: + state = _plot_state(plot, now) + crop = economy.crop_for(plot.get("crop_key", "")) + ready_at = _parse(plot.get("ready_at", "")) + remaining = 0 + if state == "growing" and ready_at: + remaining = max(0, int((ready_at - now).total_seconds())) + watered = _watered_by(plot) + is_owner = viewer_uid == owner_uid + golden = economy.is_golden(plot.get("uid", ""), plot.get("planted_at", "")) + can_water = ( + state == "growing" + and not is_owner + and bool(viewer_uid) + and viewer_uid not in watered + and len(watered) < economy.MAX_WATERS_PER_PLOT + ) + grace = economy.effective_steal_grace(defense_level) + protected = bool(ready_at) and now < ready_at + timedelta(seconds=grace) + eligible = ( + state == "ready" + and not is_owner + and bool(viewer_uid) + and bool(crop) + and ready_at is not None + ) + can_steal = eligible and not protected and steal_locked_until <= 0 + steal_reason = "" + if eligible and protected: + steal_reason = "protected" + elif eligible and steal_locked_until > 0: + steal_reason = "cooldown" + steal_coins = ( + economy.steal_reward_coins( + crop, yield_level, prestige, legacy_mult_level, defense_level + ) + if can_steal + else 0 + ) + fertilize_cost = 0 + if state == "growing" and is_owner and crop: + full_grow = economy.grow_seconds_for( + crop, ci_tier, growth_level, legacy_speed_level + ) + reduce_by = int(remaining * economy.FERTILIZE_FRACTION) + eff_reward = economy.effective_reward_coins( + crop, yield_level, prestige, legacy_mult_level + ) + fertilize_cost = economy.fertilize_click_cost(eff_reward, reduce_by, full_grow) + return { + "slot": plot.get("slot_index", 0), + "state": state, + "crop_key": plot.get("crop_key", ""), + "crop_name": crop.name if crop else "", + "crop_icon": crop.icon if crop else "", + "reward_coins": crop.reward_coins if crop else 0, + "reward_xp": crop.reward_xp if crop else 0, + "ready_at": plot.get("ready_at", ""), + "remaining_seconds": remaining, + "watered_count": len(watered), + "max_waters": economy.MAX_WATERS_PER_PLOT, + "can_water": can_water, + "can_steal": can_steal, + "steal_coins": steal_coins, + "steal_cooldown_seconds": steal_locked_until if steal_reason == "cooldown" else 0, + "steal_reason": steal_reason, + "is_golden": golden and (is_owner or can_steal), + "fertilize_cost": fertilize_cost, + } + + +def serialize_farm( + farm: dict, *, viewer: dict | None, owner: dict, now: datetime | None = None +) -> dict: + from .actions import _auto_harvest + + now = now or _now() + viewer_uid = viewer["uid"] if viewer else "" + owner_uid = owner["uid"] + is_owner = viewer_uid == owner_uid + if is_owner and _lvl(farm, "legacy_autoharvest") > 0: + farm = _auto_harvest(farm, owner_uid, now) + prestige = _lvl(farm, "prestige") + growth_level = _lvl(farm, "perk_growth") + discount_level = _lvl(farm, "perk_discount") + yield_level = _lvl(farm, "perk_yield") + legacy_mult_level = _lvl(farm, "legacy_multiplier") + legacy_speed_level = _lvl(farm, "legacy_speed") + defense_level = _lvl(farm, "legacy_defense") + ci_tier = int(farm.get("ci_tier", 1)) + steal_locked_until = ( + steal_cooldown_remaining(viewer_uid, owner_uid, now) + if viewer_uid and not is_owner + else 0 + ) + plots = get_plots(farm["uid"]) + serialized_plots = [ + serialize_plot( + plot, + viewer_uid=viewer_uid, + owner_uid=owner_uid, + now=now, + yield_level=yield_level, + prestige=prestige, + growth_level=growth_level, + ci_tier=ci_tier, + legacy_mult_level=legacy_mult_level, + legacy_speed_level=legacy_speed_level, + defense_level=defense_level, + steal_locked_until=steal_locked_until, + ) + for plot in plots + ] + progress = economy.level_progress(int(farm.get("xp", 0))) + level = progress["level"] + next_tier = economy.next_ci_tier(ci_tier) + ci_entry = economy.CI_BY_TIER.get(ci_tier) + streak = _lvl(farm, "streak") + daily_available = is_owner and _daily_available(farm, now) + return { + "owner_username": owner.get("username", ""), + "owner_uid": owner_uid, + "is_owner": is_owner, + "coins": int(farm.get("coins", 0)), + "xp": int(farm.get("xp", 0)), + "level": level, + "level_into": progress["into_level"], + "level_span": progress["span"], + "level_is_max": progress["is_max"], + "total_harvests": int(farm.get("total_harvests", 0)), + "plot_count": int(farm.get("plot_count", economy.STARTING_PLOTS)), + "max_plots": economy.MAX_PLOTS, + "next_plot_cost": ( + economy.plot_cost(int(farm.get("plot_count", economy.STARTING_PLOTS))) + if int(farm.get("plot_count", economy.STARTING_PLOTS)) < economy.MAX_PLOTS + else 0 + ), + "ci_tier": ci_tier, + "ci_label": ci_entry.label if ci_entry else "", + "ci_speed": economy.ci_speed(ci_tier), + "ci_next_tier": next_tier.tier if next_tier else 0, + "ci_next_label": next_tier.label if next_tier else "", + "ci_next_cost": next_tier.upgrade_cost if next_tier else 0, + "plots": serialized_plots, + "crops": [ + economy.crop_payload( + crop, + ci_tier, + level, + growth_level, + discount_level, + yield_level, + prestige, + legacy_mult_level, + legacy_speed_level, + ) + for crop in economy.CROPS + ], + "prestige": prestige, + "prestige_multiplier": economy.prestige_multiplier(prestige), + "prestige_min_level": economy.PRESTIGE_MIN_LEVEL, + "prestige_available": is_owner and level >= economy.PRESTIGE_MIN_LEVEL, + "streak": streak, + "daily_available": daily_available, + "daily_reward": economy.daily_reward(streak + 1 if daily_available else streak), + "perks": _serialize_perks(farm) if is_owner else [], + "quests": _serialize_quests(owner_uid, now) if is_owner else [], + "stars": _lvl(farm, "stars"), + "legacy": _serialize_legacy(farm) if is_owner else [], + "steal_cooldown_seconds": steal_locked_until, + } + + +def _daily_available(farm: dict, now: datetime) -> bool: + last = _parse_date(farm.get("last_daily_at") or "") + return last != now.date() + + +def _serialize_perks(farm: dict) -> list[dict]: + perks = [] + for perk in economy.PERKS: + level = _lvl(farm, PERK_COLUMN[perk.key]) + maxed = level >= perk.max_level + perks.append( + { + "key": perk.key, + "name": perk.name, + "icon": perk.icon, + "description": perk.description, + "level": level, + "max_level": perk.max_level, + "cost": 0 if maxed else economy.perk_cost(perk, level), + "maxed": maxed, + "effect": economy.perk_value_text(perk, level), + } + ) + return perks + + +def _serialize_legacy(farm: dict) -> list[dict]: + upgrades = [] + for up in economy.LEGACY_UPGRADES: + level = _lvl(farm, f"legacy_{up.key}") + maxed = level >= up.max_level + upgrades.append( + { + "key": up.key, + "name": up.name, + "icon": up.icon, + "description": up.description, + "level": level, + "max_level": up.max_level, + "cost": 0 if maxed else economy.legacy_cost(up, level), + "maxed": maxed, + "effect": economy.legacy_value_text(up, level), + } + ) + return upgrades + + +def _serialize_quests(user_uid: str, now: datetime) -> list[dict]: + farm = get_farm(user_uid) + if not farm: + return [] + day = now.date().isoformat() + ensure_quests(farm, day) + rows = sorted( + _quests().find(farm_uid=farm["uid"], day=day), + key=lambda row: row.get("slot_index", 0), + ) + quests = [] + for row in rows: + goal = int(row.get("goal") or 0) + progress = min(goal, int(row.get("progress") or 0)) + claimed = bool(row.get("claimed")) + quests.append( + { + "kind": row.get("kind", ""), + "label": row.get("label", ""), + "goal": goal, + "progress": progress, + "reward_coins": int(row.get("reward_coins") or 0), + "reward_xp": int(row.get("reward_xp") or 0), + "claimed": claimed, + "can_claim": (not claimed) and goal > 0 and progress >= goal, + } + ) + return quests