Files
devplacepy/devplacepy/services/game/store/actions.py
T
2026-07-23 01:14:10 +02:00

626 lines
22 KiB
Python

# retoor <retoor@molodetz.nl>
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,
conditional_update_farm,
credit_farm,
_iso,
_iso_week,
_lvl,
_now,
_parse,
_parse_date,
_plots,
_quests,
_steals,
_today,
_update_farm,
steal_cooldown_remaining,
)
from .era import active_era_name
from .farm import _create_plot, _plot_at, ensure_farm, get_farm, get_plots
from .infrastructure import owns_infrastructure, roll_canary
from .market import market_factor_for, record_harvest_tick
from .mastery import MASTERY_COLUMN
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}.")
if crop.min_mastery > _lvl(farm, "mastery_points_earned_total"):
raise GameError(f"{crop.name} unlocks after reaching Mastery.")
if crop.era_key and crop.era_key != active_era_name():
raise GameError(f"{crop.name} is not available right now.")
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"),
owns_infrastructure(farm, "registry"),
)
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_crop(farm: dict, crop, plot_uid: str, planted_at: str, now) -> dict:
prestige = _lvl(farm, "prestige")
golden = economy.is_golden(plot_uid, planted_at)
market_factor = market_factor_for(crop.key)
underdog = bool(farm.get("underdog_boost_until")) and now < (
_parse(farm.get("underdog_boost_until") or "") or now
)
coins_gain = economy.effective_reward_coins(
crop,
_lvl(farm, "perk_yield"),
prestige,
_lvl(farm, "legacy_multiplier"),
market_factor,
underdog,
)
if golden:
coins_gain *= economy.GOLDEN_MULTIPLIER
contract_boost_until = _parse(farm.get("contract_boost_until") or "")
if contract_boost_until and now < contract_boost_until:
coins_gain = round(coins_gain * economy.WEEKLY_CONTRACT_BOOST_MULTIPLIER)
if owns_infrastructure(farm, "canary"):
plant_cost = economy.effective_plant_cost(crop, _lvl(farm, "perk_discount"))
coins_gain = roll_canary(coins_gain, plant_cost)
xp_gain = economy.effective_reward_xp(crop, _lvl(farm, "perk_xp"))
record_harvest_tick(crop.key, now)
return {"coins": coins_gain, "xp": xp_gain, "golden": golden}
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.")
planted_at = plot.get("planted_at", "")
_plots().update(
{
"uid": plot["uid"],
"crop_key": "",
"planted_at": "",
"ready_at": "",
"watered_by": "[]",
"updated_at": _iso(now),
},
["uid"],
)
result = _harvest_crop(farm, crop, plot.get("uid", ""), planted_at, now)
coins_gain, xp_gain, golden = result["coins"], result["xp"], result["golden"]
extra = {}
if crop.key == "kernel" and _lvl(farm, "last_kernel_harvest_prestige") != _lvl(farm, "prestige"):
prestiged_at = _parse(farm.get("prestiged_at") or "")
if prestiged_at:
extra["time_to_kernel_seconds"] = max(0, int((now - prestiged_at).total_seconds()))
extra["last_kernel_harvest_prestige"] = _lvl(farm, "prestige")
farm = credit_farm(
farm,
coins=coins_gain,
xp=xp_gain,
harvests=1,
era_active=bool(active_era_name()),
extra=extra or None,
)
advance_quests(user["uid"], "harvest", 1)
advance_quests(user["uid"], "earn", coins_gain)
if _lvl(farm, MASTERY_COLUMN["autoreplant"]):
_try_autoreplant(farm, slot, crop, now)
return {
"slot": slot,
"crop": crop.key,
"coins": coins_gain,
"xp": xp_gain,
"golden": golden,
}
def _try_autoreplant(farm: dict, slot: int, crop, now) -> None:
plot = _plot_at(farm["uid"], slot)
if not plot or plot.get("crop_key"):
return
cost = economy.effective_plant_cost(crop, _lvl(farm, "perk_discount"))
if int(farm.get("coins", 0)) < cost:
return
grow = economy.grow_seconds_for(
crop,
int(farm.get("ci_tier", 1)),
_lvl(farm, "perk_growth"),
_lvl(farm, "legacy_speed"),
owns_infrastructure(farm, "registry"),
)
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": int(farm.get("coins", 0)) - 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")
)
if owns_infrastructure(farm, "registry") and crop.key in economy.REGISTRY_BOOST_CROPS:
bonus = round(bonus * economy.REGISTRY_BOOST_FACTOR)
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.")
if crop.steal_immune:
raise GameError("That build cannot be raided.")
defense_level = _lvl(farm, "legacy_defense")
building_level = _lvl(farm, "defense_level")
building_tier = economy.defense_tier(building_level)
floor = building_tier.steal_fraction_floor
if owns_infrastructure(farm, "observability"):
floor = max(floor, economy.OBSERVABILITY_STEAL_FLOOR)
grace = economy.effective_steal_grace(defense_level, building_tier.grace_bonus)
ready_at = _parse(plot.get("ready_at", "")) or now
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"],
)
market_factor = market_factor_for(crop.key)
coins_gain = economy.steal_reward_coins(
crop,
_lvl(farm, "perk_yield"),
_lvl(farm, "prestige"),
_lvl(farm, "legacy_multiplier"),
defense_level,
market_factor,
floor,
)
thief_farm = ensure_farm(thief["uid"])
underdog_extra = {}
if int(farm.get("coins", 0)) > int(thief_farm.get("coins", 0)) * economy.UNDERDOG_COIN_RATIO:
underdog_extra["underdog_boost_until"] = _iso(
now + timedelta(hours=economy.UNDERDOG_DURATION_HOURS)
)
credit_farm(
thief_farm,
coins=coins_gain,
era_active=bool(active_era_name()),
extra=underdog_extra or None,
)
_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"],
"underdog_triggered": bool(underdog_extra),
}
def _auto_harvest(farm: dict, owner_uid: str, now: datetime) -> dict:
coins_gain = 0
xp_gain = 0
harvested = 0
replant_slots = []
extra = {}
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
result = _harvest_crop(farm, crop, plot.get("uid", ""), plot.get("planted_at", ""), now)
_plots().update(
{
"uid": plot["uid"],
"crop_key": "",
"planted_at": "",
"ready_at": "",
"watered_by": "[]",
"updated_at": _iso(now),
},
["uid"],
)
coins_gain += result["coins"]
xp_gain += result["xp"]
harvested += 1
replant_slots.append((int(plot.get("slot_index", 0)), crop))
if crop.key == "kernel" and _lvl(farm, "last_kernel_harvest_prestige") != _lvl(
farm, "prestige"
):
prestiged_at = _parse(farm.get("prestiged_at") or "")
if prestiged_at:
extra["time_to_kernel_seconds"] = max(0, int((now - prestiged_at).total_seconds()))
extra["last_kernel_harvest_prestige"] = _lvl(farm, "prestige")
if not harvested:
return farm
farm = credit_farm(
farm,
coins=coins_gain,
xp=xp_gain,
harvests=harvested,
era_active=bool(active_era_name()),
extra=extra or None,
)
advance_quests(owner_uid, "harvest", harvested)
advance_quests(owner_uid, "earn", coins_gain)
if _lvl(farm, MASTERY_COLUMN["autoreplant"]):
for slot_index, crop in replant_slots:
_try_autoreplant(farm, slot_index, crop, now)
return get_farm(owner_uid) or farm
def claim_quest(user: dict, kind: str, scope: str = "daily") -> dict:
from .quests import ensure_weekly_contract
farm = ensure_farm(user["uid"])
now = _now()
if scope == "weekly":
day = _iso_week(now)
ensure_weekly_contract(farm, day)
else:
day = _today()
ensure_quests(farm, day)
row = _quests().find_one(farm_uid=farm["uid"], day=day, kind=kind, scope=scope)
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)
reward_stars = int(row.get("reward_stars") or 0)
_quests().update(
{"uid": row["uid"], "claimed": 1, "updated_at": _iso(now)}, ["uid"]
)
extra = {"stars": _lvl(farm, "stars") + reward_stars} if reward_stars else None
if scope == "weekly":
extra = extra or {}
extra["contract_boost_until"] = _iso(
now + timedelta(hours=economy.WEEKLY_CONTRACT_BOOST_HOURS)
)
credit_farm(farm, coins=reward_coins, xp=reward_xp, extra=extra)
return {
"kind": kind,
"reward_coins": reward_coins,
"reward_xp": reward_xp,
"reward_stars": reward_stars,
}
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)
credit_farm(
farm,
coins=reward,
era_active=bool(active_era_name()),
extra={"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)
rows = conditional_update_farm(
farm["uid"],
set_clause=f"coins = coins - :cost, {column} = :new_level",
where_clause=f"COALESCE({column}, 0) = :current_level AND coins >= :cost",
params={"cost": cost, "new_level": level + 1, "current_level": level},
)
if rows == 0:
farm = get_farm(user["uid"])
if _lvl(farm, column) != level:
raise GameError("That perk already changed - refresh and try again.")
raise GameError("Not enough coins for that upgrade.")
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)
rows = conditional_update_farm(
farm["uid"],
set_clause=f"stars = COALESCE(stars, 0) - :cost, {column} = :new_level",
where_clause=f"COALESCE({column}, 0) = :current_level AND COALESCE(stars, 0) >= :cost",
params={"cost": cost, "new_level": level + 1, "current_level": level},
)
if rows == 0:
farm = get_farm(user["uid"])
if _lvl(farm, column) != level:
raise GameError("That legacy upgrade already changed - refresh and try again.")
raise GameError("Not enough stars for that legacy upgrade.")
return {"key": upgrade.key, "level": level + 1, "spent": cost}
def prestige(user: dict) -> dict:
from .treasury import credit_treasury
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)."
)
old_prestige = _lvl(farm, "prestige")
new_prestige = old_prestige + 1
coins_now = int(farm.get("coins", 0))
fee = economy.refactor_cost(old_prestige, coins_now)
if coins_now < fee:
raise GameError(
f"Refactoring costs {fee} coins right now - keep farming to afford it."
)
carried = economy.refactor_carryover(coins_now, fee, _lvl(farm, "legacy_carryover"))
stars_award = economy.stars_for_refactor(level, old_prestige)
mastery_award = economy.mastery_points_awarded(old_prestige, new_prestige)
base_plots = economy.prestige_base_plots(_lvl(farm, "legacy_plots"))
now_dt = _now()
now = _iso(now_dt)
set_parts = [
"coins = :post_coins",
"xp = 0",
"level = 1",
"ci_tier = 1",
"plot_count = :base_plots",
"prestige = :new_prestige",
"stars = COALESCE(stars, 0) + :stars_award",
"mastery_points = COALESCE(mastery_points, 0) + :mastery_award",
"mastery_points_earned_total = COALESCE(mastery_points_earned_total, 0) + :mastery_award",
"prestiged_at = :now",
]
params = {
"post_coins": economy.STARTING_COINS + carried,
"base_plots": base_plots,
"new_prestige": new_prestige,
"stars_award": stars_award,
"mastery_award": mastery_award,
"now": now,
"old_prestige": old_prestige,
"coins_now": coins_now,
}
for perk in economy.PERKS:
column = PERK_COLUMN[perk.key]
set_parts.append(f"{column} = 0")
rows = conditional_update_farm(
farm["uid"],
set_clause=", ".join(set_parts),
where_clause="COALESCE(prestige, 0) = :old_prestige AND coins = :coins_now",
params=params,
)
if rows == 0:
farm = get_farm(user["uid"])
if _lvl(farm, "prestige") != old_prestige:
raise GameError("Your farm already refactored - refresh and try again.")
if int(farm.get("coins", 0)) != coins_now:
raise GameError("Your coin balance just changed - refresh and try again.")
raise GameError(
f"Reach level {economy.PRESTIGE_MIN_LEVEL} to refactor (prestige)."
)
credit_treasury(fee)
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)
return {
"prestige": new_prestige,
"stars_awarded": stars_award,
"mastery_awarded": mastery_award,
"fee": fee,
"carried": carried,
}
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"),
owns_infrastructure(farm, "registry"),
)
eff_reward = economy.effective_reward_coins(
crop,
_lvl(farm, "perk_yield"),
_lvl(farm, "prestige"),
_lvl(farm, "legacy_multiplier"),
market_factor_for(crop.key),
)
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}