# Copyright (C) 2018-2024 C-PAC Developers
# This file is part of C-PAC.
# C-PAC is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
# C-PAC is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
# You should have received a copy of the GNU Lesser General Public
# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
import configparser
import os
import os.path as op
import tempfile
import threading
import traceback
import uuid
import requests
from CPAC.info import __version__, ga_tracker
temp_dir = False
udir = op.expanduser("~")
if udir == "/":
udir = tempfile.mkdtemp()
temp_dir = True
tracking_path = op.join(udir, ".cpac")
[docs]
def get_or_create_config():
if not op.exists(tracking_path):
parser = configparser.ConfigParser()
parser.read_dict({"user": {"uid": uuid.uuid1().hex, "track": True}})
with open(tracking_path, "w+") as fhandle:
parser.write(fhandle)
else:
parser = configparser.ConfigParser()
parser.read(tracking_path)
return parser
[docs]
def get_uid():
if os.environ.get("CPAC_TRACKING", "").lower() not in ["", "0", "false", "off"]:
return os.environ.get("CPAC_TRACKING")
parser = get_or_create_config()
if parser["user"].getboolean("track"):
return parser["user"]["uid"]
return None
[docs]
def do_it(data, timeout):
try:
headers = {"User-Agent": f"C-PAC/{__version__} (https://fcp-indi.github.io)"}
_done = requests.post(
"https://www.google-analytics.com/collect",
data=data,
timeout=timeout,
headers=headers,
)
except:
_done = False
finally:
if temp_dir:
try:
os.remove(tracking_path)
os.rmdir(udir)
except (TypeError, OSError) as e:
msg = f"Unable to delete temporary tracking path {tracking_path}."
raise OSError(msg) from e
return _done
[docs]
def track_event(
category,
action,
uid=None,
label=None,
value=0,
software_version=None,
timeout=2,
thread=True,
):
"""
Record an event with Google Analytics.
Parameters
----------
tracking_id : str
Google Analytics tracking ID.
category : str
Event category.
action : str
Event action.
uid : str
User unique ID, assigned when popylar was installed.
label : str
Event label.
value : int
Event value.
software_version : str
Records a version of the software.
timeout : float
Maximal duration (in seconds) for the network connection to track the
event. After this duration has elapsed with no response (e.g., on a
slow network connection), the tracking is dropped.
"""
if os.environ.get("CPAC_TRACKING", "").lower() in ["0", "false", "off"]:
return
if uid is None:
uid = get_uid()
if not uid:
return
this = "/CPAC/utils/ga.py"
exec_stack = list(reversed(traceback.extract_stack()))
assert exec_stack[0][0].endswith(this)
package_path = exec_stack[0][0][: -len(this)]
# only CPAC paths are going to be recorded
file_path = ""
for s in exec_stack:
if s[0].endswith(this):
continue
if not s[0].startswith(package_path):
break
file_path = s[0][len(package_path) :]
data = {
"v": "1", # API version.
"tid": ga_tracker, # GA tracking ID
"dp": file_path,
"cid": uid, # User unique ID, stored in `tracking_path`
"t": "event", # Event hit type.
"ec": category, # Event category.
"ea": action, # Event action.
"el": label, # Event label.
"ev": value, # Event value, must be an integer
"aid": "CPAC",
"an": "CPAC",
"av": __version__,
"aip": 1, # anonymize IP by removing last octet, slightly worse
# geolocation
}
if thread:
t = threading.Thread(target=do_it, args=(data, timeout))
t.start()
else:
do_it(data, timeout)
[docs]
def track_config(cpac_interface):
track_event("config", cpac_interface, label=None, value=None, thread=False)
[docs]
def track_run(level="participant", participants=0):
if level in ["participant", "group"]:
track_event(
"run", level, label="participants", value=participants, thread=False
)
else:
track_event(
"config", "test", label="participants", value=participants, thread=False
)