i2p-python / src / i2p_apps / setup / wizard.py

"""First-run configuration wizard — interactive and non-interactive modes.

Runs three-layer diagnostics and produces a router config:
  Layer 1: Local-only probes (CPU, RAM, disk)
  Layer 2: Passive network probes (STUN, DNS, RTT, bandwidth)
  Layer 3: Post-bootstrap (deferred to router runtime)
"""

from __future__ import annotations

import json
import logging
import os
from pathlib import Path

from i2p_apps.setup.config_advisor import advise
from i2p_apps.setup.local_probe import run_local_probes
from i2p_apps.setup.network_probe import run_network_probes
from i2p_apps.setup.stun_probe import detect_nat_type

log = logging.getLogger(__name__)

CONFIG_FILENAME = "router.config.json"


class SetupWizard:
    """First-run configuration wizard."""

    def __init__(self, data_dir: str = "~/.i2p-python"):
        self.data_dir = os.path.expanduser(data_dir)
        os.makedirs(self.data_dir, exist_ok=True)

    @property
    def config_path(self) -> str:
        return os.path.join(self.data_dir, CONFIG_FILENAME)

    def is_first_run(self) -> bool:
        """Check if router.config.json exists."""
        return not os.path.exists(self.config_path)

    def save_config(self, config: dict) -> None:
        """Save config to data_dir/router.config.json."""
        with open(self.config_path, "w") as f:
            json.dump(config, f, indent=2)
        log.info("Config saved to %s", self.config_path)

    def load_config(self) -> dict | None:
        """Load saved config, or None if first run."""
        if not os.path.exists(self.config_path):
            return None
        with open(self.config_path) as f:
            return json.load(f)

    def run_auto(self, mode: str = "normal") -> dict:
        """Run non-interactive wizard.

        Modes:
          - normal: all probes including network
          - paranoid: local probes only, no network traffic
          - performance: all probes, maximize resource allocation
        """
        log.info("Running setup wizard in %s mode", mode)

        # Layer 1: Local probes (always run)
        local = run_local_probes(data_dir=self.data_dir)
        log.info(
            "Local probes: %d crypto ops/s, %d MB RAM, %.0f MB/s disk",
            local.crypto_ops_per_sec,
            local.available_ram_bytes // (1024 * 1024),
            local.disk_write_mbps,
        )

        nat = None
        network = None

        if mode != "paranoid":
            # Layer 2a: STUN NAT detection
            nat = detect_nat_type()
            log.info("NAT probe: type=%s, external=%s:%s",
                     nat.nat_type, nat.external_ip, nat.external_port)

            # Layer 2b+2c: Network probes
            network = run_network_probes(
                include_bandwidth=(mode != "paranoid"),
            )
            log.info("Network probe: DNS=%s, RTT=%.0fms",
                     network.dns.can_resolve_neutral, network.latency.rtt_ms)

        # Generate recommendation
        recommendation = advise(local, nat=nat, network=network)

        config = recommendation.to_router_config_overrides()
        config["mode"] = mode
        config["version"] = 1

        if recommendation.warnings:
            for w in recommendation.warnings:
                log.warning("Setup: %s", w)

        self.save_config(config)
        return config

    def run_interactive(self) -> dict:
        """Run interactive CLI wizard. Returns config dict.

        For non-TTY environments, falls back to run_auto("normal").
        """
        if not os.isatty(0):
            return self.run_auto("normal")

        print("\nWelcome to I2P Python Router\n")
        print("Checking your system...\n")

        # Layer 1
        local = run_local_probes(data_dir=self.data_dir)
        print(f"  CPU: {local.crypto_ops_per_sec:,} crypto ops/sec")
        print(f"  RAM: {local.available_ram_bytes // (1024**3)} GB available")
        print(f"  Disk: {local.disk_write_mbps:.0f} MB/s, "
              f"{local.disk_free_bytes // (1024**3)} GB free\n")

        # Ask about network tests
        print("Would you like to test your network connection?")
        print("This uses standard protocols (STUN, HTTPS) that look like")
        print("normal video call and web browsing traffic.\n")
        print("[Y]es  [S]kip  [P]aranoid (no network tests at all)")

        choice = input("> ").strip().lower()

        nat = None
        network = None

        if choice in ("y", "yes", ""):
            nat = detect_nat_type()
            network = run_network_probes()
            print(f"\n  NAT: {nat.nat_type}")
            if nat.external_ip:
                print(f"  External IP: {nat.external_ip}")
            print(f"  DNS: {'OK' if network.dns.can_resolve_neutral else 'FAILED'}")
            print(f"  Latency: {network.latency.rtt_ms:.0f}ms median")
            if network.bandwidth:
                print(f"  Bandwidth: ~{network.bandwidth.download_mbps:.0f} Mbps")

        recommendation = advise(local, nat=nat, network=network)
        config = recommendation.to_router_config_overrides()

        print(f"\nRecommended configuration:")
        print(f"  Bandwidth share: {config['share_percentage']}%")
        print(f"  Transit tunnels: {config['inbound_tunnel_count']}")
        print(f"  Floodfill: {'yes' if config['floodfill'] else 'no'}")
        print(f"  Transport port: {config['listen_port']}")
        print(f"  NAT traversal: {'UPnP enabled' if config['upnp_enabled'] else 'disabled'}")

        for w in recommendation.warnings:
            print(f"  WARNING: {w}")

        print("\n[A]ccept  [Q]uit")
        accept = input("> ").strip().lower()

        if accept in ("a", "accept", ""):
            config["version"] = 1
            self.save_config(config)
            return config
        else:
            raise SystemExit("Setup cancelled.")