# retoor <retoor@molodetz.nl>
from __future__ import annotations
import json
from datetime import datetime, timedelta
from .. import economy
from .common import (
PERK_COLUMN,
_iso_week,
_lvl,
_now,
_parse,
_parse_date,
_quests,
steal_cooldown_remaining,
)
from .cosmetics import owned_cosmetic_keys
from .era import active_era_name
from .farm import get_farm, get_plots
from .infrastructure import INFRA_COLUMN, owns_infrastructure
from .market import market_factor_for
from .mastery import MASTERY_COLUMN
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)
immune = bool(crop and crop.steal_immune)
eligible = (
state == "ready"
and not is_owner
and bool(viewer_uid)
and bool(crop)
and ready_at is not None
and not immune
)
can_steal = eligible and not protected and steal_locked_until <= 0
steal_reason = ""
if immune and state == "ready" and not is_owner:
steal_reason = "immune"
elif eligible and protected:
steal_reason = "protected"
elif eligible and steal_locked_until > 0:
steal_reason = "cooldown"
market_factor = market_factor_for(crop.key) if crop else 1.0
steal_coins = (
economy.steal_reward_coins(
crop, yield_level, prestige, legacy_mult_level, defense_level, market_factor
)
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, market_factor
)
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
from .defense import charge_upkeep
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)
if is_owner:
farm = charge_upkeep(farm, now)
farm = _reset_harvests_week(farm, 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)
mastery_earned = _lvl(farm, "mastery_points_earned_total")
era_name = active_era_name()
registry_boost = owns_infrastructure(farm, "registry")
underdog_until = _parse(farm.get("underdog_boost_until") or "")
underdog_seconds = (
max(0, int((underdog_until - now).total_seconds())) if underdog_until else 0
)
building_defense_level = _lvl(farm, "defense_level")
defense_tier_info = economy.defense_tier(building_defense_level)
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,
market_factor_for(crop.key),
mastery_earned,
era_name,
registry_boost,
)
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,
"mastery_points": _lvl(farm, "mastery_points"),
"mastery_points_earned_total": mastery_earned,
"mastery": _serialize_mastery(farm) if is_owner else [],
"infrastructure": _serialize_infrastructure(farm) if is_owner else [],
"defense_level": building_defense_level,
"defense_tier_name": defense_tier_info.name,
"defense_upkeep_daily": economy.daily_upkeep(defense_tier_info, int(farm.get("coins", 0))),
"defense_next_cost": (
economy.next_defense_tier(building_defense_level).upgrade_cost
if economy.next_defense_tier(building_defense_level)
else 0
),
"cosmetics": _serialize_cosmetics(owner_uid) if is_owner else [],
"active_title": farm.get("active_title") or "",
"underdog_boost_seconds_remaining": underdog_seconds,
"mastery_analytics_unlocked": bool(_lvl(farm, "mastery_analytics")),
"lifetime_coins_earned": _lvl(farm, "lifetime_coins_earned"),
"lifetime_harvests": _lvl(farm, "lifetime_harvests"),
"harvests_week": _lvl(farm, "harvests_week"),
"era_active": era_name is not None,
"era_name": era_name or "",
"era_coins": _lvl(farm, "era_coins"),
"era_harvests": _lvl(farm, "era_harvests"),
}
def _reset_harvests_week(farm: dict, now: datetime) -> dict:
from .common import _update_farm
current_week = _iso_week(now)
if farm.get("harvests_week_start") == current_week:
return farm
fields = {"harvests_week": 0, "harvests_week_start": current_week}
_update_farm(farm["uid"], fields)
return {**farm, **fields}
def _serialize_mastery(farm: dict) -> list[dict]:
upgrades = []
for m in economy.MASTERY_UPGRADES:
level = _lvl(farm, MASTERY_COLUMN[m.key])
maxed = level >= m.max_level
upgrades.append(
{
"key": m.key,
"name": m.name,
"icon": m.icon,
"description": m.description,
"level": level,
"max_level": m.max_level,
"cost": 0 if maxed else economy.mastery_cost(m, level),
"maxed": maxed,
"effect": m.description,
}
)
return upgrades
def _serialize_infrastructure(farm: dict) -> list[dict]:
rows = []
for infra in economy.INFRASTRUCTURE:
owned = bool(_lvl(farm, INFRA_COLUMN[infra.key]))
rows.append(
{
"key": infra.key,
"name": infra.name,
"icon": infra.icon,
"description": infra.description,
"cost": infra.cost,
"min_prestige": infra.min_prestige,
"owned": owned,
}
)
return rows
def _serialize_cosmetics(user_uid: str) -> list[dict]:
owned = owned_cosmetic_keys(user_uid)
return [
{
"key": c.key,
"name": c.name,
"icon": c.icon,
"description": c.description,
"cost_coins": c.cost_coins,
"kind": c.kind,
"owned": c.key in owned,
}
for c in economy.COSMETICS
if c.era_key is None
]
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]:
from .quests import ensure_weekly_contract
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, scope="daily"),
key=lambda row: row.get("slot_index", 0),
)
if _lvl(farm, "mastery_contracts"):
iso_week = _iso_week(now)
ensure_weekly_contract(farm, iso_week)
rows += sorted(
_quests().find(farm_uid=farm["uid"], day=iso_week, scope="weekly"),
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,
"scope": row.get("scope", "daily"),
"reward_stars": int(row.get("reward_stars") or 0),
}
)
return quests
View Git Blame Copy Permalink