feat(worktime): publish RDP activity to Grafana

2026-05-15 01:23:23 +03:00
parent 22de305c8a
commit 873e43fb2e
10 changed files with 726 additions and 18 deletions
@@ -99,6 +99,21 @@ def resolve_report_date(day=None, date_text=None):
    return now_local.date()


+def get_report_bounds(report_date):
+    start_local = datetime(report_date.year, report_date.month, report_date.day, tzinfo=REPORT_TZ)
+    end_local = start_local + timedelta(days=1) - timedelta(seconds=1)
+    start = start_local.astimezone(timezone.utc)
+    end = end_local.astimezone(timezone.utc)
+    end_exclusive = end + timedelta(seconds=1)
+    return {
+        "start_local": start_local,
+        "end_local": end_local,
+        "start": start,
+        "end": end,
+        "end_exclusive": end_exclusive,
+    }
+
+
 def _is_machine_user(user: str):
    u = (user or "").strip().lower()
    return u.endswith("$") or u in {"system", "localservice", "networkservice"}
@@ -169,7 +184,8 @@ def _merge_intervals(intervals):
    return merged


-def aggregate_rows(events, start, end, host):
+def _collect_user_rows(events, start, end, host):
+    end_exclusive = end + timedelta(seconds=1)
    by_user = {}
    by_identity = {}

@@ -212,10 +228,14 @@ def aggregate_rows(events, start, end, host):
            if active:
                row["active_samples"] += 1
                interval_start = sample["_ts"]
-                interval_end = min(sample["_ts"] + timedelta(seconds=sample_seconds), end + timedelta(seconds=1))
+                interval_end = min(sample["_ts"] + timedelta(seconds=sample_seconds), end_exclusive)
                if interval_end > interval_start:
                    row["intervals"].append((interval_start, interval_end))
+    return by_user

+
+def aggregate_rows(events, start, end, host):
+    by_user = _collect_user_rows(events, start, end, host)
    rows = []
    full_range = int((end - start).total_seconds()) + 1
    for username in sorted(by_user):
@@ -242,6 +262,62 @@ def aggregate_rows(events, start, end, host):
    return rows


