Add telemetry support

pyproject.toml

@@ -22,6 +22,8 @@ rich = "^12.6.0"
 pyxdg = "^0.28"
 tinynetrc = "^1.3.1"
 git-url-parse = "^1.2.2"
+tomli = "^2.4.0"
+tomli-w = "^1.2.0"
 
 [tool.poetry.group.dev.dependencies]
 autoflake = "*"

src/kup/__main__.py

@@ -50,6 +50,7 @@
     PackageName,
     PackageVersion,
 )
+from .telemetry import _emit_event
 
 console = Console(theme=Theme({'markdown.code': 'green'}))
 
@@ -192,7 +193,7 @@ def package_metadata_tree(
         follows = (' - follows [green]' + '/'.join(p.follows)) if type(p) == Follows else ''
         status = ''
         if show_status and type(p) == PackageMetadata:
-            auth = {'Authorization': f'Bearer {os.getenv("GH_TOKEN")}'} if os.getenv('GH_TOKEN') else {}
+            auth = {'Authorization': f"Bearer {os.getenv('GH_TOKEN')}"} if os.getenv('GH_TOKEN') else {}
             commits = requests.get(f'https://api.github.com/repos/{p.org}/{p.repo}/commits', headers=auth)
             if commits.ok:
                 commits_list = [c['sha'] for c in commits.json()]
@@ -254,8 +255,8 @@ def reload_packages(load_versions: bool = True) -> None:
     if pinned.ok:
         pinned_package_cache = {r['name']: r['lastRevision']['storePath'] for r in pinned.json()}
 
-    if os.path.exists(f'{os.getenv("HOME")}/.nix-profile/manifest.json'):
-        manifest_file = open(f'{os.getenv("HOME")}/.nix-profile/manifest.json')
+    if os.path.exists(f"{os.getenv('HOME')}/.nix-profile/manifest.json"):
+        manifest_file = open(f"{os.getenv('HOME')}/.nix-profile/manifest.json")
         manifest = json.loads(manifest_file.read())['elements']
         if type(manifest) is list:
             manifest = dict(enumerate(manifest))
@@ -334,7 +335,7 @@ def list_package(
             auth = (
                 {'Authorization': f'Bearer {listed_package.access_token}'}
                 if listed_package.access_token
-                else {'Authorization': f'Bearer {os.getenv("GH_TOKEN")}'}
+                else {'Authorization': f"Bearer {os.getenv('GH_TOKEN')}"}
                 if os.getenv('GH_TOKEN')
                 else {}
             )
@@ -355,7 +356,7 @@ def list_package(
                     c['commit']['message'],
                     tagged_releases[c['sha']]['name'] if c['sha'] in tagged_releases else None,
                     c['commit']['committer']['date'],
-                    f'github:{listed_package.org}/{listed_package.repo}/{c["sha"]}#{listed_package.package_name}'
+                    f"github:{listed_package.org}/{listed_package.repo}/{c['sha']}#{listed_package.package_name}"
                     in pinned_package_cache.keys(),
                 )
                 for c in commits.json()
@@ -382,7 +383,7 @@ def list_package(
         table_data = [['Package name (alias)', 'Installed version', 'Status'],] + [
             [
                 str(PackageName(alias, p.package_name.ext).pretty_name),
-                f'{p.commit[:7] if TERMINAL_WIDTH < 80 else p.commit}{" (" + p.tag + ")" if p.tag else ""}'
+                f"{p.commit[:7] if TERMINAL_WIDTH < 80 else p.commit}{' (' + p.tag + ')' if p.tag else ''}"
                 if type(p) == ConcretePackage
                 else '\033[3mlocal checkout\033[0m'
                 if type(p) == LocalPackage
@@ -482,6 +483,15 @@ def install_package(
     _, git_token_options = package.concrete_repo_path_with_access
     overrides = mk_override_args(package, package_overrides)
 
+    _emit_event(
+        'kup_install_start',
+        {
+            'package': package_name.base,
+            'version': package_version,
+            'has_overrides': len(package_overrides) > 0 if package_overrides else False,
+        },
+    )
+
     if not overrides and package.uri in pinned_package_cache:
         rich.print(f" ⌛ Fetching cached version of '[green]{package_name.pretty_name}[/]' ...")
         nix(
@@ -526,6 +536,16 @@ def install_package(
         display_version = None
     display_version = f' ({display_version})' if display_version is not None else ''
 
+    _emit_event(
+        'kup_install_complete',
+        {
+            'package': package_name.base,
+            'version': package_version or 'latest',
+            'was_update': verb == 'updated',
+            'from_cache': package.uri in pinned_package_cache and not overrides,
+        },
+    )
+
     rich.print(
         f" ✅ Successfully {verb} '[green]{package_name.base}[/]' version [blue]{package.uri}{display_version}[/]."
     )
@@ -595,7 +615,7 @@ def check_github_api_accessible(org: str, repo: str, access_token: Optional[str]
     auth = (
         {'Authorization': f'Bearer {access_token}'}
         if access_token
-        else {'Authorization': f'Bearer {os.getenv("GH_TOKEN")}'}
+        else {'Authorization': f"Bearer {os.getenv('GH_TOKEN')}"}
         if os.getenv('GH_TOKEN')
         else {}
     )

src/kup/telemetry.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import logging
+import os
+import uuid
+from pathlib import Path
+from typing import Final
+
+import requests
+import tomli
+import tomli_w
+
+_LOGGER: Final = logging.getLogger(__name__)
+
+KPROFILE_CONFIG_DIR: Final = Path.home() / '.config' / 'kprofile'
+KPROFILE_CONFIG_FILE: Final = KPROFILE_CONFIG_DIR / 'config.toml'
+TELEMETRY_MESSAGE: Final = f'Telemetry: sending anonymous usage data. You can opt out by setting KPROFILE_TELEMETRY_DISABLED=true or consent=false in {KPROFILE_CONFIG_FILE}'
+
+
+def _get_user_id() -> str:
+    """Get or create persistent anonymous user ID"""
+    if not KPROFILE_CONFIG_FILE.exists():
+        KPROFILE_CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
+        config = {'user': {'user_id': str(uuid.uuid4()), 'consent': True}}
+        with open(KPROFILE_CONFIG_FILE, 'wb') as f:
+            tomli_w.dump(config, f)
+        return str(config['user']['user_id'])
+
+    with open(KPROFILE_CONFIG_FILE, 'rb') as f:
+        config = tomli.load(f)
+
+    return str(config['user']['user_id'])
+
+
+def _has_permission() -> bool:
+    """Check if telemetry is enabled"""
+    if os.getenv('KPROFILE_TELEMETRY_DISABLED', '').lower() == 'true':
+        return False
+
+    _get_user_id()
+
+    with open(KPROFILE_CONFIG_FILE, 'rb') as f:
+        config = tomli.load(f)
+
+    return config.get('user', {}).get('consent', True)
+
+
+def _emit_event(event: str, properties: dict | None = None) -> None:
+    """Send telemetry event to proxy server"""
+    if not _has_permission():
+        return
+
+    _LOGGER.info(TELEMETRY_MESSAGE)
+
+    try:
+        requests.post(
+            'https://ojlk1fzi13.execute-api.us-east-1.amazonaws.com/dev/track',
+            json={'user_id': _get_user_id(), 'event': event, 'properties': properties},
+            timeout=2,
+        )
+    except Exception:
+        pass  # Fail silently