"""Daily/weekly quests tracked from lifetime counters.""" from __future__ import annotations import random from typing import TypedDict from .store import ( UserData, _commit, _locked_by, _log, _now, _prestige_mult, get_user, ) # --------------------------------------------------------------------------- # Quest system # --------------------------------------------------------------------------- # Quests reuse the monotonic lifetime counters already tracked on each user. # Progress = (current counter value) - (snapshot taken when the quest was rolled). # Reset is lazy & per-user: the active set is regenerated the first time a user # interacts after the day / ISO-week rolls over (mirrors the streak-date logic). # Rotation is seeded by user id + period key, so each player gets their own set. class QuestDef(TypedDict): stat: str # UserData counter field the quest tracks goal: int coins: int exp: int QUESTS_DAILY: dict[str, QuestDef] = { "work3": {"stat": "work_count", "goal": 3, "coins": 150, "exp": 20}, "beg5": {"stat": "beg_count", "goal": 5, "coins": 100, "exp": 15}, "wager500": {"stat": "total_wagered", "goal": 500, "coins": 150, "exp": 20}, "fish2": {"stat": "total_fish_caught", "goal": 2, "coins": 150, "exp": 20}, "crime1": {"stat": "crimes_succeeded", "goal": 1, "coins": 200, "exp": 25}, "earn1000": {"stat": "lifetime_earned", "goal": 1000, "coins": 150, "exp": 20}, "give200": {"stat": "total_given", "goal": 200, "coins": 100, "exp": 15}, } QUESTS_WEEKLY: dict[str, QuestDef] = { "work20": {"stat": "work_count", "goal": 20, "coins": 1000, "exp": 100}, "fish15": {"stat": "total_fish_caught", "goal": 15, "coins": 1200, "exp": 100}, "wager5000": {"stat": "total_wagered", "goal": 5000, "coins": 1000, "exp": 100}, "crime5": {"stat": "crimes_succeeded", "goal": 5, "coins": 1200, "exp": 120}, "heist1": {"stat": "heists_joined", "goal": 1, "coins": 800, "exp": 80}, "earn10000": {"stat": "lifetime_earned", "goal": 10000, "coins": 1500, "exp": 150}, } DAILY_QUEST_COUNT = 3 WEEKLY_QUEST_COUNT = 2 def _period_keys() -> tuple[str, str]: """Return (day_key, week_key) for the current UTC time.""" today = _now().date() iso = today.isocalendar() return today.isoformat(), f"{iso[0]}-W{iso[1]:02d}" def _pick_quests(pool: dict[str, QuestDef], count: int, seed: str) -> list[str]: """Deterministically choose `count` quest ids from `pool` for a period.""" rng = random.Random(seed) return rng.sample(sorted(pool.keys()), min(count, len(pool))) def _new_quest_block( user: UserData, user_id: int, pool: dict[str, QuestDef], count: int, period_val: str, period_field: str ) -> dict: chosen = _pick_quests(pool, count, f"{user_id}:{period_field}:{period_val}") return { period_field: period_val, "quests": { qid: {"snap": int(user.get(pool[qid]["stat"], 0) or 0), "claimed": False} for qid in chosen }, } def _ensure_quests(user: UserData, user_id: int) -> bool: """Roll fresh daily/weekly quest sets if their period elapsed. Mutates `user` in place; returns True if anything changed (caller commits).""" changed = False day_key, week_key = _period_keys() if (user.get("quest_daily") or {}).get("date") != day_key: user["quest_daily"] = _new_quest_block(user, user_id, QUESTS_DAILY, DAILY_QUEST_COUNT, day_key, "date") changed = True if (user.get("quest_weekly") or {}).get("week") != week_key: user["quest_weekly"] = _new_quest_block(user, user_id, QUESTS_WEEKLY, WEEKLY_QUEST_COUNT, week_key, "week") changed = True return changed def _quest_progress(user: UserData, pool: dict[str, QuestDef], qid: str, state: dict) -> int: cur = int(user.get(pool[qid]["stat"], 0) or 0) return max(0, cur - int(state.get("snap", 0))) def _quest_view(user: UserData) -> dict: def build(pool: dict[str, QuestDef], block: dict) -> list[dict]: out: list[dict] = [] for qid, state in (block.get("quests") or {}).items(): if qid not in pool: continue d = pool[qid] prog = min(d["goal"], _quest_progress(user, pool, qid, state)) out.append({ "id": qid, "goal": d["goal"], "coins": d["coins"], "exp": d["exp"], "progress": prog, "done": prog >= d["goal"], "claimed": bool(state.get("claimed")), }) return out return { "daily": build(QUESTS_DAILY, user.get("quest_daily") or {}), "weekly": build(QUESTS_WEEKLY, user.get("quest_weekly") or {}), } @_locked_by(0) async def get_quests(user_id: int) -> dict: """Return the user's active quests, rolling new sets if the period elapsed.""" user = await get_user(user_id) if _ensure_quests(user, user_id): saved = await _commit(user_id, user) if saved is not None and "quest_daily" not in saved: _log.warning( "PocketBase collection has no quest fields - quest state is not " "persisted and progress will stay at 0. Run scripts/add_quest_fields.py." ) return _quest_view(user) @_locked_by(0) async def claim_quests(user_id: int) -> dict: """Grant coins for every completed-but-unclaimed quest and mark them claimed. Coins (with prestige coin_mult) are paid here; EXP is returned raw for the caller to award via the shared award_exp path (keeps level-up notices).""" user = await get_user(user_id) _ensure_quests(user, user_id) coin_mult, _ = _prestige_mult(user) total_coins = total_exp = claimed = 0 for pool, block in ( (QUESTS_DAILY, user.get("quest_daily") or {}), (QUESTS_WEEKLY, user.get("quest_weekly") or {}), ): for qid, state in (block.get("quests") or {}).items(): if qid not in pool or state.get("claimed"): continue if _quest_progress(user, pool, qid, state) < pool[qid]["goal"]: continue total_coins += pool[qid]["coins"] total_exp += pool[qid]["exp"] state["claimed"] = True claimed += 1 if not claimed: return {"ok": False, "reason": "nothing"} coins_awarded = int(total_coins * coin_mult) user["balance"] += coins_awarded user["lifetime_earned"] = user.get("lifetime_earned", 0) + coins_awarded user["peak_balance"] = max(user.get("peak_balance", 0), user["balance"]) await _commit(user_id, user) return {"ok": True, "claimed": claimed, "coins": coins_awarded, "exp": total_exp, "balance": user["balance"]}