-
Notifications
You must be signed in to change notification settings - Fork 1
/
screenflux.py
108 lines (91 loc) · 3.3 KB
/
screenflux.py
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 sqlite3
from datetime import datetime
from influxdb_client import InfluxDBClient
from influxdb_client.client.write_api import SYNCHRONOUS
knowledge_db = os.path.expanduser("~/Library/Application Support/Knowledge/knowledgeC.db")
def query_database():
# Check if knowledgeC.db exists
if not os.path.exists(knowledge_db):
print("Could not find knowledgeC.db at %s." % (knowledge_db))
exit(1)
# Check if knowledgeC.db is readable
if not os.access(knowledge_db, os.R_OK):
print("The knowledgeC.db at %s is not readable.\nPlease grant full disk access to the application running the script (e.g. Terminal, iTerm, VSCode etc.)." % (knowledge_db))
exit(1)
# Connect to the SQLite database
with sqlite3.connect(knowledge_db) as con:
cur = con.cursor()
# Execute the SQL query to fetch data
# Modified from https://rud.is/b/2019/10/28/spelunking-macos-screentime-app-usage-with-r/
query = """
SELECT
ZOBJECT.ZVALUESTRING AS "app",
(ZOBJECT.ZENDDATE - ZOBJECT.ZSTARTDATE) AS "usage",
(ZOBJECT.ZSTARTDATE + 978307200) as "start_time",
(ZOBJECT.ZENDDATE + 978307200) as "end_time",
(ZOBJECT.ZCREATIONDATE + 978307200) as "created_at",
ZOBJECT.ZSECONDSFROMGMT AS "tz",
ZSOURCE.ZDEVICEID AS "device_id",
ZMODEL AS "device_model"
FROM
ZOBJECT
LEFT JOIN
ZSTRUCTUREDMETADATA
ON ZOBJECT.ZSTRUCTUREDMETADATA = ZSTRUCTUREDMETADATA.Z_PK
LEFT JOIN
ZSOURCE
ON ZOBJECT.ZSOURCE = ZSOURCE.Z_PK
LEFT JOIN
ZSYNCPEER
ON ZSOURCE.ZDEVICEID = ZSYNCPEER.ZDEVICEID
WHERE
ZSTREAMNAME = "/app/usage"
ORDER BY
ZSTARTDATE DESC
"""
cur.execute(query)
# Fetch all rows from the result set
return cur.fetchall()
def transform_data(rows):
data = []
for r in rows:
app = r[0]
usage = r[1]
time = r[3]
device_id = r[6]
device_model = r[7]
# Transform the data into the desired format for InfluxDB
data.append({
"measurement": "usage",
"tags": {
"app": app,
"device_id": device_id or "Unknown",
"device_model": device_model or "Unknown"
},
"fields": {
"usage": usage
},
"time": datetime.utcfromtimestamp(time)
})
return data
def write_to_influxdb(data):
# InfluxDB configuration
db_url = "..."
db_token = "..."
db_org = "..."
db_bucket = "screentime"
# Create InfluxDB client and write API
client = InfluxDBClient(url=db_url, token=db_token, org=db_org)
write_api = client.write_api(write_options=SYNCHRONOUS)
# Write the data to InfluxDB
write_api.write(bucket=db_bucket, org=db_org, record=data, time_precision='s')
def main():
# Query the database and fetch the rows
rows = query_database()
# Transform the data into InfluxDB format
data = transform_data(rows)
# Write the transformed data to InfluxDB
write_to_influxdb(data)
if __name__ == "__main__":
main()