screen-locker/screen_locker/_phone_verification.py

"""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__)


def _in_current_week(date_str: str) -> bool:
    """True if *date_str* parses and falls in this ISO week up to today."""
    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 False
    return week_start <= entry_date <= today


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 or not _in_current_week(date_str):
            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 _pull_sync_result(self) -> tuple[tuple[str, str] | None, str | None]:
        """Firebase sync first: the validated result (if any) and sync error.

        Firebase sync is the primary channel -- it works without the phone
        being on the same network as the PC. Only a *verified* 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 by the caller as a fallback-of-last-resort
        candidate. 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()
        if synced_data is None:
            if sync_error is not None:
                _logger.info(
                    "Firebase sync unavailable (%s) — trying ADB/HTTP...", sync_error
                )
            return None, sync_error
        sync_result = self._validate_json_data(synced_data)
        if sync_result[0] != "verified":
            _logger.info(
                "Synced data not fully verified (%s) — trying ADB/HTTP...",
                sync_result[0],
            )
        return sync_result, sync_error

    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

        sync_result, sync_error = self._pull_sync_result()
        if sync_result is not None and sync_result[0] == "verified":
            return sync_result

        # 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)
        return _no_data_result(sync_result, sync_error, adb_connected=adb_connected)


def _no_data_result(
    sync_result: tuple[str, str] | None,
    sync_error: str | None,
    *,
    adb_connected: bool,
) -> tuple[str, str]:
    """The verdict when neither ADB nor HTTP produced a workout JSON."""
    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"