1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
import os
import socket
from typing import Any, Dict, Optional
from dotenv import load_dotenv
from ..constants.main import CONTINUE_SERVER_VERSION_FILE
from .commonregex import clean_pii_from_any
from .paths import getServerFolderPath
load_dotenv()
in_codespaces = os.getenv("CODESPACES") == "true"
POSTHOG_API_KEY = "phc_JS6XFROuNbhJtVCEdTSYk6gl5ArRrTNMpCcguAXlSPs"
def is_connected():
try:
# connect to the host -- tells us if the host is actually reachable
socket.create_connection(("www.google.com", 80))
return True
except OSError:
pass
return False
class PostHogLogger:
unique_id: str = "NO_UNIQUE_ID"
allow_anonymous_telemetry: bool = False
ide_info: Optional[Dict] = None
posthog = None
def __init__(self, api_key: str):
self.api_key = api_key
def setup(
self, unique_id: str, allow_anonymous_telemetry: bool, ide_info: Optional[Dict]
):
self.unique_id = unique_id or "NO_UNIQUE_ID"
self.allow_anonymous_telemetry = allow_anonymous_telemetry or False
self.ide_info = ide_info
# Capture initial event
self.capture_event("session_start", {"os": os.name})
def capture_event(self, event_name: str, event_properties: Any):
"""Safely capture event. Telemetry should never be the reason Continue doesn't work"""
try:
self._capture_event(event_name, event_properties)
except Exception as e:
print(f"Failed to capture event: {e}")
pass
_found_disconnected: bool = False
def _capture_event(self, event_name: str, event_properties: Any):
# logger.debug(
# f"Logging to PostHog: {event_name} ({self.unique_id}, {self.allow_anonymous_telemetry}): {event_properties}")
telemetry_path = os.path.expanduser("~/.continue/telemetry.log")
# Make sure the telemetry file exists
if not os.path.exists(telemetry_path):
os.makedirs(os.path.dirname(telemetry_path), exist_ok=True)
open(telemetry_path, "w").close()
with open(telemetry_path, "a") as f:
str_to_write = f"{event_name}: {event_properties}\n{self.unique_id}\n{self.allow_anonymous_telemetry}\n\n"
f.write(str_to_write)
if not self.allow_anonymous_telemetry:
return
# Clean PII from event properties
event_properties = clean_pii_from_any(event_properties)
# Add additional properties that are on every event
if in_codespaces:
event_properties["codespaces"] = True
server_version_file = os.path.join(
getServerFolderPath(), CONTINUE_SERVER_VERSION_FILE
)
if os.path.exists(server_version_file):
with open(server_version_file, "r") as f:
event_properties["server_version"] = f.read()
# Add operating system
event_properties["os"] = os.name
if self.ide_info:
event_properties["ide_name"] = self.ide_info.get("name", None)
event_properties["ide_version"] = self.ide_info.get("version", None)
event_properties["ide_remote_name"] = self.ide_info.get("remoteName", None)
# Send event to PostHog
if self.posthog is None:
from posthog import Posthog
# The personal API key is necessary only if you want to use local evaluation of feature flags.
self.posthog = Posthog(self.api_key, host="https://app.posthog.com")
if is_connected():
self.posthog.capture(self.unique_id, event_name, event_properties)
else:
if not self._found_disconnected:
self._found_disconnected = True
raise ConnectionError("No internet connection")
posthog_logger = PostHogLogger(api_key=POSTHOG_API_KEY)
|
