#!/usr/bin/env python3
"""Validate Claude Code telemetry configuration against the Aliz baseline.

Resolves the effective value of each telemetry setting across the settings
precedence chain (server-managed > machine-managed > user), reports what is
actually in effect and where it came from, and optionally repairs the user
settings file.

Exit codes: 0 = compliant, 1 = misconfigured, 2 = script error.
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import sys
import time
from pathlib import Path

VERSION = "1.0.0"

OTLP_ENDPOINT = "https://otel-collector-505600134356.europe-central2.run.app"

# The keys every machine must have in effect. Values are exact-match.
REQUIRED_ENV = {
    "CLAUDE_CODE_ENABLE_TELEMETRY": "1",
    "OTEL_METRICS_EXPORTER": "otlp",
    "OTEL_LOGS_EXPORTER": "otlp",
    "OTEL_TRACES_EXPORTER": "otlp",
    "OTEL_METRICS_INCLUDE_REPOSITORY": "1",
    "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
    "OTEL_EXPORTER_OTLP_ENDPOINT": OTLP_ENDPOINT,
}

# Privacy flags the org ships via server-managed settings. Reported for
# visibility but never written by --fix: loosening them is an org decision,
# not something a local repair script should do on someone's behalf.
PRIVACY_ENV = {
    "OTEL_LOG_USER_PROMPTS": "0",
    "OTEL_LOG_ASSISTANT_RESPONSES": "0",
    "OTEL_LOG_TOOL_CONTENT": "0",
    "OTEL_LOG_RAW_API_BODIES": "0",
    "OTEL_LOG_TOOL_DETAILS": "1",
}

# Claude Code reads only whether these are set at all: any non-empty value,
# including the string "0", switches the suppression ON. Unset or "" is off.
SUPPRESSORS = (
    "DISABLE_TELEMETRY",
    "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC",
)

STATUS_OK = "OK"
STATUS_MISSING = "MISSING"
STATUS_WRONG = "WRONG"
STATUS_SUPPRESSED = "SUPPRESSED"


# --------------------------------------------------------------------------
# Terminal output
# --------------------------------------------------------------------------

class Style:
    """ANSI styling, silently disabled when the terminal cannot render it."""

    def __init__(self) -> None:
        self.enabled = self._supports_color()

    @staticmethod
    def _supports_color() -> bool:
        if os.environ.get("NO_COLOR"):
            return False
        if not sys.stdout.isatty():
            return False
        if platform.system() == "Windows":
            # Windows 10+ needs virtual terminal processing switched on explicitly.
            try:
                import ctypes

                kernel32 = ctypes.windll.kernel32
                kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7)
            except Exception:
                return False
        return True

    def _wrap(self, code: str, text: str) -> str:
        return f"\033[{code}m{text}\033[0m" if self.enabled else text

    def bold(self, t: str) -> str:
        return self._wrap("1", t)

    def dim(self, t: str) -> str:
        return self._wrap("2", t)

    def green(self, t: str) -> str:
        return self._wrap("32", t)

    def red(self, t: str) -> str:
        return self._wrap("31", t)

    def yellow(self, t: str) -> str:
        return self._wrap("33", t)

    def cyan(self, t: str) -> str:
        return self._wrap("36", t)


style = Style()


def status_label(status: str) -> str:
    if status == STATUS_OK:
        return style.green("PASS")
    if status == STATUS_SUPPRESSED:
        return style.red("SUPPRESSED")
    if status == STATUS_MISSING:
        return style.red("MISSING")
    return style.red("WRONG")


# --------------------------------------------------------------------------
# Settings sources
# --------------------------------------------------------------------------

def managed_settings_path() -> Path:
    system = platform.system()
    if system == "Darwin":
        return Path("/Library/Application Support/ClaudeCode/managed-settings.json")
    if system == "Windows":
        program_data = os.environ.get("PROGRAMDATA", r"C:\ProgramData")
        return Path(program_data) / "ClaudeCode" / "managed-settings.json"
    return Path("/etc/claude-code/managed-settings.json")


def user_settings_path() -> Path:
    return Path.home() / ".claude" / "settings.json"


def remote_settings_path() -> Path:
    return Path.home() / ".claude" / "remote-settings.json"


class Source:
    """One settings file in the precedence chain."""

    def __init__(self, label: str, path: Path, writable: bool) -> None:
        self.label = label
        self.path = path
        self.writable = writable
        self.exists = False
        self.error: str | None = None
        self.env: dict[str, str] = {}
        self._load()

    def _load(self) -> None:
        if not self.path.is_file():
            return
        self.exists = True
        try:
            data = json.loads(self.path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            self.error = f"invalid JSON: {exc}"
            return
        except OSError as exc:
            self.error = f"unreadable: {exc}"
            return
        env = data.get("env")
        if isinstance(env, dict):
            self.env = {k: str(v) for k, v in env.items()}


def build_sources() -> list[Source]:
    """Settings sources, highest precedence first."""
    return [
        Source("org (remote-settings.json)", remote_settings_path(), writable=False),
        Source("managed (managed-settings.json)", managed_settings_path(), writable=False),
        Source("user (settings.json)", user_settings_path(), writable=True),
    ]


def resolve(sources: list[Source], key: str) -> tuple[str | None, Source | None]:
    """Return the winning value for `key` and the source that set it."""
    for source in sources:
        if key in source.env:
            return source.env[key], source
    return None, None


def shell_override(key: str) -> str | None:
    """A value exported in the calling shell, which can shadow the settings file."""
    return os.environ.get(key)


# --------------------------------------------------------------------------
# Checks
# --------------------------------------------------------------------------

class Finding:
    def __init__(
        self,
        key: str,
        expected: str | None,
        actual: str | None,
        source: Source | None,
        status: str,
        note: str = "",
    ) -> None:
        self.key = key
        self.expected = expected
        self.actual = actual
        self.source = source
        self.status = status
        self.note = note

    @property
    def ok(self) -> bool:
        return self.status == STATUS_OK

    @property
    def source_label(self) -> str:
        return self.source.label.split(" ")[0] if self.source else "-"

    def as_dict(self) -> dict:
        return {
            "key": self.key,
            "expected": self.expected,
            "actual": self.actual,
            "source": self.source.label if self.source else None,
            "status": self.status,
            "note": self.note,
        }


def check_required(sources: list[Source]) -> list[Finding]:
    findings = []
    for key, expected in REQUIRED_ENV.items():
        actual, source = resolve(sources, key)
        if actual is None:
            findings.append(Finding(key, expected, None, None, STATUS_MISSING))
            continue

        note = ""
        exported = shell_override(key)
        if exported is not None and exported != actual:
            note = f"shell exports {exported!r}, which may shadow this"

        status = STATUS_OK if actual == expected else STATUS_WRONG
        findings.append(Finding(key, expected, actual, source, status, note))
    return findings


def check_privacy(sources: list[Source]) -> list[Finding]:
    findings = []
    for key, expected in PRIVACY_ENV.items():
        actual, source = resolve(sources, key)
        status = STATUS_OK if actual == expected else STATUS_MISSING if actual is None else STATUS_WRONG
        findings.append(Finding(key, expected, actual, source, status))
    return findings


def check_suppressors(sources: list[Source]) -> list[Finding]:
    """A suppressor counts as active whenever it is set to any non-empty value.

    Settings files are attributed before the process environment. Claude Code
    injects its own `env` block into the subprocesses it spawns, so a value
    visible in os.environ usually originates from a settings file rather than
    from the user's shell profile.
    """
    findings = []
    for key in SUPPRESSORS:
        actual, source = resolve(sources, key)

        if actual:
            note = 'set to "0", which still switches suppression ON' if actual == "0" else ""
            findings.append(Finding(key, "unset", actual, source, STATUS_SUPPRESSED, note))
            continue

        exported = shell_override(key)
        if exported:
            findings.append(
                Finding(key, "unset", exported, None, STATUS_SUPPRESSED,
                        "exported in your shell - settings files cannot override this")
            )
            continue

        findings.append(Finding(key, "unset", None, None, STATUS_OK))
    return findings


# --------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------

def print_table(title: str, findings: list[Finding]) -> None:
    print()
    print(style.bold(title))

    rows = []
    for f in findings:
        actual = f.actual if f.actual is not None else style.dim("(not set)")
        rows.append((f.key, actual, f.source_label, status_label(f.status), f))

    key_width = max((len(r[0]) for r in rows), default=3)
    # Measured on the raw value so ANSI codes do not skew the column width.
    val_width = max((len(f.actual) if f.actual is not None else 9 for f in findings), default=5)
    val_width = min(val_width, 58)
    src_width = max((len(r[2]) for r in rows), default=6)

    for key, actual, src, status, f in rows:
        raw_len = len(f.actual) if f.actual is not None else 9
        truncated = actual
        if f.actual is not None and raw_len > 58:
            truncated = f.actual[:55] + "..."
            raw_len = 58
        pad = " " * max(0, val_width - raw_len)
        print(f"  {key.ljust(key_width)}  {truncated}{pad}  {src.ljust(src_width)}  {status}")
        if f.status != STATUS_OK and f.expected is not None and f.status == STATUS_WRONG:
            print(f"  {' ' * key_width}  {style.dim('expected: ' + f.expected)}")
        if f.note:
            print(f"  {' ' * key_width}  {style.yellow('! ' + f.note)}")


def print_sources(sources: list[Source]) -> None:
    print()
    print(style.bold("Settings sources") + style.dim("  (highest precedence first)"))
    for source in sources:
        if source.error:
            state = style.red(source.error)
        elif not source.exists:
            state = style.dim("not present")
        else:
            state = style.green(f"{len(source.env)} env keys")
        print(f"  {source.label.ljust(34)}  {state}")
        print(f"  {style.dim('  ' + str(source.path))}")


# --------------------------------------------------------------------------
# Repair
# --------------------------------------------------------------------------

def plan_fix(sources: list[Source], findings: list[Finding], suppressors: list[Finding]) -> dict:
    """Work out what can be repaired by writing the user settings file alone."""
    user = next(s for s in sources if s.writable)
    higher = [s for s in sources if not s.writable]

    to_set: dict[str, str] = {}
    to_remove: list[str] = []
    blocked: list[str] = []

    for f in findings:
        if f.ok:
            continue
        # A managed tier that sets the key wrong cannot be beaten from user settings.
        if f.source is not None and not f.source.writable:
            blocked.append(f"{f.key}: set to {f.actual!r} by {f.source.label}")
            continue
        to_set[f.key] = f.expected

    for f in suppressors:
        if f.ok:
            continue
        if f.source is None:
            blocked.append(f"{f.key}: exported in your shell; unset it in your shell profile")
            continue
        if not f.source.writable:
            blocked.append(f"{f.key}: set by {f.source.label}")
            continue
        to_remove.append(f.key)

    return {"user": user, "set": to_set, "remove": to_remove, "blocked": blocked, "higher": higher}


def describe_fix(plan: dict) -> None:
    print()
    print(style.bold("Proposed changes to ") + style.cyan(str(plan["user"].path)))
    for key, value in plan["set"].items():
        shown = value if len(value) <= 58 else value[:55] + "..."
        print(f"  {style.green('+')} env.{key} = {shown}")
    for key in plan["remove"]:
        print(f"  {style.red('-')} env.{key}  (remove)")
    if plan["blocked"]:
        print()
        print(style.yellow("Cannot be fixed by this script:"))
        for item in plan["blocked"]:
            print(f"  ! {item}")


def apply_fix(plan: dict) -> Path | None:
    """Merge the planned changes into the user settings file, preserving everything else."""
    path: Path = plan["user"].path
    path.parent.mkdir(parents=True, exist_ok=True)

    if path.is_file():
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            print(style.red(f"Refusing to edit {path}: it is not valid JSON ({exc})."))
            print("Fix the syntax by hand, then run this check again.")
            return None
        backup = path.with_name(f"{path.name}.bak-{time.strftime('%Y%m%d-%H%M%S')}")
        backup.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    else:
        data = {}
        backup = None

    env = data.get("env")
    if not isinstance(env, dict):
        env = {}
    env.update(plan["set"])
    for key in plan["remove"]:
        env.pop(key, None)
    data["env"] = env

    path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    return backup


def confirm(prompt: str) -> bool:
    try:
        answer = input(prompt).strip().lower()
    except (EOFError, KeyboardInterrupt):
        print()
        return False
    return answer in ("y", "yes")


# --------------------------------------------------------------------------
# Entry point
# --------------------------------------------------------------------------

def main() -> int:
    parser = argparse.ArgumentParser(
        prog="check.py",
        description="Validate Claude Code telemetry configuration against the Aliz baseline.",
    )
    parser.add_argument("--fix", action="store_true", help="repair without prompting")
    parser.add_argument("--no-fix", action="store_true", help="never offer to repair (for CI)")
    parser.add_argument("--json", action="store_true", help="machine-readable output")
    parser.add_argument("--version", action="version", version=f"aliz-claude-check {VERSION}")
    args = parser.parse_args()

    sources = build_sources()
    required = check_required(sources)
    privacy = check_privacy(sources)
    suppressors = check_suppressors(sources)

    failures = [f for f in required + suppressors if not f.ok]

    if args.json:
        print(json.dumps({
            "version": VERSION,
            "compliant": not failures,
            "sources": [
                {"label": s.label, "path": str(s.path), "exists": s.exists, "error": s.error}
                for s in sources
            ],
            "required": [f.as_dict() for f in required],
            "privacy": [f.as_dict() for f in privacy],
            "suppressors": [f.as_dict() for f in suppressors],
        }, indent=2))
        return 1 if failures else 0

    print()
    print(style.bold(f"Aliz - Claude Code telemetry check  {style.dim('v' + VERSION)}"))
    print(style.dim(f"{platform.system()} {platform.release()}  -  python {platform.python_version()}"))

    print_sources(sources)
    print_table("Required telemetry settings", required)
    print_table("Privacy flags (reported, never auto-changed)", privacy)
    print_table("Suppressors (must be unset)", suppressors)

    print()
    if not failures:
        print(style.green("PASS") + "  Telemetry is configured correctly on this machine.")
        return 0

    print(style.red("FAIL") + f"  {len(failures)} problem(s) found.")

    if args.no_fix:
        return 1

    plan = plan_fix(sources, required, suppressors)
    if not plan["set"] and not plan["remove"]:
        describe_fix(plan)
        print()
        print(style.yellow("Nothing this script can repair automatically."))
        return 1

    describe_fix(plan)
    print()

    if not args.fix:
        if not sys.stdin.isatty():
            print(style.yellow("Non-interactive shell; re-run with --fix to apply."))
            return 1
        if not confirm("Apply these changes? [y/N] "):
            print("No changes made.")
            return 1

    backup = apply_fix(plan)
    if backup is None and plan["user"].path.is_file() is False:
        return 2

    print()
    print(style.green("Applied.") + f"  Updated {plan['user'].path}")
    if backup:
        print(style.dim(f"Backup written to {backup}"))
    print("Restart any running Claude Code sessions for the change to take effect.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print()
        sys.exit(130)