"""Phone workout verification: Firebase sync first, ADB/HTTP as fallback.""" from __future__ import annotations from datetime import UTC, date, datetime, timedelta from http import client as _http_client import json import logging from pathlib import Path import tempfile from screen_locker._adb_transport import AdbTransportMixin from screen_locker._constants import ( MIN_WORKOUT_DURATION_MINUTES, WORKOUT_APP_JSON_REMOTES, WORKOUT_DURATION_ACCEPT_MINUTES, ) from screen_locker._day import today_str from screen_locker._http_workout_fetch import HttpWorkoutFetchMixin from screen_locker._log_mixin import write_signed_entry from screen_locker._time_check import check_clock_skew from screen_locker._workout_sync import pull_synced_workout _HTTPConnection = _http_client.HTTPConnection _HTTPException = _http_client.HTTPException _HTTP_OK = _http_client.OK _logger = logging.getLogger(__name__) class PhoneVerificationMixin(AdbTransportMixin, HttpWorkoutFetchMixin): """Mixin providing phone-based workout verification via ADB.""" def _pull_workout_app_json(self) -> dict | None: """Pull workout_result.json from the phone (ADB, no root needed). The app writes to one of several candidate paths (see sync_service.dart), so we pull every candidate and prefer the one dated today. A stale file can linger at a fallback path from a day the primary write failed, so "first that parses" is not safe — we explicitly favour today's data and only fall back to a stale/older payload if no candidate is from today. """ tmp = Path(tempfile.gettempdir()) / "workout_result.json" today = today_str() first_parsed: dict | None = None for remote in WORKOUT_APP_JSON_REMOTES: ok, _ = self._run_adb(["pull", remote, str(tmp)]) if not ok: continue try: data = json.loads(tmp.read_text()) except (json.JSONDecodeError, OSError) as exc: _logger.warning( "Workout JSON pulled from %s is unreadable (%s) — skipping " "that candidate and trying the remaining phone paths", remote, exc, ) continue if data.get("date") == today: return data if first_parsed is None: first_parsed = data return first_parsed def _validate_json_content(self, data: dict) -> tuple[str, str]: """Validate a workout JSON's exercises/duration, regardless of its date. Split out of :meth:`_validate_json_data` so the week-scan backfill (``_try_fill_stronglifts_for_week``) can reuse the same content checks against a past-in-week date without the today-only gate. """ if not data.get("exercises"): return "no_exercises", "No exercises found in the workout JSON" duration_min = data.get("duration_seconds", 0) / 60.0 # Accept bar carries the hidden leeway; the message advertises the # round number only (see _constants.py). if duration_min < WORKOUT_DURATION_ACCEPT_MINUTES: return ( "too_short", f"Workout too short! {duration_min:.0f} min logged, " f"need at least {MIN_WORKOUT_DURATION_MINUTES} min.", ) flag = "all succeeded" if data.get("succeeded") else "partial" return "verified", f"Workout verified! ({duration_min:.0f} min, {flag})" def _validate_json_data(self, data: dict) -> tuple[str, str]: """Validate parsed workout JSON. Returns (status, message). Requires today's date — this is the live unlock-chain check, which must reflect a workout done *right now*. For crediting a workout found dated earlier in the week, see ``_try_fill_stronglifts_for_week``. """ today = today_str() if data.get("date") != today: return "stale", f"Workout JSON is from {data.get('date')}, not today" return self._validate_json_content(data) def _try_fill_stronglifts_for_week(self, log_file: Path) -> int: """Backfill a past-this-week StrongLifts workout found on the phone. Unlike RunnerUp's per-activity TCX exports, the phone holds only one ``workout_result.json`` snapshot at a time, so this fills at most one day per call — that mirrors :meth:`_scan_and_fill_week_runnerup`'s contract (0 when nothing new to fill) without pretending there could be more than one candidate. Still pulls and validates the real file via ADB before writing anything — never trusts a claim. Returns 1 if a new verified entry was appended, else 0 (also 0 when no phone is connected, no JSON is available, or its date falls outside the current ISO week). """ if not self._has_adb_device(): _logger.info( "Phone not connected; skipping StrongLifts week-scan backfill." ) return 0 data = self._pull_workout_app_json() date_str = data.get("date") if data else None if not date_str: return 0 today = datetime.now(tz=UTC).astimezone().date() week_start = today - timedelta(days=today.weekday()) try: entry_date = date.fromisoformat(date_str) except ValueError: _logger.warning("StrongLifts JSON has an unparsable date: %r", date_str) return 0 if not week_start <= entry_date <= today: return 0 status, msg = self._validate_json_content(data) if status != "verified": return 0 workout_data = { "type": "phone_verified", "source": f"Auto-scanned: {msg}", "duration_minutes": round(data.get("duration_seconds", 0) / 60.0, 1), } if write_signed_entry(log_file, date_str, workout_data).appended: _logger.info("Auto-filled StrongLifts entry for %s: %s", date_str, msg) return 1 return 0 # ── HTTP fallback (no ADB / developer options required) ─────────────────── def _verify_phone_workout(self) -> tuple[str, str]: """Verify today's workout: Firebase sync first, ADB/HTTP as fallback. Returns (status, message). Status values: verified / too_short / not_verified / no_phone / sync_failed / stale / no_exercises / clock_tampered. """ clock_ok, clock_msg = check_clock_skew() if not clock_ok: return "clock_tampered", clock_msg # Firebase sync is the primary channel — it works without the phone # being on the same network as the PC. Only a *verified* sync result # short-circuits ADB/HTTP: a stale/incomplete cloud entry (e.g. the # phone's last push predates today, or a later push failed offline) # must not preempt a fresher ADB pull, so anything less than # "verified" is kept as a fallback-of-last-resort candidate instead # of being returned immediately. A sync error never blocks unlocking # either; `sync_failed` only surfaces if ADB/HTTP also come up empty. synced_data, sync_error = pull_synced_workout() sync_result: tuple[str, str] | None = None if synced_data is not None: sync_result = self._validate_json_data(synced_data) if sync_result[0] == "verified": return sync_result _logger.info( "Synced data not fully verified (%s) — trying ADB/HTTP...", sync_result[0], ) elif sync_error is not None: _logger.info( "Firebase sync unavailable (%s) — trying ADB/HTTP...", sync_error ) # Prefer ADB when a device is visible, but if the pull yields no usable # JSON, fall through to the HTTP/WiFi scan — the app's in-memory HTTP # server may still hold today's workout even when the file pull fails. adb_connected = self._is_phone_connected() if adb_connected: data = self._pull_workout_app_json() if data is not None: return self._validate_json_data(data) _logger.info("ADB pull found no workout JSON — trying HTTP scan...") else: _logger.info("No ADB device — trying HTTP scan on local network...") data = self._fetch_http_workout() if data is not None: return self._validate_json_data(data) if sync_result is not None: return sync_result if sync_error is not None: return "sync_failed", sync_error if adb_connected: return ( "not_verified", "Workout app JSON not found. Complete a workout in the app first.", ) return "no_phone", "Phone not reachable via ADB or HTTP on local network"