diff --git a/README.md b/README.md index 2ca1223..1ccf4d6 100644 --- a/README.md +++ b/README.md @@ -72,3 +72,4 @@ The [`examples/`](examples/) directory contains a couple of example scripts, inc - [`load_dataframe.py`](https://github.com/ActivityWatch/aw-client/blob/master/examples/load_dataframe.py) - loads events from a host using a categorizing & AFK-filtering query, put result in a pandas dataframe, and export as CSV. - [`merge_buckets.py`](examples/merge_buckets.py) - merges two buckets with non-intersecting events by moving all events from one into the other. - [`redact_sensitive.py`](examples/redact_sensitive.py) - redact sensitive events. + - [`export_for_workflow_mining.py`](examples/export_for_workflow_mining.py) - exports canonical, categorized events to a flat JSON file (timestamp/duration/data per event) for use with external process-mining or workflow-discovery tools. diff --git a/examples/export_for_workflow_mining.py b/examples/export_for_workflow_mining.py new file mode 100644 index 0000000..e175f24 --- /dev/null +++ b/examples/export_for_workflow_mining.py @@ -0,0 +1,137 @@ +""" +Export canonical window events to a flat JSON file shaped for external +process-mining / workflow-discovery tools: a list of +{"timestamp": , "duration": , "data": {...}} records, one +per event, sorted by time. + +This is deliberately *not* a bucket dump -- it reuses the same categorizing ++ AFK-filtering canonical query as load_dataframe.py, so the exported events +are already the "what was actually happening" view rather than every raw +watcher heartbeat, which is what a downstream sequence-clustering or +process-mining tool actually wants as input. +""" + +import argparse +import json +import os +import socket +from datetime import datetime, timedelta, timezone +from typing import Any + +import iso8601 +from aw_client import ActivityWatchClient +from aw_client.classes import default_classes +from aw_client.queries import DesktopQueryParams, canonicalEvents + + +def build_query(hostname: str) -> str: + canonicalQuery = canonicalEvents( + DesktopQueryParams( + bid_window=f"aw-watcher-window_{hostname}", + bid_afk=f"aw-watcher-afk_{hostname}", + classes=default_classes, + ) + ) + return f""" + {canonicalQuery} + RETURN = {{"events": events}}; + """ + + +def serialize_events(raw_events: list) -> list: + """Keep only the fields downstream tools need, sorted by *parsed* time. + Sorting on the raw timestamp string is not equivalent: two valid + ISO8601 timestamps using different UTC offsets (e.g. "-05:00" vs "Z") + can sort backwards as strings even though one is unambiguously earlier + than the other once parsed.""" + events = [ + {"timestamp": e["timestamp"], "duration": e["duration"], "data": e["data"]} + for e in raw_events + ] + events.sort(key=lambda e: iso8601.parse_date(e["timestamp"])) + return events + + +def write_json_atomic(path: str, data: Any) -> None: + """Write to a temp file in the same directory, then atomically replace + the destination. Without this, a failure partway through serialization + (a very large export, disk full mid-write) would leave a truncated file + in place of whatever good export was already there.""" + tmp_path = f"{path}.tmp" + with open(tmp_path, "w") as f: + json.dump(data, f, indent=2) + os.replace(tmp_path, path) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--days", type=int, default=7, help="How many days back to export (default: 7)" + ) + parser.add_argument( + "--hostname", + default="fakedata" if os.getenv("CI") else socket.gethostname(), + help="Hostname whose window/AFK buckets to query (default: this machine's)", + ) + parser.add_argument( + "--out", default="activitywatch_export.json", help="Output file path" + ) + args = parser.parse_args() + + now = datetime.now(tz=timezone.utc) + since = now - timedelta(days=args.days) + + aw = ActivityWatchClient(client_name="export_for_workflow_mining") + print(f"Querying the last {args.days} day(s) of events for host '{args.hostname}'...") + query = build_query(args.hostname) + data = aw.query(query, [(since, now)]) + + events = serialize_events(data[0]["events"]) + write_json_atomic(args.out, events) + + print(f"Wrote {len(events)} events to {args.out}") + + +def test_build_query_embeds_hostname(): + query = build_query("myhost") + assert "aw-watcher-window_myhost" in query + assert "aw-watcher-afk_myhost" in query + + +def test_serialize_events_sorts_by_parsed_time_not_string(): + # Both timestamps are valid ISO8601, but a plain string sort gets this + # backwards: the first is 2024-01-02 04:30 UTC, the second is + # 2024-01-02 01:00 UTC (earlier) -- yet "...01-01T23..." < "...01-02T01..." + # as strings, because the offsets differ. + raw = [ + {"timestamp": "2024-01-01T23:30:00-05:00", "duration": 60, "data": {"app": "later"}}, + {"timestamp": "2024-01-02T01:00:00Z", "duration": 60, "data": {"app": "earlier"}}, + ] + result = serialize_events(raw) + assert [e["data"]["app"] for e in result] == ["earlier", "later"] + + +def test_serialize_events_keeps_only_expected_fields(): + raw = [{"timestamp": "2024-01-01T09:00:00Z", "duration": 60, "data": {"app": "a"}, "id": 123}] + result = serialize_events(raw) + assert set(result[0].keys()) == {"timestamp", "duration", "data"} + + +def test_write_json_atomic_preserves_prior_file_on_failed_write(tmp_path): + path = str(tmp_path / "export.json") + write_json_atomic(path, [{"ok": True}]) + + class Unserializable: + pass + + try: + write_json_atomic(path, {"bad": Unserializable()}) + except TypeError: + pass + + with open(path) as f: + assert json.load(f) == [{"ok": True}] + + +if __name__ == "__main__": + main()