screen-locker/screen_locker/_sick_tracker.py
Krzysztof kuhy Rudnicki 094a1fe5ed Split _workout_sync.py and _sick_tracker.py
_workout_sync.py (332) -> _sync_client.py   token, Firebase/GitHub client
                            _sync_records.py  payload predicates + log decoding
  _sick_tracker.py (317) -> _sick_budget.py   SickHistory + rolling budget

This one was not a pure move until the fixtures followed it. Three autouse
test guards patch names by module, and all three pointed at _workout_sync:

  isolate_sync_token      SYNC_TOKEN_FILE -> now also on _sync_client
  _no_real_firebase_config CONFIG_FILE    -> now on _sync_client
  _firebase_config helper  (7 call sites) -> now passed _sync_client

With them stale, sync_client built a REAL client: the suite went from 14s to
over 120s and test_workout_sync_part5 asserted against live workout data
pulled from the actual backend. Fixed, the same tests run in 0.7s. Both
modules keep SYNC_TOKEN_FILE so either patch point works, and every relocated
guard target is asserted to exist.

977 tests pass at 100% branch coverage; pylint back to 8.70 after fixing the
findings the split introduced (a dropped `return None` in _parse_iso and
leftover dead imports). `screen_lock --status` unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EveB813BQ8abAP68VSqZ5Y
2026-08-15 20:30:42 +02:00

240 lines
6.7 KiB
Python

