Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "kup"
version = "0.2.5"
version = "0.2.6"
description = "kup is a tool for managing installations of the K framework along with the different available semantics"
readme = "README.md"
requires-python = ">=3.10,<4"
Expand All @@ -15,6 +15,8 @@ dependencies = [
"rich>=12.6.0,<13",
"terminaltables>=3.1.10,<4",
"tinynetrc>=1.3.1,<2",
"tomli>=2.4.0",
"tomli-w>=1.2.0",
]

[[project.authors]]
Expand Down
28 changes: 24 additions & 4 deletions src/kup/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
PackageName,
PackageVersion,
)
from .telemetry import emit_event

if TYPE_CHECKING:
from argparse import Namespace
Expand Down Expand Up @@ -199,7 +200,7 @@ def package_metadata_tree(p: PackageMetadata | Follows, lbl: str | None = None,
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()]
Expand Down Expand Up @@ -261,8 +262,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))
Expand Down Expand Up @@ -360,7 +361,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()
Expand Down Expand Up @@ -487,6 +488,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(
Expand Down Expand Up @@ -531,6 +541,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}[/]."
)
Expand Down
66 changes: 66 additions & 0 deletions src/kup/telemetry.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
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 requests.exceptions.ReadTimeout:
_LOGGER.debug(f'Telemetry event timed out: {event}')
except Exception as e:
_LOGGER.warning(f'Telemetry event failed: {event}', exc_info=e)
Loading