+def aggregate_hourly_rows(events, start, end, host):
+    by_user = _collect_user_rows(events, start, end, host)
+    rows = []
+    for username in sorted(by_user):
+        row = by_user[username]
+        merged = _merge_intervals(row["intervals"])
+        per_bucket = {}
+        for interval_start, interval_end in merged:
+            cursor = interval_start
+            while cursor < interval_end:
+                bucket_local = cursor.astimezone(REPORT_TZ).replace(minute=0, second=0, microsecond=0)
+                bucket_start = bucket_local.astimezone(timezone.utc)
+                bucket_end = (bucket_local + timedelta(hours=1)).astimezone(timezone.utc)
+                overlap_start = max(interval_start, bucket_start)
+                overlap_end = min(interval_end, bucket_end)
+                if overlap_end > overlap_start:
+                    key = bucket_start
+                    per_bucket[key] = per_bucket.get(key, 0) + int((overlap_end - overlap_start).total_seconds())
+                cursor = bucket_end
+
+        for bucket_start in sorted(per_bucket):
+            active_seconds = per_bucket[bucket_start]
+            if active_seconds <= 0:
+                continue
+            bucket_local = bucket_start.astimezone(REPORT_TZ)
+            rows.append(
+                {
+                    "user": row["user"],
+                    "user_id": row["user_id"],
+                    "bucket_start_utc": to_iso_utc(bucket_start),
+                    "bucket_start_local": bucket_local.isoformat(),
+                    "report_date": bucket_local.date().isoformat(),
+                    "hour_local": bucket_local.strftime("%H:00"),
+                    "active_seconds": active_seconds,
+                    "active_hhmm": hhmm(active_seconds),
+                }
+            )
+    return rows
+
+
+def fetch_events_for_date(host, report_date):
+    bounds = get_report_bounds(report_date)
+    bucket_id = get_sessions_bucket_id(host)
+    try:
+        get(f"{AW}/buckets/{bucket_id}")
+    except Exception:
+        log_warning(f"bucket lookup failed for host={host} bucket={bucket_id} aw_base={AW}")
+        return bounds, []
+    try:
+        events = get(f"{AW}/buckets/{bucket_id}/events?limit=50000")
+    except Exception:
+        log_warning(f"events fetch failed for host={host} bucket={bucket_id} aw_base={AW}")
+        return bounds, []
+    return bounds, events
+
+
 def build_report_summary(rows):
    if not rows:
        return {
@@ -270,22 +346,8 @@ def build_report_summary(rows):


 def report_for_date(host, report_date):
-    start_local = datetime(report_date.year, report_date.month, report_date.day, tzinfo=REPORT_TZ)
-    end_local = start_local + timedelta(days=1) - timedelta(seconds=1)
-    start = start_local.astimezone(timezone.utc)
-    end = end_local.astimezone(timezone.utc)
-    bucket_id = get_sessions_bucket_id(host)
-    try:
-        get(f"{AW}/buckets/{bucket_id}")
-    except Exception:
-        log_warning(f"bucket lookup failed for host={host} bucket={bucket_id} aw_base={AW}")
-        return []
-    try:
-        events = get(f"{AW}/buckets/{bucket_id}/events?limit=50000")
-    except Exception:
-        log_warning(f"events fetch failed for host={host} bucket={bucket_id} aw_base={AW}")
-        return []
-    return aggregate_rows(events, start, end, host)
+    bounds, events = fetch_events_for_date(host, report_date)
+    return aggregate_rows(events, bounds["start"], bounds["end"], host)


 def report_today(host):
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+import importlib.util
+import os
+import sys
+import urllib.request
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+MODULE_PATH = Path(__file__).with_name("aw-worktime-api.py")
+SPEC = importlib.util.spec_from_file_location("aw_worktime_api", MODULE_PATH)
+WORKTIME = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(WORKTIME)
+
+INFLUX_URL = os.environ.get("AW_WORKTIME_INFLUX_URL", "").strip().rstrip("/")
+INFLUX_ORG = os.environ.get("AW_WORKTIME_INFLUX_ORG", "proxmox").strip() or "proxmox"
+INFLUX_BUCKET = os.environ.get("AW_WORKTIME_INFLUX_BUCKET", "aw_metrics").strip() or "aw_metrics"
+INFLUX_TOKEN = os.environ.get("AW_WORKTIME_INFLUX_TOKEN", "").strip()
+INFLUX_ENABLED = os.environ.get("AW_WORKTIME_INFLUX_ENABLED", "").strip().lower() in {"1", "true", "yes", "on"}
+HOSTS = [item.strip() for item in os.environ.get("AW_WORKTIME_INFLUX_HOSTS", WORKTIME.DEFAULT_HOST).split(",") if item.strip()]
+DAYS = [item.strip() for item in os.environ.get("AW_WORKTIME_INFLUX_DAYS", "today,yesterday").split(",") if item.strip()]
+
+
+def _escape_tag(value):
+    return (
+        str(value or "")
+        .replace("\\", "\\\\")
+        .replace(" ", "\\ ")
+        .replace(",", "\\,")
+        .replace("=", "\\=")
+    )
+
+
+def _line(measurement, tags, fields, timestamp_ns):
+    tag_part = ",".join(f"{key}={_escape_tag(value)}" for key, value in sorted(tags.items()) if value is not None and value != "")
+    field_parts = []
+    for key, value in fields.items():
+        if isinstance(value, bool):
+            field_parts.append(f"{key}={'true' if value else 'false'}")
+        elif isinstance(value, int):
+            field_parts.append(f"{key}={value}i")
+        elif isinstance(value, float):
+            field_parts.append(f"{key}={value}")
+        else:
+            text = str(value or "").replace("\\", "\\\\").replace('"', '\\"')
+            field_parts.append(f'{key}="{text}"')
+    if not field_parts:
+        return ""
+    if tag_part:
+        return f"{measurement},{tag_part} {','.join(field_parts)} {timestamp_ns}"
+    return f"{measurement} {','.join(field_parts)} {timestamp_ns}"
+
+
+def _timestamp_ns(dt):
+    return int(dt.astimezone(timezone.utc).timestamp() * 1_000_000_000)
+
+
+def build_lines_for_day(host, report_date):
+    bounds, events = WORKTIME.fetch_events_for_date(host, report_date)
+    rows = WORKTIME.aggregate_rows(events, bounds["start"], bounds["end"], host)
+    hourly_rows = WORKTIME.aggregate_hourly_rows(events, bounds["start"], bounds["end"], host)
+    summary = WORKTIME.build_report_summary(rows)
+
+    lines = []
+    daily_ts = _timestamp_ns(bounds["start"])
+
+    for row in rows:
+        lines.append(
+            _line(
+                "aw_rdp_worktime_daily",
+                {
+                    "host": host,
+                    "user": row["user"],
+                    "user_id": row["user_id"],
+                    "report_date": report_date.isoformat(),
+                },
+                {
+                    "active_seconds": int(row["active_seconds"]),
+                    "idle_seconds": int(row["idle_seconds"]),
+                    "sessions_count": int(row["sessions_count"]),
+                    "samples_count": int(row["samples_count"]),
+                    "active_samples": int(row["active_samples"]),
+                },
+                daily_ts,
+            )
+        )
+
+    for row in hourly_rows:
+        lines.append(
+            _line(
+                "aw_rdp_worktime_hourly",
+                {
+                    "host": host,
+                    "user": row["user"],
+                    "user_id": row["user_id"],
+                    "report_date": row["report_date"],
+                    "hour_local": row["hour_local"],
+                },
+                {
+                    "active_seconds": int(row["active_seconds"]),
+                },
+                _timestamp_ns(WORKTIME.pts(row["bucket_start_utc"])),
+            )
+        )
+
+    lines.append(
+        _line(
+            "aw_rdp_worktime_summary_daily",
+            {
+                "host": host,
+                "report_date": report_date.isoformat(),
+            },
+            {
+                "users_count": int(summary["users_count"]),
+                "total_active_seconds": int(summary["total_active_seconds"]),
+                "top_user": summary["top_user"],
+            },
+            daily_ts,
+        )
+    )
+    return [line for line in lines if line]
+
+
+def write_lines(lines):
+    if not lines:
+        return 0
+    if not INFLUX_URL or not INFLUX_TOKEN:
+        raise RuntimeError("InfluxDB destination is not configured")
+    payload = ("\n".join(lines) + "\n").encode("utf-8")
+    req = urllib.request.Request(
+        f"{INFLUX_URL}/api/v2/write?org={INFLUX_ORG}&bucket={INFLUX_BUCKET}&precision=ns",
+        data=payload,
+        method="POST",
+        headers={"Authorization": f"Token {INFLUX_TOKEN}", "Content-Type": "text/plain; charset=utf-8"},
+    )
+    with urllib.request.urlopen(req, timeout=30) as response:
+        if response.status not in {204, 200}:
+            raise RuntimeError(f"InfluxDB write failed with status={response.status}")
+    return len(lines)
+
+
+def main():
+    if not INFLUX_ENABLED:
+        print("[aw-worktime-influx-exporter] disabled by AW_WORKTIME_INFLUX_ENABLED", file=sys.stderr)
+        return 0
+
+    lines = []
+    for host in HOSTS:
+        for day in DAYS:
+            report_date = WORKTIME.resolve_report_date(day=day)
+            lines.extend(build_lines_for_day(host, report_date))
+    written = write_lines(lines)
+    print(f"[aw-worktime-influx-exporter] wrote {written} points to {INFLUX_BUCKET}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,15 @@
+[Unit]
+Description=AW Worktime InfluxDB exporter
+After=network-online.target aw-worktime-api.service
+Wants=network-online.target aw-worktime-api.service
+
+[Service]
+Type=oneshot
+EnvironmentFile=/etc/activitywatch/aw-server.env
+ExecStart=/usr/bin/python3 /usr/local/bin/aw-worktime-influx-exporter.py
+User=activitywatch
+Group=activitywatch
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=aw-worktime-influx-exporter
+
@@ -0,0 +1,11 @@
+[Unit]
+Description=Run AW Worktime InfluxDB exporter every 10 minutes
+
+[Timer]
+OnBootSec=3min
+OnUnitActiveSec=10min
+AccuracySec=1min
+Unit=aw-worktime-influx-exporter.service
+
+[Install]
+WantedBy=timers.target
@@ -71,3 +71,19 @@ def test_build_aw_api_base_accepts_root_and_api_urls():
    assert MODULE.build_aw_api_base("http://127.0.0.1:5600") == "http://127.0.0.1:5600/api/0"
    assert MODULE.build_aw_api_base("http://127.0.0.1:5600/") == "http://127.0.0.1:5600/api/0"
    assert MODULE.build_aw_api_base("http://127.0.0.1:5600/api/0") == "http://127.0.0.1:5600/api/0"
+
+
+def test_aggregate_hourly_rows_splits_interval_by_local_hour():
+    start = datetime(2026, 5, 14, 6, 0, 0, tzinfo=timezone.utc)
+    end = datetime(2026, 5, 14, 8, 59, 59, tzinfo=timezone.utc)
+    rows = MODULE.aggregate_hourly_rows(
+        [
+            _event("2026-05-14T06:50:00Z", "user5", 4, True, sampleSeconds=1800),
+            _event("2026-05-14T07:20:00Z", "user5", 4, True, sampleSeconds=1800),
+        ],
+        start,
+        end,
+        "SHARKON2025",
+    )
+    assert [row["hour_local"] for row in rows] == ["09:00", "10:00"]
+    assert [row["active_seconds"] for row in rows] == [600, 3000]
@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import importlib.util
+from datetime import date
+from pathlib import Path
+
+
+MODULE_PATH = Path(__file__).with_name("aw-worktime-influx-exporter.py")
+SPEC = importlib.util.spec_from_file_location("aw_worktime_influx_exporter", MODULE_PATH)
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+
+
+def test_build_lines_for_day_emits_daily_hourly_and_summary(monkeypatch):
+    bounds = MODULE.WORKTIME.get_report_bounds(date(2026, 5, 14))
+    events = [
+        {
+            "timestamp": "2026-05-14T06:00:00Z",
+            "duration": 0.0,
+            "data": {
+                "username": "user5",
+                "userId": "WORKGROUP\\user5",
+                "sessionId": 4,
+                "state": "Активно",
+                "active": True,
+                "sampleSeconds": 1800,
+            },
+        }
+    ]
+
+    monkeypatch.setattr(MODULE.WORKTIME, "fetch_events_for_date", lambda host, report_date: (bounds, events))
+    lines = MODULE.build_lines_for_day("SHARKON2025", date(2026, 5, 14))
+
+    assert any(line.startswith("aw_rdp_worktime_daily,") for line in lines)
+    assert any(line.startswith("aw_rdp_worktime_hourly,") for line in lines)
+    assert any(line.startswith("aw_rdp_worktime_summary_daily,") for line in lines)