"""Sick-day rate-limiting, workout debt, commitment, and justification tracking.
Pure logic — no Tk imports. The UI calls into these helpers and persists
state via :func:`save_history`.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import json
import logging
from typing import Any
from gatelock.log_integrity import compute_entry_hmac
from screen_locker._constants import (
SICK_COMMITMENT_PENALTY_DAYS,
SICK_HISTORY_FILE,
SICK_HISTORY_REVIEW_COUNT,
SICK_JUSTIFICATION_MIN_CHARS,
)
from screen_locker._sick_budget import (
SickHistory,
_parse_iso,
_today_iso,
budget_summary,
compute_lockout_seconds,
count_in_window,
is_budget_exhausted,
is_sick_day,
)
__all__ = [
"SickHistory",
"add_sick_day",
"budget_summary",
"clear_one_debt",
"compute_lockout_seconds",
"count_in_window",
"had_commitment_for_today",
"is_budget_exhausted",
"is_sick_day",
"load_history",
"mark_commitment_broken",
"record_commitment_for_tomorrow",
"save_history",
]
_logger = logging.getLogger(__name__)
def load_history() -> SickHistory:
"""Read the persistent sick-day history file.
Missing or unreadable files yield an empty :class:`SickHistory`.
"""
if not SICK_HISTORY_FILE.exists():
return SickHistory()
try:
with SICK_HISTORY_FILE.open() as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
_logger.warning("Could not read sick history; starting fresh")
return SickHistory()
return SickHistory(
sick_days=list(data.get("sick_days", [])),
debt=int(data.get("debt", 0)),
commitments=dict(data.get("commitments", {})),
broken_commitments=list(data.get("broken_commitments", [])),
justifications=list(data.get("justifications", [])),
)
def save_history(history: SickHistory) -> bool:
"""Persist ``history``. Returns True on success."""
payload = {
"sick_days": history.sick_days,
"debt": history.debt,
"commitments": history.commitments,
"broken_commitments": history.broken_commitments,
"justifications": history.justifications,
}
try:
with SICK_HISTORY_FILE.open("w") as f:
json.dump(payload, f, indent=2)
except OSError as exc:
_logger.warning("Failed to save sick history: %s", exc)
return False
return True
def add_sick_day(history: SickHistory, *, today: str | None = None) -> int:
"""Append today's date and increment debt. Returns new debt.
If today appears in ``broken_commitments`` the debt grows by
:data:`SICK_COMMITMENT_PENALTY_DAYS` instead of 1.
"""
today_str = today or _today_iso()
if today_str not in history.sick_days:
history.sick_days.append(today_str)
increment = (
SICK_COMMITMENT_PENALTY_DAYS if today_str in history.broken_commitments else 1
)
history.debt += increment
return history.debt
def clear_one_debt(history: SickHistory) -> int:
"""Decrement debt by one (clamped at zero). Returns new debt."""
if history.debt > 0:
history.debt -= 1
return history.debt
def record_commitment_for_tomorrow(
history: SickHistory,
*,
today: str | None = None,
) -> str:
"""Record that the user committed to working out tomorrow.
Returns the ISO date for tomorrow.
"""
today_str = today or _today_iso()
today_dt = _parse_iso(today_str)
if today_dt is None:
return today_str
tomorrow = (today_dt + timedelta(days=1)).strftime("%Y-%m-%d")
history.commitments[tomorrow] = True
return tomorrow
def had_commitment_for_today(
history: SickHistory,
*,
today: str | None = None,
) -> bool:
"""Return True if a commitment exists for today."""
today_str = today or _today_iso()
return bool(history.commitments.get(today_str, False))
def mark_commitment_broken(
history: SickHistory,
*,
today: str | None = None,
) -> None:
"""Mark today's commitment as broken (idempotent)."""
today_str = today or _today_iso()
if today_str in history.commitments and today_str not in history.broken_commitments:
history.broken_commitments.append(today_str)
SICK_SEVERITY_MIN = 1
SICK_SEVERITY_MAX = 10
@dataclass
class JustificationDraft:
"""User-supplied justification fields for a sick-day request."""
symptom: str
onset: str
severity: int
text: str
def validate_justification(draft: JustificationDraft) -> str | None:
"""Return an error message if the justification is invalid, else None."""
if not draft.symptom.strip():
return "Symptom is required"
if not draft.onset.strip():
return "Onset time is required"
if not SICK_SEVERITY_MIN <= draft.severity <= SICK_SEVERITY_MAX:
return f"Severity must be between {SICK_SEVERITY_MIN} and {SICK_SEVERITY_MAX}"
if len(draft.text.strip()) < SICK_JUSTIFICATION_MIN_CHARS:
return (
f"Description must be at least "
f"{SICK_JUSTIFICATION_MIN_CHARS} characters "
f"(currently {len(draft.text.strip())})"
)
return None
def add_justification(
history: SickHistory,
draft: JustificationDraft,
*,
today: str | None = None,
) -> dict[str, Any]:
"""HMAC-sign and append a sick-day justification.
Returns the stored entry (with ``hmac`` field if a key was available).
"""
today_str = today or _today_iso()
entry: dict[str, Any] = {
"date": today_str,
"timestamp": datetime.now(tz=timezone.utc).isoformat(),
"symptom": draft.symptom.strip(),
"onset": draft.onset.strip(),
"severity": int(draft.severity),
"text": draft.text.strip(),
}
signature = compute_entry_hmac(entry)
if signature is not None:
entry["hmac"] = signature
history.justifications.append(entry)
return entry
def recent_justifications(
history: SickHistory,
n: int = SICK_HISTORY_REVIEW_COUNT,
) -> list[dict[str, Any]]:
"""Return the last ``n`` justifications (oldest first)."""
if n <= 0:
return []
return list(history.justifications[-n:])
def format_recent_justifications(
history: SickHistory,
n: int = SICK_HISTORY_REVIEW_COUNT,
) -> str:
"""Human-readable multi-line summary of recent justifications.
Empty string when there are no past entries.
"""
entries = recent_justifications(history, n)
if not entries:
return ""
lines: list[str] = []
for entry in entries:
date_str = entry.get("date", "?")
symptom = entry.get("symptom", "?")
severity = entry.get("severity", "?")
lines.append(f"{date_str} sev {severity}/10 — {symptom}")
return "\n".join(lines)