Function bodies 373 total

def setup_gmail_mfa() -> None:
    header("Gmail MFA Automation Setup")
    print("  When Garmin requires a security code (~every 30 days), this module")
    print("  polls your Gmail inbox and submits it automatically. Without it,")
    print("  you'll be prompted to paste the code manually.\n")
    hr()

    creds_file = ROOT / "google_credentials.json"
    if not creds_file.exists():
        print()
        console.print("  [red]✗[/]  google_credentials.json not found.\n")
        print("  To create it:")
        print("    1. Go to console.cloud.google.com")
        print("    2. Create a project")
        print("    3. Enable the Gmail API")
        print("    4. Credentials → Create → OAuth 2.0 Client ID → Desktop app")
        print("    5. Download JSON → save as google_credentials.json in this directory")
        print()
        print("  Then run this option again to complete authorization.")
        _continue()
        return

    print("  google_credentials.json found. Start

def _parse_date(s: str) -> str | None:
    """Accept YYYY-MM-DD, MM/DD/YYYY, MM/DD/YY, MM-DD-YYYY, MM-DD-YY."""
    for fmt in ("%Y-%m-%d", "%m/%d/%Y", "%m/%d/%y", "%m-%d-%Y", "%m-%d-%y"):
        try:
            return datetime.strptime(s.strip(), fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return None

def _run_pull(start_date: str, days: int, no_skip: bool = False) -> None:
    """Run the Garmin puller, then immediately rebuild CSVs."""
    cmd = [
        PYTHON,
        str(ROOT / "pullers" / "garmin.py"),
        "--date",
        start_date,
        "--days",
        str(days),
    ]
    if no_skip:
        cmd.append("--no-skip")

    subprocess.run(cmd)

    print()
    print("  Building CSV reports...")
    subprocess.run([PYTHON, str(ROOT / "reports" / "build_garmin_csvs.py")])
    print()
    console.print("  [green]✓[/]  Done.")
    print("  · reports/garmin_daily.csv")
    print("  · reports/garmin_activities.csv")

def _pull_yesterday() -> None:
    yesterday = (date.today() - timedelta(days=1)).isoformat()
    print(f"\n  Pulling {yesterday}...\n")
    _run_pull(yesterday, 1)
    _continue()

def _pull_last_n(n: int, label: str) -> None:
    start = (date.today() - timedelta(days=n)).isoformat()
    print(f"\n  Pulling {label} ({start} → yesterday)...\n")
    _run_pull(start, n)
    _continue()

def _pull_custom() -> None:
    header("Custom Date Pull")
    print("  Accepted date formats: YYYY-MM-DD  |  MM/DD/YYYY  |  MM/DD/YY\n")

    raw = prompt_with_navigation("  Start date: ")
    if not raw:
        return
    start = _parse_date(raw)
    if not start:
        print(f"\n  Could not parse '{raw}' as a date. Try: 2025-04-07 or 04/07/2025")
        _continue()
        return

    days_in = prompt_with_navigation("  Number of days to pull [Enter for 1]: ")
    days = int(days_in) if days_in.isdigit() and int(days_in) > 0 else 1

    end = (datetime.strptime(start, "%Y-%m-%d").date() + timedelta(days=days - 1)).isoformat()
    range_label = start if days == 1 else f"{start} → {end}"
    print(f"\n  Pulling {days} day(s): {range_label}")

    reskip = prompt_with_navigation("  Re-pull dates that already have data? [y/N]: ")
    print()
    _run_pull(start, days, no_skip=(reskip.lower() == "y"))
    _continue()

def _pull_everything() -> None:
    header("Pull Full History")
    print("  Pulls every day from a start date through yesterday.")
    print("  Use this to build a complete data history via the live API.\n")
    print("  Each day takes ~15 seconds to pull.")
    print("  · 1 month  ≈  7 minutes")
    print("  · 6 months ≈  45 minutes")
    print("  · 1 year   ≈  90 minutes\n")
    print("  For very large history pulls, the Garmin bulk export (option 6)")
    print("  is faster — request it from Garmin and import the .zip.\n")
    print("  Accepted date formats: YYYY-MM-DD  |  MM/DD/YYYY  |  MM/DD/YY\n")

    raw = prompt_with_navigation("  Pull data starting from: ")
    if not raw:
        return
    start_str = _parse_date(raw)
    if not start_str:
        print(f"\n  Could not parse '{raw}'. Try: 2023-01-01 or 01/01/2023")
        _continue()
        return

    start_dt = datetime.strptime(start_str, "%Y-%m-%d").date()
    yesterday = date.today() - timedelta(days=1)
    days = (

def import_export() -> None:
    header("Import from Garmin Bulk Export")
    print("  The Garmin bulk export is the fastest way to load years of")
    print("  historical data. Request it at:")
    print("  Garmin Connect → Profile → Account → Your Garmin Data")
    print("  (The .zip file arrives within 24–48 hours.)\n")

    zip_path = prompt_with_navigation("  Path to export .zip: ").strip('"').strip("'")
    if not zip_path:
        print("  Cancelled.")
        _continue()
        return

    if not Path(zip_path).exists():
        print(f"\n  File not found: {zip_path}")
        _continue()
        return

    reskip = prompt_with_navigation("  Overwrite dates that already have data? [y/N]: ")
    cmd = [PYTHON, str(ROOT / "pullers" / "garmin_import_export.py"), zip_path]
    if reskip.lower() == "y":
        cmd.append("--no-skip")

    print()
    subprocess.run(cmd)

    print()
    print("  Building CSV reports...")
    subprocess.run([PYTHON, str(ROOT / "reports" / "build_g

def build_csvs() -> None:
    header("Build CSV Reports")
    print("  Flattens all JSON data files into:")
    print("  · reports/garmin_daily.csv       — one row per day")
    print("  · reports/garmin_activities.csv  — one row per workout\n")

    since = prompt_with_navigation("  Include data from [Enter for all time, or YYYY-MM-DD]: ")
    cmd = [PYTHON, str(ROOT / "reports" / "build_garmin_csvs.py")]
    if since:
        cmd += ["--since", since]

    print()
    run(cmd)
    _continue()

def _first_run_notice() -> None:
    if not ENV.exists():
        header("garmin-extract")
        print("  Looks like your first run — .env not found.")
        print()
        print("  Start with option 1 (Initial Setup) to get everything")
        print("  installed and configured before pulling data.")
        _continue("\n  Press Enter to open the menu...")

def _submenu(title: str, options: list[MenuOption]) -> None:
    """Generic sub-menu loop. BackSignal from prompt exits; others propagate up."""
    keys: dict[str, Callable[[], None]] = {k: fn for k, label, fn in options if fn is not None}
    while True:
        header(title)
        for key, label, fn in options:
            if fn is None:
                print(f"  {label}")
            else:
                print(f"    {key}  {label}")
        print()
        hr()
        print("    b  Back    x  Main menu    q  Quit")
        hr()
        try:
            choice = prompt_with_navigation("\n  Choice: ").lower()
        except BackSignal:
            return
        # ExitToMainSignal and QuitSignal propagate naturally

        if choice in keys:
            try:
                keys[choice]()
            except BackSignal:
                pass  # stay in this submenu; action was aborted
            # ExitToMainSignal and QuitSignal propagate
        else:
            print("  Unreco

def menu_initial_setup() -> None:
    _submenu(
        "Initial Setup",
        [
            ("1", "Setup wizard  (prerequisites + credentials)", check_prerequisites),
            ("2", "Update Garmin credentials", configure_credentials),
        ],
    )

def menu_pull_data() -> None:
    _submenu(
        "Pull Data",
        [
            ("", "─── Recent ──────────────────────────────────────────", None),
            ("1", "Yesterday", _pull_yesterday),
            ("2", "Last 7 days", lambda: _pull_last_n(7, "last 7 days")),
            ("3", "Last 30 days", lambda: _pull_last_n(30, "last 30 days")),
            ("", "─── Custom ──────────────────────────────────────────", None),
            ("4", "Specific date or date range", _pull_custom),
            ("5", "Full history  (from a date you choose to today)", _pull_everything),
            ("", "─── Historical import ───────────────────────────────", None),
            ("6", "Import from Garmin bulk export (.zip)", import_export),
            ("", "─── Reports ─────────────────────────────────────────", None),
            ("7", "Rebuild CSV reports  (from existing pulled data)", build_csvs),
        ],
    )

def menu_automation() -> None:
    _submenu(
        "Configure Automation",
        [
            ("", "─── Unattended MFA ─────────────────────────────────", None),
            ("1", "Set up Gmail MFA  (auto-handle Garmin security codes)", setup_gmail_mfa),
        ],
    )

def main(dry_run: bool = False, verbose: int = 0) -> None:
    # TODO Phase 2: thread dry_run through action functions
    _first_run_notice()

    actions: dict[str, Callable[[], None]] = {
        "1": menu_initial_setup,
        "2": menu_pull_data,
        "3": menu_automation,
    }

    while True:
        header("garmin-extract")
        print("    1  Initial Setup")
        print("    2  Pull Data")
        print("    3  Configure Automation")
        print()
        hr()
        print("    q  Quit")
        hr()

        try:
            choice = prompt_with_navigation("\n  Choice: ").lower()
        except (BackSignal, ExitToMainSignal):
            continue  # no-op at main menu level
        except QuitSignal:
            break

        if choice in actions:
            try:
                actions[choice]()
            except ExitToMainSignal:
                pass  # return to main menu loop
            except QuitSignal:
                break
        else:
            pri

def app_root() -> Path:
    """Return the directory where user-facing files (config, data, scripts) live.

    - Dev install: project root (containing pyproject.toml)
    - Frozen (PyInstaller onedir): the directory containing the .exe
    """
    if getattr(sys, "frozen", False):
        return Path(sys.executable).parent
    # Dev install: this file is at <root>/garmin_extract/_paths.py
    return Path(__file__).resolve().parent.parent

def bundle_root() -> Path:
    """Return the directory containing bundled data files (pullers/, reports/, scripts/).

    - Dev install: same as app_root() — the project root
    - Frozen: `sys._MEIPASS` (PyInstaller's _internal directory) where data files land
    """
    if getattr(sys, "frozen", False):
        meipass = getattr(sys, "_MEIPASS", None)
        if meipass:
            return Path(meipass)
        return Path(sys.executable).parent / "_internal"
    return Path(__file__).resolve().parent.parent

def _build_menu(cursor: int = 0) -> str:
    top = "  ┌" + "─" * _W + "┐"
    bottom = "  └" + "─" * _W + "┘"
    rows = ["\n" + top, _EMPTY_ROW]
    for i, (key, label, hint) in enumerate(_ITEMS):
        sel = i == cursor
        cur = "❯" if sel else " "
        lbl = f"[bold]{label}[/]" if sel else label
        h = f"[bold]{hint}[/]" if sel else hint
        lbl_pad = " " * (_W - 8 - len(label))
        hint_pad = " " * (_W - 8 - len(hint))
        rows.append(f"  │ {cur} [bold cyan][{key}][/]  {lbl}{lbl_pad}│")
        rows.append(f"  │        [dim]{h}[/]{hint_pad}│")
        rows.append(_EMPTY_ROW)
    rows.append(bottom)
    return "\n".join(rows)

def _read_crontab() -> str:
    """Return current crontab contents, or empty string if none set."""
    try:
        result = subprocess.run(
            ["crontab", "-l"],
            capture_output=True,
            text=True,
        )
        return result.stdout if result.returncode == 0 else ""
    except Exception:
        return ""

def _write_crontab(contents: str) -> bool:
    """Write new crontab contents. Returns True on success."""
    try:
        result = subprocess.run(
            ["crontab", "-"],
            input=contents,
            capture_output=True,
            text=True,
        )
        return result.returncode == 0
    except Exception:
        return False

def _find_cron_entry(crontab: str) -> str | None:
    """Return the garmin-extract cron line, or None if not installed."""
    for line in crontab.splitlines():
        if _CRON_MARKER in line:
            return line
    return None

def _install_cron(hour: int) -> tuple[bool, str]:
    """Install or replace the garmin-extract cron entry."""
    crontab = _read_crontab()
    lines = [ln for ln in crontab.splitlines() if _CRON_MARKER not in ln]
    lines.append(_build_cron_entry(hour))
    new_contents = "\n".join(lines) + "\n"
    ok = _write_crontab(new_contents)
    return ok, new_contents if ok else "Failed to write crontab"

def _remove_cron() -> tuple[bool, str]:
    """Remove the garmin-extract cron entry."""
    crontab = _read_crontab()
    lines = [ln for ln in crontab.splitlines() if _CRON_MARKER not in ln]
    new_contents = ("\n".join(lines) + "\n") if lines else ""
    ok = _write_crontab(new_contents)
    return ok, "Removed" if ok else "Failed to remove"

def _check_gmail_automation() -> tuple[str, str]:
    """Return (status, detail) where status is 'ok' | 'partial' | 'unconfigured'."""
    has_creds = GMAIL_CREDS_FILE.exists()
    has_token = GMAIL_TOKEN_FILE.exists()

    if not has_creds and not has_token:
        return "unconfigured", "Not set up"
    if not has_creds:
        return "partial", "google_credentials.json missing — re-run Gmail OAuth setup"
    if not has_token:
        return "partial", "Not yet authorized — run Gmail OAuth setup to generate token"

    try:
        import json as _json

        from google.auth.transport.requests import Request
        from google.oauth2.credentials import Credentials

        tok = _json.loads(GMAIL_TOKEN_FILE.read_text())
        creds = Credentials(
            token=tok.get("token"),
            refresh_token=tok.get("refresh_token"),
            token_uri=tok.get("token_uri"),
            client_id=tok.get("client_id"),
            client_secret=tok.get("client_secret"),
     

class AutomationScreen(Screen[None]):
    """Automation landing — Gmail MFA status, cron schedule, Drive/Sheets."""

    _ITEM_COUNT = 3

    BINDINGS = [
        Binding("1", "go_gmail", show=False),
        Binding("2", "go_cron", show=False),
        Binding("3", "go_sheets", show=False),
        Binding("up", "cursor_up", show=False),
        Binding("k", "cursor_up", show=False),
        Binding("down", "cursor_down", show=False),
        Binding("j", "cursor_down", show=False),
        Binding("enter", "cursor_select", show=False),
        Binding("b", "back", "Back", show=True),
        Binding("q", "quit", "Quit", show=True),
    ]

    CSS = """
    AutomationScreen {
        align: center middle;
    }

    #auto-header {
        width: 57;
        text-align: center;
        color: $accent;
        text-style: bold;
        margin-bottom: 1;
    }

    #auto-options {
        width: 57;
        color: $text;
    }

    #auto-hint {
        width: 57;
        text-align: cent

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("Automation", id="auto-header")
        yield Static(_build_menu(0), id="auto-options")
        yield Static(
            "↑↓  j/k  navigate  ·  enter  select  ·  1–3  direct  ·  b  back",
            id="auto-hint",
        )
        yield Footer()

    def action_cursor_up(self) -> None:
        if self._cursor > 0:
            self._cursor -= 1
            self._refresh_menu()

    def action_cursor_down(self) -> None:
        if self._cursor < self._ITEM_COUNT - 1:
            self._cursor += 1
            self._refresh_menu()

    def action_go_gmail(self) -> None:
        self._cursor = 0
        self.app.push_screen(GmailMfaScreen())

    def action_go_cron(self) -> None:
        self._cursor = 1
        self.app.push_screen(CronScreen())

    def action_go_sheets(self) -> None:
        self._cursor = 2
        from garmin_extract.screens.drive_sheets import DriveSheetsScreen

        self.app.push_screen(DriveSheetsScreen())

class GmailMfaScreen(Screen[None]):
    """Gmail MFA automation status screen."""

    BINDINGS = [
        Binding("s", "go_setup", "Go to Setup", show=False),
        Binding("b", "back", "Back", show=True),
        Binding("q", "quit", "Quit", show=True),
    ]

    CSS = """
    GmailMfaScreen {
        layout: vertical;
        padding: 2 4;
    }

    #gmail-mfa-header {
        text-style: bold;
        color: $accent;
        margin-bottom: 2;
    }

    #gmail-mfa-creds {
        height: auto;
        margin-bottom: 1;
    }

    #gmail-mfa-token {
        height: auto;
        margin-bottom: 1;
    }

    #gmail-mfa-status {
        height: auto;
        margin-bottom: 2;
    }

    #gmail-mfa-hint {
        color: $text-muted;
        height: auto;
    }
    """

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("Gmail MFA Automation", id="gmail-mfa-header")
        yield Static("Checking…", id="gmail-mfa-creds")
        yield

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("Gmail MFA Automation", id="gmail-mfa-header")
        yield Static("Checking…", id="gmail-mfa-creds")
        yield Static("", id="gmail-mfa-token")
        yield Static("", id="gmail-mfa-status")
        yield Static("", id="gmail-mfa-hint")
        yield Footer()

    def _check(self) -> None:
        has_creds = GMAIL_CREDS_FILE.exists()
        has_token = GMAIL_TOKEN_FILE.exists()
        status, detail = _check_gmail_automation()

        creds_text = (
            "[green]✓[/] google_credentials.json found"
            if has_creds
            else "[red]✗[/] google_credentials.json not found"
        )
        token_text = (
            "[green]✓[/] Authorization token present"
            if has_token
            else "[yellow]○[/] Not yet authorized"
        )

        if status == "ok":
            status_text = (
                "[green]● Gmail automation is active[/]"
                "  —  MFA codes will be fetched automatically"
            )
            hint_text = ""
            show_setup = False
        else:
            status_text = (
                "[dim]Not configured[/]" if status == "unconfigured" else f"[yellow]⚠[/]  {detail}"
            )
            hint_text = "Press  [bold]s[/]  to go to Initial Setup → Gmail OAuth"

    def _apply_state(
        self,
        creds_text: str,
        token_text: str,
        status_text: str,
        hint_text: str,
        show_setup: bool,
    ) -> None:
        self.query_one("#gmail-mfa-creds", Static).update(creds_text)
        self.query_one("#gmail-mfa-token", Static).update(token_text)
        self.query_one("#gmail-mfa-status", Static).update(status_text)
        self.query_one("#gmail-mfa-hint", Static).update(hint_text)
        self.BINDINGS = [
            Binding("s", "go_setup", "Go to Setup", show=show_setup),
            Binding("b", "back", "Back", show=True),
            Binding("q", "quit", "Quit", show=True),
        ]
        self.refresh_bindings()

    def action_go_setup(self) -> None:
        from garmin_extract.screens.setup import SetupScreen

        self.app.push_screen(SetupScreen())

class _EditTimeModal(ModalScreen[int | None]):
    """Modal to pick cron hour (0–23)."""

    BINDINGS = [Binding("escape", "dismiss", "Cancel", show=True)]

    CSS = """
    _EditTimeModal {
        align: center middle;
    }

    #edit-time-box {
        width: 52;
        height: auto;
        border: round $accent;
        padding: 1 2;
    }

    #edit-time-title {
        text-style: bold;
        color: $accent;
        margin-bottom: 1;
    }

    #edit-time-hint {
        color: $text-muted;
        margin-bottom: 1;
    }

    #edit-time-error {
        color: $error;
        height: 1;
    }
    """

    def __init__(self, current_hour: int = 6) -> None:
        super().__init__()
        self._current_hour = current_hour

    def compose(self) -> ComposeResult:
        with Static(id="edit-time-box"):
            yield Static("Set Pull Time", id="edit-time-title")
            yield Static(
                f"Enter hour in 24h format (0–23).  Current: {self._current_hour:02

    def __init__(self, current_hour: int = 6) -> None:
        super().__init__()
        self._current_hour = current_hour

    def compose(self) -> ComposeResult:
        with Static(id="edit-time-box"):
            yield Static("Set Pull Time", id="edit-time-title")
            yield Static(
                f"Enter hour in 24h format (0–23).  Current: {self._current_hour:02d}:00",
                id="edit-time-hint",
            )
            yield Input(
                placeholder="6",
                value=str(self._current_hour),
                id="edit-time-input",
                max_length=2,
            )
            yield Static("", id="edit-time-error")

    def on_input_submitted(self, event: Input.Submitted) -> None:
        try:
            hour = int(event.value.strip())
            if not 0 <= hour <= 23:
                raise ValueError
        except ValueError:
            self.query_one("#edit-time-error", Static).update(
                "Enter a whole number between 0 and 23."
            )
            return
        self.dismiss(hour)

class CronScreen(Screen[None]):
    """Cron schedule management screen."""

    BINDINGS = [
        Binding("i", "install", "Enable / Re-enable", show=True),
        Binding("e", "edit_time", "Change Time", show=True),
        Binding("r", "remove", "Disable", show=True),
        Binding("b", "back", "Back", show=True),
        Binding("q", "quit", "Quit", show=True),
    ]

    CSS = """
    CronScreen {
        layout: vertical;
        padding: 2 4;
    }

    #cron-header {
        text-style: bold;
        color: $accent;
        margin-bottom: 2;
    }

    #cron-status {
        height: auto;
        margin-bottom: 1;
    }

    #cron-entry {
        color: $text-muted;
        height: auto;
        margin-bottom: 2;
    }

    #cron-feedback {
        height: 1;
    }
    """

    def __init__(self) -> None:
        super().__init__()
        self._installed = False
        self._current_hour = 6

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
  

    def __init__(self) -> None:
        super().__init__()
        self._installed = False
        self._current_hour = 6

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("Scheduled Pulls", id="cron-header")
        yield Static("Checking schedule…", id="cron-status")
        yield Static("", id="cron-entry")
        yield Static("", id="cron-feedback")
        yield Footer()

    def _check_cron(self) -> None:
        crontab = _read_crontab()
        entry = _find_cron_entry(crontab)

        if entry:
            try:
                hour = int(entry.split()[1])
            except (IndexError, ValueError):
                hour = 6
            self.app.call_from_thread(self._apply_state, True, hour, entry.strip())
        else:
            self.app.call_from_thread(self._apply_state, False, 6, "")

    def _apply_state(self, installed: bool, hour: int, entry: str) -> None:
        self._installed = installed
        self._current_hour = hour

        if installed:
            self.query_one("#cron-status", Static).update(
                f"[green]● Active[/]  —  pulls data every day at [bold]{hour:02d}:00[/]"
            )
            self.query_one("#cron-entry", Static).update(
                "[dim]Output is logged to  /tmp/garmin-pull.log[/]"
            )
        else:
            self.query_one("#cron-status", Static).update("[dim]Not scheduled[/]")
            self.query_one("#cron-entry", Static).update("[dim]Default pull time: 6:00 AM daily[/]")

    def action_install(self) -> None:
        ok, _ = _install_cron(self._current_hour)
        feedback = (
            f"[green]Scheduled — will pull data every day at {self._current_hour:02d}:00[/]"
            if ok
            else "[red]Failed to save schedule — check system permissions[/]"
        )
        self.query_one("#cron-feedback", Static).update(feedback)
        self._refresh_cron_display()

    def action_remove(self) -> None:
        if not self._installed:
            self.query_one("#cron-feedback", Static).update(
                "[dim]No schedule is set — nothing to disable.[/]"
            )
            return
        ok, _ = _remove_cron()
        feedback = (
            "[dim]Schedule disabled — automatic pulls are turned off.[/]"
            if ok
            else "[red]Failed to remove schedule — check system permissions[/]"
        )
        self.query_one("#cron-feedback", Static).update(feedback)
        self._refresh_cron_display()

    def action_edit_time(self) -> None:
        if not self._installed:
            self.query_one("#cron-feedback", Static).update(
                "[dim]Enable a schedule first, then use  e  to change the time.[/]"
            )
            return
        self.app.push_screen(_EditTimeModal(self._current_hour), self._on_hour_selected)

    def _on_hour_selected(self, hour: int | None) -> None:
        if hour is None:
            return
        ok, _ = _install_cron(hour)
        feedback = (
            f"[green]Updated — will pull data every day at {hour:02d}:00[/]"
            if ok
            else "[red]Failed to update schedule — check system permissions[/]"
        )
        self.query_one("#cron-feedback", Static).update(feedback)
        self._refresh_cron_display()

def _date_range_label(days: int) -> str:
    today = date.today()
    start = today - timedelta(days=days)
    end = today - timedelta(days=1)
    return f"{start.isoformat()}  →  {end.isoformat()}"