"""Persist daily scoreboard rows and compute server-authoritative points. Rows are unique by date; upserts rewrite the JSONL file under lock. """ from __future__ import annotations from datetime import date, datetime, timedelta, timezone from typing import Any from app.fsutil import read_jsonl, rewrite_jsonl from app.models import Court, DailyRow, DailyUpsert, Daydream, Rerun from app.paths import Paths def compute_daily_points( *, brick_done: bool, corn_sessions: int, delay_ok: bool, daydream: Daydream | str, rerun: Rerun | str, court: Court | str, ) -> int: """Compute daily points from the authoritative formula (max 6).""" points = 0 if brick_done: points += 2 if corn_sessions == 0 or (corn_sessions <= 1 and delay_ok): points += 1 daydream_value = daydream.value if isinstance(daydream, Daydream) else daydream if daydream_value != Daydream.fc.value: points += 1 rerun_value = rerun.value if isinstance(rerun, Rerun) else rerun if rerun_value == Rerun.clean.value: points += 1 court_value = court.value if isinstance(court, Court) else court if court_value == Court.closed.value: points += 1 return points def week_band(days_present: int, week_points: int) -> str: """Classify a 7-day window from present days and point sum.""" if days_present < 7: return "incomplete" if week_points >= 28: return "strong" if week_points >= 18: return "mixed" return "escape_heavy" def indoors_streak(rows: list[DailyRow], *, as_of: date) -> int: """Count consecutive stayed_indoors_all_day days ending at as_of.""" by_date = {row.date: row for row in rows} streak = 0 day = as_of while True: row = by_date.get(day) if row is None or not row.stayed_indoors_all_day: break streak += 1 day = day - timedelta(days=1) return streak def resolve_movement_minutes(data: DailyUpsert) -> int: """Prefer sum of walk splits when present; else client movement_minutes.""" split = int(data.interrupt_walk_minutes or 0) + int(data.fantasy_walk_minutes or 0) if split > 0: return min(24 * 60, split) return int(data.movement_minutes or 0) def movement_summary(rows: list[DailyRow], *, as_of: date, lookback: int = 3) -> dict[str, Any]: """Indoors streak + walk/proof/fantasy snapshot for agent / Home nudges.""" by_date = {row.date: row for row in rows} last_left: list[dict[str, Any]] = [] for offset in range(lookback): day = as_of - timedelta(days=offset) row = by_date.get(day) last_left.append( { "date": day.isoformat(), "left_home": bool(row.left_home) if row else None, "left_room": bool(row.left_room) if row else None, "stayed_indoors_all_day": bool(row.stayed_indoors_all_day) if row else None, "movement_minutes": int(row.movement_minutes) if row else None, "interrupt_walk_minutes": int(row.interrupt_walk_minutes) if row else None, "fantasy_walk_minutes": int(row.fantasy_walk_minutes) if row else None, "headphones_on_walk": bool(row.headphones_on_walk) if row else None, "music_cinematic_on_walk": bool(row.music_cinematic_on_walk) if row else None, "proof_brick_done": bool(row.proof_brick_done) if row else None, "proof_brick_kind": row.proof_brick_kind if row else None, "fantasy_minutes_scheduled": int(row.fantasy_minutes_scheduled) if row else None, "fantasy_minutes_unplanned": int(row.fantasy_minutes_unplanned) if row else None, } ) today = by_date.get(as_of) streak = indoors_streak(rows, as_of=as_of) return { "indoors_streak": streak, "nudge": streak >= 2, "last_3_days_left_home": last_left, "today": ( { "left_room": today.left_room, "left_home": today.left_home, "interrupt_walk_minutes": today.interrupt_walk_minutes, "fantasy_walk_minutes": today.fantasy_walk_minutes, "headphones_on_walk": today.headphones_on_walk, "music_cinematic_on_walk": today.music_cinematic_on_walk, "proof_brick_done": today.proof_brick_done, "proof_brick_kind": today.proof_brick_kind, "fantasy_minutes_scheduled": today.fantasy_minutes_scheduled, "fantasy_minutes_unplanned": today.fantasy_minutes_unplanned, } if today else None ), } def _utc_now() -> datetime: return datetime.now(timezone.utc) class DailyStore: """Read and mutate the durable daily JSONL file.""" def __init__(self, paths: Paths) -> None: self.paths = paths def _load(self) -> list[DailyRow]: return [DailyRow.model_validate(record) for record in read_jsonl(self.paths.daily)] def get(self, day: date) -> DailyRow | None: """Return the row for one calendar date, or None.""" for row in self._load(): if row.date == day: return row return None def upsert(self, day: date, data: DailyUpsert) -> DailyRow: """Create or replace a daily row and recompute points.""" points = compute_daily_points( brick_done=data.brick_done, corn_sessions=data.corn_sessions, delay_ok=data.delay_ok, daydream=data.daydream, rerun=data.rerun, court=data.court, ) movement = resolve_movement_minutes(data) row = DailyRow( date=day, primary_brick=data.primary_brick, brick_done=data.brick_done, corn_sessions=data.corn_sessions, delay_ok=data.delay_ok, daydream=data.daydream, rerun=data.rerun, court=data.court, stayed_indoors_all_day=data.stayed_indoors_all_day, left_room=data.left_room, left_home=data.left_home, movement_minutes=movement, interrupt_walk_minutes=data.interrupt_walk_minutes, fantasy_walk_minutes=data.fantasy_walk_minutes, headphones_on_walk=data.headphones_on_walk, music_cinematic_on_walk=data.music_cinematic_on_walk, proof_brick_done=data.proof_brick_done, proof_brick_kind=data.proof_brick_kind, fantasy_minutes_scheduled=data.fantasy_minutes_scheduled, fantasy_minutes_unplanned=data.fantasy_minutes_unplanned, points=points, note=data.note, win=data.win, updated_at=_utc_now(), ) records = read_jsonl(self.paths.daily) out: list[dict] = [] replaced = False for record in records: if record.get("date") == day.isoformat(): out.append(row.model_dump(mode="json")) replaced = True else: out.append(record) if not replaced: out.append(row.model_dump(mode="json")) rewrite_jsonl(self.paths.daily, out) return row def range(self, start: date, end: date) -> tuple[list[DailyRow], int, str, int]: """Return rows in [start, end], week_points, band, and days_present.""" items = [row for row in self._load() if start <= row.date <= end] items.sort(key=lambda row: row.date) days_present = len(items) week_points = sum(row.points for row in items) band = week_band(days_present, week_points) return items, week_points, band, days_present def movement_for(self, as_of: date) -> dict[str, Any]: """Movement / indoors summary ending at as_of.""" return movement_summary(self._load(), as_of=as_of)