#!/usr/bin/python3
"""motel-telemetry — opt-in usage reporting.

The single most important property of this program: with no consent recorded,
it exits before doing anything else. Not before sending — before reading
hardware, before resolving DNS, before touching the network stack at all. That
is asserted by the test suite, because "we promise it is off" is not a claim
anyone should have to take on faith.

Everything collected is listed in the privacy document shipped alongside this
file, and the same list is rendered in the welcome application's interface.
"""
from __future__ import annotations

import argparse
import json
import logging
import os
import re
import sys
import uuid
from pathlib import Path

log = logging.getLogger("motel-telemetry")

CONFIG_PATH = Path(
    os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")
) / "motel/telemetry.conf"

STATE_PATH = Path(
    os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share")
) / "motel/telemetry-id"

# https://telemetry.motelos.systems/v1/report is substituted at package build time from brand.yaml. The
# environment variable still wins, which is what the test suite uses.
ENDPOINT = os.environ.get("MOTEL_TELEMETRY_ENDPOINT", "https://telemetry.motelos.systems/v1/report")

REQUEST_TIMEOUT = 10


# ---------------------------------------------------------------------------
# Consent
# ---------------------------------------------------------------------------
def consent_given() -> bool:
    """True only if the user explicitly opted in.

    Every failure mode returns False. A missing file, an unreadable file, a
    malformed file and a permissions error all mean "no". Consent is never
    inferred from the absence of a refusal.
    """
    try:
        text = CONFIG_PATH.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return False

    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("#"):
            continue
        if stripped == "enabled=true":
            return True
    return False


# ---------------------------------------------------------------------------
# Identity
# ---------------------------------------------------------------------------
def install_id(persist: bool = True) -> str:
    """A random identifier, generated locally, tied to nothing else.

    Deleting the file severs the link to previous reports, which is documented
    as the way to reset it.

    With persist=False nothing is read and nothing is written: a throwaway id
    is returned. That is the path --show takes, because inspecting what would
    be sent must not leave durable identity on a machine that never opted in.
    """
    if not persist:
        return str(uuid.uuid4())

    try:
        existing = STATE_PATH.read_text(encoding="utf-8").strip()
        if re.fullmatch(r"[0-9a-f-]{36}", existing):
            return existing
    except (OSError, UnicodeDecodeError):
        pass

    generated = str(uuid.uuid4())
    try:
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        STATE_PATH.write_text(generated + "\n", encoding="utf-8")
    except OSError as exc:
        log.warning("could not persist the install id: %s", exc)
    return generated


# ---------------------------------------------------------------------------
# Collection
# ---------------------------------------------------------------------------
def _os_release(key: str) -> str:
    try:
        for line in Path("/etc/os-release").read_text(encoding="utf-8").splitlines():
            if line.startswith(key + "="):
                return line.split("=", 1)[1].strip().strip('"')
    except OSError:
        pass
    return ""


def _ram_bucket() -> str:
    """Rounded to a range. An exact byte count is closer to a fingerprint."""
    try:
        for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():
            if line.startswith("MemTotal:"):
                kb = int(line.split()[1])
                gb = kb / (1024 * 1024)
                for limit, label in (
                    (2, "<2GB"), (4, "2-4GB"), (8, "4-8GB"),
                    (16, "8-16GB"), (32, "16-32GB"), (64, "32-64GB"),
                ):
                    if gb < limit:
                        return label
                return ">64GB"
    except (OSError, ValueError, IndexError):
        pass
    return "unknown"


def _gpu_vendor() -> str:
    """Vendor only. The model is identifying in combination with other fields."""
    try:
        for path in Path("/sys/class/drm").glob("card*/device/vendor"):
            vendor = path.read_text(encoding="utf-8").strip().lower()
            return {
                "0x1002": "amd",
                "0x10de": "nvidia",
                "0x8086": "intel",
            }.get(vendor, "other")
    except OSError:
        pass
    return "unknown"


def _locale() -> str:
    """Language only, never the full locale with region and encoding."""
    raw = os.environ.get("LANG", "")
    match = re.match(r"([a-z]{2,3})(?:_([A-Z]{2}))?", raw)
    if not match:
        return "unknown"
    return match.group(1)


def collect(persist_id: bool = True) -> dict[str, str]:
    """Build the report. This is the complete set of fields, with no exceptions.

    Anything added here must be added to privacy.md and to the list shown in
    the welcome application, in the same commit.
    """
    return {
        "install_id": install_id(persist=persist_id),
        "edition": _os_release("VARIANT_ID") or "core",
        "version": _os_release("VERSION_ID") or "unknown",
        "locale": _locale(),
        "cpu_arch": os.uname().machine,
        "ram_bucket": _ram_bucket(),
        "gpu_vendor": _gpu_vendor(),
    }


# ---------------------------------------------------------------------------
# Sending
# ---------------------------------------------------------------------------
def send(payload: dict[str, str]) -> bool:
    # Imported here rather than at module scope so that a run without consent
    # does not even load the networking stack.
    import urllib.error
    import urllib.request

    data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        ENDPOINT,
        data=data,
        headers={
            "Content-Type": "application/json",
            # No User-Agent beyond the version already in the payload; a
            # detailed one would add entropy we just went to trouble to avoid.
            "User-Agent": "motel-telemetry",
        },
        method="POST",
    )

    try:
        with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
            return 200 <= response.status < 300
    except (urllib.error.URLError, OSError, TimeoutError) as exc:
        # A failed report is not worth telling the user about, and must never
        # be retried aggressively: this is the least important process on the
        # machine.
        log.info("report not sent: %s", exc)
        return False


def main(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(
        description="Send an opt-in usage report for Motel OS.")
    parser.add_argument("--dry-run", action="store_true",
                        help="Print what would be sent and exit.")
    parser.add_argument("--show", action="store_true",
                        help="Show the collected data without sending it.")
    parser.add_argument("--verbose", action="store_true")
    args = parser.parse_args(argv[1:])

    logging.basicConfig(
        level=logging.INFO if args.verbose else logging.WARNING,
        format="%(levelname)s: %(message)s",
    )

    # --show is explicitly allowed without consent: a user must be able to
    # inspect what this would send before deciding whether to enable it.
    if args.show:
        # persist_id=False: the id shown is a sample, not this machine's. A
        # user who has not consented ends the command with nothing new on disk.
        print(json.dumps(collect(persist_id=False), indent=2))
        return 0

    if not consent_given():
        log.info("telemetry is not enabled; nothing to do")
        return 0

    payload = collect()

    if args.dry_run:
        print(json.dumps(payload, indent=2))
        return 0

    # A failed report is never the caller's problem: this runs from a systemd
    # timer, and a non-zero exit would put a failed unit in front of a user
    # over the least important process on the machine.
    send(payload)
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))
