From 6c41717ef93b7d3e3473b89e93062f18a7db93fd Mon Sep 17 00:00:00 2001 From: igor04091968 Date: Fri, 15 May 2026 00:37:15 +0300 Subject: [PATCH] fix(worktime): restore real per-user RDP report pipeline --- ansible/deploy_aw_server.yml | 2 +- ansible/tasks/provision_ct_and_deploy_aw.yml | 1 + aw-server/aw-server.env.example | 1 + aw-server/aw-worktime-api.py | 322 +++++++++++++++---- aw-server/aw-worktime-panel.js | 20 +- aw-server/test_aw_worktime_api.py | 73 +++++ scripts/rdp-worktime-report.sh | 146 +++++++-- windows/worktime-session-collector.ps1 | 43 ++- 8 files changed, 504 insertions(+), 104 deletions(-) create mode 100644 aw-server/test_aw_worktime_api.py diff --git a/ansible/deploy_aw_server.yml b/ansible/deploy_aw_server.yml index 2d1dfc3..7f9341d 100644 --- a/ansible/deploy_aw_server.yml +++ b/ansible/deploy_aw_server.yml @@ -310,7 +310,7 @@ AW_DLP_CASE_DB_PATH={{ aw_dlp_case_db_path }} AW_DLP_COMPLIANCE_REPORT_DIR={{ aw_dlp_compliance_report_dir }} AW_DLP_COMPLIANCE_TEMPLATE={{ aw_dlp_compliance_template_path }} - AW_SERVER_URL=http://127.0.0.1:5600/api/0 + AW_SERVER_URL=http://127.0.0.1:5600 XDG_DATA_HOME={{ aw_server_data_dir }}/.local/share XDG_CONFIG_HOME={{ aw_server_data_dir }}/.config diff --git a/ansible/tasks/provision_ct_and_deploy_aw.yml b/ansible/tasks/provision_ct_and_deploy_aw.yml index 0551e2f..a2610fd 100644 --- a/ansible/tasks/provision_ct_and_deploy_aw.yml +++ b/ansible/tasks/provision_ct_and_deploy_aw.yml @@ -159,6 +159,7 @@ AW_SERVER_LOG_DIR={{ aw_server_log_dir }} AW_SERVER_USER={{ aw_server_user }} AW_SERVER_GROUP={{ aw_server_group }} + AW_SERVER_URL=http://127.0.0.1:{{ aw_server_port }} AW_WORKTIME_REPORT_BASE={{ aw_worktime_report_base }} AW_WORKTIME_TZ={{ aw_worktime_timezone }} no_log: true diff --git a/aw-server/aw-server.env.example b/aw-server/aw-server.env.example index 5c02cd4..5b113e8 100755 --- a/aw-server/aw-server.env.example +++ b/aw-server/aw-server.env.example @@ -15,6 +15,7 @@ AW_SERVER_GROUP=activitywatch # Worktime API Configuration AW_WORKTIME_REPORT_BASE=http://10.10.10.13:5610 AW_WORKTIME_TZ=Europe/Moscow +AW_SERVER_URL=http://127.0.0.1:5600 # DLP IOC Configuration AW_DLP_IOC_DIR=/opt/activitywatch/dlp-ioc/output diff --git a/aw-server/aw-worktime-api.py b/aw-server/aw-worktime-api.py index f47e4a9..c1c1b63 100644 --- a/aw-server/aw-worktime-api.py +++ b/aw-server/aw-worktime-api.py @@ -1,16 +1,35 @@ #!/usr/bin/env python3 -from http.server import BaseHTTPRequestHandler, HTTPServer import csv import io +import importlib.util import json import os +import sys import urllib.request from datetime import datetime, timezone, timedelta +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from urllib.parse import parse_qs, urlparse from zoneinfo import ZoneInfo -AW = "http://127.0.0.1:5600/api/0" + +def build_aw_api_base(raw_url): + url = (raw_url or "http://127.0.0.1:5600").strip().rstrip("/") + if url.endswith("/api/0"): + return url + return url + "/api/0" + + +AW_SERVER_URL = os.environ.get("AW_SERVER_URL", "http://127.0.0.1:5600") +AW = build_aw_api_base(AW_SERVER_URL) REPORT_TZ = ZoneInfo(os.environ.get("AW_WORKTIME_TZ", "Europe/Moscow")) IOC_DIR = os.environ.get("AW_DLP_IOC_DIR", "/opt/activitywatch/dlp-ioc/output") +DEFAULT_HOST = os.environ.get("AW_WORKTIME_HOST", "SHARKON2025").strip() or "SHARKON2025" +DEFAULT_SAMPLE_SECONDS = max(1.0, float(os.environ.get("AW_WORKTIME_DEFAULT_SAMPLE_SECONDS", "30"))) +MAX_SAMPLE_SECONDS = max(DEFAULT_SAMPLE_SECONDS, float(os.environ.get("AW_WORKTIME_MAX_SAMPLE_SECONDS", "300"))) +LISTEN_HOST = os.environ.get("AW_WORKTIME_LISTEN_HOST", "0.0.0.0") +LISTEN_PORT = int(os.environ.get("AW_WORKTIME_PORT", "5610")) +MODULE_PATH = Path(__file__).resolve() def get(u): @@ -18,24 +37,64 @@ def get(u): return json.loads(r.read().decode()) +def log_warning(message): + print(f"[aw-worktime-api] {message}", file=sys.stderr, flush=True) + + def pts(s): return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc) -def _is_machine_user(user: str) -> bool: +def to_iso_utc(dt): + return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def hhmm(total_seconds): + total_seconds = max(0, int(total_seconds)) + return "%02d:%02d" % (total_seconds // 3600, (total_seconds % 3600) // 60) + + +def clamp_seconds(value, fallback=DEFAULT_SAMPLE_SECONDS): + try: + seconds = float(value) + except Exception: + seconds = float(fallback) + if seconds <= 0: + seconds = float(fallback) + return min(seconds, MAX_SAMPLE_SECONDS) + + +def resolve_host(request_host=None): + host = (request_host or DEFAULT_HOST).strip() + if not host: + host = DEFAULT_HOST + return host + + +def get_sessions_bucket_id(host): + return f"aw-worktime-sessions_{resolve_host(host)}" + + +def resolve_report_date(day=None, date_text=None): + now_local = datetime.now(REPORT_TZ) + if date_text: + return datetime.strptime(date_text, "%Y-%m-%d").date() + if day == "yesterday": + return (now_local - timedelta(days=1)).date() + return now_local.date() + + +def _is_machine_user(user: str): u = (user or "").strip().lower() return u.endswith("$") or u in {"system", "localservice", "networkservice"} -def _is_active_sample(data: dict) -> bool: +def _is_active_sample(data: dict): state = str(data.get("state") or "").strip().lower() - if isinstance(data.get("active"), bool): - if data.get("active"): - return True + if isinstance(data.get("active"), bool) and data.get("active"): + return True if ("актив" in state) or (state == "active"): return True - # query user can intermittently return "Unknown" on RDP hosts; if session id is valid - # and user is not a machine/service account, treat it as activity sample. if state == "unknown": try: sid = int(data.get("sessionId")) @@ -48,69 +107,178 @@ def _is_active_sample(data: dict) -> bool: return False -def report_today(): - now_local = datetime.now(REPORT_TZ) - start_local = datetime(now_local.year, now_local.month, now_local.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) - b = get(AW + "/buckets") - sb = next((k for k in b if k.startswith("aw-worktime-sessions_")), None) - if not sb: +def _normalize_user_id(data, host, username): + user_id = str(data.get("userId") or "").strip() + if user_id: + left, sep, right = user_id.partition("\\") + if sep and right: + return f"{resolve_host(host)}\\{right}" + return user_id + return f"{resolve_host(host)}\\{username}" + + +def _event_sample_seconds(event, next_same_session_ts=None): + data = event.get("data") or {} + for key in ("sampleSeconds", "pollSeconds"): + value = data.get(key) + try: + if float(value) > 0: + return clamp_seconds(value) + except Exception: + pass + try: + duration = float(event.get("duration") or 0.0) + except Exception: + duration = 0.0 + if duration > 0: + return clamp_seconds(duration) + if next_same_session_ts is not None: + delta = (next_same_session_ts - event["_ts"]).total_seconds() + if delta > 0: + return clamp_seconds(delta) + return clamp_seconds(DEFAULT_SAMPLE_SECONDS) + + +def _merge_intervals(intervals): + if not intervals: return [] - ev = get(f"{AW}/buckets/{sb}/events?limit=50000") - by = {} - for e in ev: - ts = pts(e.get("timestamp")) + ordered = sorted(intervals, key=lambda item: item[0]) + merged = [ordered[0]] + for start, end in ordered[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + if end > last_end: + merged[-1] = (last_start, end) + continue + merged.append((start, end)) + return merged + + +def aggregate_rows(events, start, end, host): + by_user = {} + by_identity = {} + + for event in events: + ts = pts(event.get("timestamp")) if ts < start or ts > end: continue - d = e.get("data") or {} - user = (d.get("username") or "").strip() - if not user: + data = event.get("data") or {} + username = str(data.get("username") or "").strip() + if not username: continue - active = _is_active_sample(d) - row = by.setdefault(user, {"active": set(), "first": None, "last": None, "rows": 0}) - row["rows"] += 1 - if active: - second = ts.replace(microsecond=0) - row["active"].add(second) - row["first"] = second if row["first"] is None or second < row["first"] else row["first"] - row["last"] = second if row["last"] is None or second > row["last"] else row["last"] + session_id = str(data.get("sessionId") or "").strip() or "unknown" + event_copy = { + "_ts": ts, + "data": data, + "duration": event.get("duration"), + } + by_identity.setdefault((username, session_id), []).append(event_copy) + + for (username, session_id), samples in by_identity.items(): + ordered = sorted(samples, key=lambda item: item["_ts"]) + for idx, sample in enumerate(ordered): + data = sample["data"] + active = _is_active_sample(data) + next_ts = ordered[idx + 1]["_ts"] if idx + 1 < len(ordered) else None + sample_seconds = _event_sample_seconds(sample, next_ts) + row = by_user.setdefault( + username, + { + "user": username, + "user_id": _normalize_user_id(data, host, username), + "samples_count": 0, + "active_samples": 0, + "session_ids": set(), + "intervals": [], + }, + ) + row["samples_count"] += 1 + row["session_ids"].add(session_id) + if active: + row["active_samples"] += 1 + interval_start = sample["_ts"] + interval_end = min(sample["_ts"] + timedelta(seconds=sample_seconds), end + timedelta(seconds=1)) + if interval_end > interval_start: + row["intervals"].append((interval_start, interval_end)) + rows = [] - full = int((end_local - start_local).total_seconds()) - for user in sorted(by): - row = by[user] - active_seconds = len(row["active"]) - rows.append({ - "user": user, - "active_seconds": active_seconds, - "active_hhmm": "%02d:%02d" % (active_seconds // 3600, (active_seconds % 3600) // 60), - "first_activity": row["first"].isoformat().replace("+00:00", "Z") if row["first"] else "", - "last_activity": row["last"].isoformat().replace("+00:00", "Z") if row["last"] else "", - "idle_seconds": max(0, full - active_seconds), - "sessions_count": row["rows"], - }) + full_range = int((end - start).total_seconds()) + 1 + for username in sorted(by_user): + row = by_user[username] + merged = _merge_intervals(row["intervals"]) + active_seconds = int(sum((end_dt - start_dt).total_seconds() for start_dt, end_dt in merged)) + active_seconds = min(active_seconds, full_range) + first_activity = to_iso_utc(merged[0][0]) if merged else "" + last_activity = to_iso_utc(merged[-1][1]) if merged else "" + rows.append( + { + "user": row["user"], + "user_id": row["user_id"], + "active_seconds": active_seconds, + "active_hhmm": hhmm(active_seconds), + "first_activity": first_activity, + "last_activity": last_activity, + "idle_seconds": max(0, full_range - active_seconds), + "sessions_count": len(row["session_ids"]), + "samples_count": row["samples_count"], + "active_samples": row["active_samples"], + } + ) return rows -def render_html(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) + + +def report_today(host): + return report_for_date(host, resolve_report_date()) + + +def report_for_date_fresh(host, report_date): + spec = importlib.util.spec_from_file_location("aw_worktime_runtime", MODULE_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.report_for_date(host, report_date) + + +def render_html(rows, host, report_date, selected_day=None): generated = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") - date_local = datetime.now(REPORT_TZ).strftime("%Y-%m-%d") + date_local = report_date.strftime("%Y-%m-%d") + day_query = f"&day={selected_day}" if selected_day in {"today", "yesterday"} else "" + date_query = f"&date={date_local}" if not day_query else "" trs = [] for row in rows: trs.append( "" f"{row['user']}" - f"{row['active_hhmm']}" + f"{row['user_id']}" + f"{row['active_hhmm']}" f"{row['active_seconds']}" f"{row['first_activity']}" f"{row['last_activity']}" f"{row['idle_seconds']}" f"{row['sessions_count']}" + f"{row['samples_count']}" "" ) if not trs: - trs.append('No data for today yet.') + trs.append('No data for today yet.') return f""" @@ -126,7 +294,6 @@ def render_html(rows): --text: #0f172a; --muted: #475569; --accent: #0f766e; - --accent-2: #1d4ed8; }} * {{ box-sizing: border-box; }} body {{ @@ -138,7 +305,7 @@ def render_html(rows): radial-gradient(circle at top right, rgba(15,118,110,.10), transparent 24%), var(--bg); }} - .wrap {{ max-width: 1180px; margin: 0 auto; padding: 24px; }} + .wrap {{ max-width: 1340px; margin: 0 auto; padding: 24px; }} .hero {{ background: linear-gradient(135deg, #0f172a, #1e293b 58%, #0f766e); color: #fff; @@ -169,14 +336,12 @@ def render_html(rows): th, td {{ padding: 12px 14px; border-bottom: 1px solid var(--line); text-align: left; }} th {{ background: #eef4fb; color: var(--muted); font-weight: 600; position: sticky; top: 0; }} tr:nth-child(even) td {{ background: rgba(148,163,184,.06); }} - .num {{ font-variant-numeric: tabular-nums; }} .good {{ color: var(--accent); font-weight: 700; }} - .muted {{ color: var(--muted); }} @media (max-width: 900px) {{ .wrap {{ padding: 14px; }} .hero h1 {{ font-size: 22px; }} .card {{ overflow-x: auto; }} - table {{ min-width: 820px; }} + table {{ min-width: 1080px; }} }} @@ -184,10 +349,12 @@ def render_html(rows):

RDP Worktime Report

-
Date: {date_local} · Timezone: {REPORT_TZ} · Generated UTC: {generated}
+
Host: {resolve_host(host)} · Date: {date_local} · Timezone: {REPORT_TZ} · Generated UTC: {generated}
- Download CSV - View JSON + Today + Yesterday + Download CSV + View JSON
@@ -195,11 +362,13 @@ def render_html(rows): User + User ID Active Active sec First activity Last activity Idle sec + Sessions Samples @@ -215,8 +384,9 @@ def render_html(rows): class H(BaseHTTPRequestHandler): def do_GET(self): - if self.path.startswith("/dlp-ioc/"): - name = self.path.split("?", 1)[0].rsplit("/", 1)[-1] + parsed = urlparse(self.path) + if parsed.path.startswith("/dlp-ioc/"): + name = parsed.path.rsplit("/", 1)[-1] if name not in {"ioc_blacklist.json", "ioc_blacklist.csv", "ioc_blacklist.sql"}: self.send_response(404) self.end_headers() @@ -241,28 +411,38 @@ class H(BaseHTTPRequestHandler): self.wfile.write(data) return - if not self.path.startswith("/reports/worktime/today"): + if parsed.path != "/reports/worktime/today": self.send_response(404) self.end_headers() return + + params = parse_qs(parsed.query, keep_blank_values=False) fmt = "json" - if "format=csv" in self.path: + if params.get("format", ["json"])[0] == "csv": fmt = "csv" - elif "format=html" in self.path: + elif params.get("format", ["json"])[0] == "html": fmt = "html" - rows = report_today() + host = resolve_host(params.get("host", [DEFAULT_HOST])[0]) + day = params.get("day", ["today"])[0] + date_text = params.get("date", [None])[0] + report_date = resolve_report_date(day=day, date_text=date_text) + rows = report_for_date_fresh(host, report_date) + if fmt == "csv": out = io.StringIO() writer = csv.DictWriter( out, fieldnames=[ "user", + "user_id", "active_seconds", "active_hhmm", "first_activity", "last_activity", "idle_seconds", "sessions_count", + "samples_count", + "active_samples", ], ) writer.writeheader() @@ -274,17 +454,22 @@ class H(BaseHTTPRequestHandler): self.end_headers() self.wfile.write(data) return + if fmt == "html": - data = render_html(rows).encode("utf-8") + data = render_html(rows, host, report_date, selected_day=day if day in {"today", "yesterday"} else None).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return + obj = { "generated_at_utc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "report_timezone": str(REPORT_TZ), + "host": host, + "report_date": report_date.isoformat(), + "bucket_id": get_sessions_bucket_id(host), "rows": rows, } data = json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8") @@ -295,4 +480,9 @@ class H(BaseHTTPRequestHandler): self.wfile.write(data) -HTTPServer(("0.0.0.0", 5610), H).serve_forever() +def main(): + HTTPServer((LISTEN_HOST, LISTEN_PORT), H).serve_forever() + + +if __name__ == "__main__": + main() diff --git a/aw-server/aw-worktime-panel.js b/aw-server/aw-worktime-panel.js index c38b29e..5c7c62b 100644 --- a/aw-server/aw-worktime-panel.js +++ b/aw-server/aw-worktime-panel.js @@ -1,14 +1,22 @@ (function () { var reportBase = "__AW_WORKTIME_REPORT_BASE__"; - var reportUrl = reportBase + "/reports/worktime/today?format=html"; + function defaultDayQuery() { + var now = new Date(); + return now.getHours() < 6 ? "day=yesterday" : "day=today"; + } + + var dayQuery = defaultDayQuery(); + var htmlUrl = reportBase + "/reports/worktime/today?format=html&" + dayQuery; + var csvUrl = reportBase + "/reports/worktime/today?format=csv&" + dayQuery; + var jsonUrl = reportBase + "/reports/worktime/today?" + dayQuery; var existing = document.getElementById("aw-report-links"); if (!existing) return; existing.innerHTML = 'RDP report: ' + - 'HTML | ' + - 'CSV | ' + - 'JSON | ' + + 'HTML | ' + + 'CSV | ' + + 'JSON | ' + 'Panel'; var panel = document.createElement("div"); @@ -32,10 +40,10 @@ '
' + '
RDP Worktime Report
' + '
' + - 'Open' + + 'Open' + 'Close' + "
" + - ''; + ''; document.body.appendChild(panel); diff --git a/aw-server/test_aw_worktime_api.py b/aw-server/test_aw_worktime_api.py new file mode 100644 index 0000000..3f73855 --- /dev/null +++ b/aw-server/test_aw_worktime_api.py @@ -0,0 +1,73 @@ +#!/usr/bin/env python3 +import importlib.util +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) +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + + +def _event(ts, username, session_id, active, **extra): + data = { + "username": username, + "userId": f"WORKGROUP\\{username}", + "sessionId": session_id, + "state": "Активно" if active else "Диск", + "active": active, + } + data.update(extra) + return {"timestamp": ts, "duration": 0.0, "data": data} + + +def test_aggregate_rows_uses_sample_seconds_and_merges_overlap(): + start = datetime(2026, 5, 14, 6, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 5, 14, 6, 59, 59, tzinfo=timezone.utc) + rows = MODULE.aggregate_rows( + [ + _event("2026-05-14T06:00:00Z", "user5", 4, True, sampleSeconds=30), + _event("2026-05-14T06:00:30Z", "user5", 4, True, sampleSeconds=30), + _event("2026-05-14T06:00:15Z", "user5", 5, True, sampleSeconds=30), + _event("2026-05-14T06:01:00Z", "user5", 4, False, sampleSeconds=30), + ], + start, + end, + "SHARKON2025", + ) + assert len(rows) == 1 + row = rows[0] + assert row["user"] == "user5" + assert row["user_id"] == "SHARKON2025\\user5" + assert row["active_seconds"] == 60 + assert row["active_hhmm"] == "00:01" + assert row["sessions_count"] == 2 + assert row["samples_count"] == 4 + assert row["active_samples"] == 3 + assert row["first_activity"] == "2026-05-14T06:00:00Z" + assert row["last_activity"] == "2026-05-14T06:01:00Z" + + +def test_aggregate_rows_falls_back_to_next_sample_delta(): + start = datetime(2026, 5, 14, 7, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 5, 14, 7, 59, 59, tzinfo=timezone.utc) + rows = MODULE.aggregate_rows( + [ + _event("2026-05-14T07:00:00Z", "user1", 3, True), + _event("2026-05-14T07:00:05Z", "user1", 3, True), + _event("2026-05-14T07:00:10Z", "user1", 3, False), + ], + start, + end, + "SHARKON2025", + ) + row = rows[0] + assert row["active_seconds"] == 10 + assert row["active_hhmm"] == "00:00" + + +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" diff --git a/scripts/rdp-worktime-report.sh b/scripts/rdp-worktime-report.sh index 0ff435d..b56b55c 100644 --- a/scripts/rdp-worktime-report.sh +++ b/scripts/rdp-worktime-report.sh @@ -5,6 +5,9 @@ DAY="" FROM="" TO="" AW_BASE_URL="${AW_BASE_URL:-http://10.10.10.13:5600/api/0}" +AW_WORKTIME_HOST="${AW_WORKTIME_HOST:-SHARKON2025}" +AW_WORKTIME_DEFAULT_SAMPLE_SECONDS="${AW_WORKTIME_DEFAULT_SAMPLE_SECONDS:-30}" +AW_WORKTIME_MAX_SAMPLE_SECONDS="${AW_WORKTIME_MAX_SAMPLE_SECONDS:-300}" OUT_DIR="${OUT_DIR:-reports}" usage() { @@ -14,6 +17,7 @@ Usage: $0 --from YYYY-MM-DD --to YYYY-MM-DD Env: AW_BASE_URL (default: ${AW_BASE_URL}) + AW_WORKTIME_HOST (default: ${AW_WORKTIME_HOST}) OUT_DIR (default: ${OUT_DIR}) EOF } @@ -50,14 +54,19 @@ mkdir -p "$OUT_DIR" CSV_OUT="${OUT_DIR}/rdp-worktime-${FROM}_${TO}.csv" JSON_OUT="${OUT_DIR}/rdp-worktime-${FROM}_${TO}.json" -python3 - "$AW_BASE_URL" "$FROM" "$TO" "$CSV_OUT" "$JSON_OUT" <<'PY' +python3 - "$AW_BASE_URL" "$AW_WORKTIME_HOST" "$AW_WORKTIME_DEFAULT_SAMPLE_SECONDS" "$AW_WORKTIME_MAX_SAMPLE_SECONDS" "$FROM" "$TO" "$CSV_OUT" "$JSON_OUT" <<'PY' import csv import json import sys import urllib.request from datetime import datetime, timedelta, timezone -base, from_d, to_d, csv_out, json_out = sys.argv[1:6] +base, host, default_sample, max_sample, from_d, to_d, csv_out, json_out = sys.argv[1:9] +base = (base or "http://10.10.10.13:5600").rstrip("/") +if not base.endswith("/api/0"): + base = base + "/api/0" +default_sample = max(1.0, float(default_sample)) +max_sample = max(default_sample, float(max_sample)) def get_json(url: str): with urllib.request.urlopen(url, timeout=30) as r: @@ -68,21 +77,55 @@ def parse_ts(s): return None return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc) -buckets = get_json(f"{base}/buckets") -sessions_bucket = None -for k in buckets.keys(): - if k.startswith("aw-worktime-sessions_"): - sessions_bucket = k - break +def clamp_seconds(value, fallback=default_sample): + try: + seconds = float(value) + except Exception: + seconds = float(fallback) + if seconds <= 0: + seconds = float(fallback) + return min(seconds, max_sample) -if not sessions_bucket: - raise SystemExit("No aw-worktime-sessions_* bucket found") +def merge_intervals(intervals): + if not intervals: + return [] + intervals = sorted(intervals, key=lambda item: item[0]) + merged = [intervals[0]] + for start, end in intervals[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + if end > last_end: + merged[-1] = (last_start, end) + continue + merged.append((start, end)) + return merged + +def is_active(data): + state = str(data.get("state") or "").strip().lower() + if isinstance(data.get("active"), bool) and data.get("active"): + return True + return ("актив" in state) or (state == "active") + +def normalize_user_id(data, host_name, username): + raw = str(data.get("userId") or "").strip() + if raw and "\\" in raw: + _, right = raw.split("\\", 1) + return f"{host_name}\\{right}" + if raw: + return raw + return f"{host_name}\\{username}" + +bucket_id = f"aw-worktime-sessions_{host}" +try: + get_json(f"{base}/buckets/{bucket_id}") +except Exception: + raise SystemExit(f"Bucket not found: {bucket_id}") start = datetime.fromisoformat(from_d + "T00:00:00+00:00") end = datetime.fromisoformat(to_d + "T23:59:59+00:00") -ev = get_json(f"{base}/buckets/{sessions_bucket}/events?limit=20000") -by_user = {} +ev = get_json(f"{base}/buckets/{bucket_id}/events?limit=50000") +by_identity = {} for e in ev: ts = parse_ts(e.get("timestamp")) if ts is None or ts < start or ts > end: @@ -91,40 +134,89 @@ for e in ev: user = (d.get("username") or "").strip() if not user: continue - state = (d.get("state") or "").strip().lower() - is_active = ("актив" in state) or (state == "active") - rec = by_user.setdefault(user, {"active_ts": set(), "first": None, "last": None, "rows": 0}) - rec["rows"] += 1 - if is_active: - rec["active_ts"].add(ts.replace(microsecond=0)) - rec["first"] = ts if rec["first"] is None or ts < rec["first"] else rec["first"] - rec["last"] = ts if rec["last"] is None or ts > rec["last"] else rec["last"] + session_id = str(d.get("sessionId") or "").strip() or "unknown" + by_identity.setdefault((user, session_id), []).append({ + "ts": ts, + "duration": e.get("duration"), + "data": d, + }) rows = [] -full_range = int((end - start).total_seconds()) +full_range = int((end - start).total_seconds()) + 1 +by_user = {} +for (user, session_id), samples in by_identity.items(): + samples = sorted(samples, key=lambda item: item["ts"]) + for idx, sample in enumerate(samples): + data = sample["data"] + rec = by_user.setdefault(user, { + "user": user, + "user_id": normalize_user_id(data, host, user), + "sessions": set(), + "samples_count": 0, + "active_samples": 0, + "intervals": [], + }) + rec["sessions"].add(session_id) + rec["samples_count"] += 1 + if not is_active(data): + continue + rec["active_samples"] += 1 + sample_seconds = None + for key in ("sampleSeconds", "pollSeconds"): + value = data.get(key) + try: + if float(value) > 0: + sample_seconds = clamp_seconds(value) + break + except Exception: + pass + if sample_seconds is None: + try: + duration = float(sample.get("duration") or 0.0) + except Exception: + duration = 0.0 + if duration > 0: + sample_seconds = clamp_seconds(duration) + else: + next_ts = samples[idx + 1]["ts"] if idx + 1 < len(samples) else None + if next_ts is not None: + sample_seconds = clamp_seconds((next_ts - sample["ts"]).total_seconds()) + else: + sample_seconds = clamp_seconds(default_sample) + interval_end = min(sample["ts"] + timedelta(seconds=sample_seconds), end + timedelta(seconds=1)) + if interval_end > sample["ts"]: + rec["intervals"].append((sample["ts"], interval_end)) + for user in sorted(by_user.keys()): rec = by_user[user] - active = len(rec["active_ts"]) + merged = merge_intervals(rec["intervals"]) + active = int(sum((finish - begin).total_seconds() for begin, finish in merged)) + active = min(active, full_range) idle = max(0, full_range - active) rows.append({ - "user": user, + "user": rec["user"], + "user_id": rec["user_id"], "active_seconds": int(active), "active_hhmm": f"{int(active)//3600:02d}:{(int(active)%3600)//60:02d}", - "first_activity": rec["first"].isoformat().replace("+00:00","Z") if rec["first"] else "", - "last_activity": rec["last"].isoformat().replace("+00:00","Z") if rec["last"] else "", + "first_activity": merged[0][0].isoformat().replace("+00:00","Z") if merged else "", + "last_activity": merged[-1][1].isoformat().replace("+00:00","Z") if merged else "", "idle_seconds": int(idle), - "sessions_count": rec["rows"], + "sessions_count": len(rec["sessions"]), + "samples_count": rec["samples_count"], + "active_samples": rec["active_samples"], }) with open(csv_out, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=[ - "user","active_seconds","active_hhmm","first_activity","last_activity","idle_seconds","sessions_count" + "user","user_id","active_seconds","active_hhmm","first_activity","last_activity","idle_seconds","sessions_count","samples_count","active_samples" ]) w.writeheader() w.writerows(rows) with open(json_out, "w", encoding="utf-8") as f: json.dump({ + "host": host, + "bucket_id": bucket_id, "from": from_d, "to": to_d, "generated_at_utc": datetime.now(timezone.utc).isoformat().replace("+00:00","Z"), diff --git a/windows/worktime-session-collector.ps1 b/windows/worktime-session-collector.ps1 index 7db2292..84fbd04 100644 --- a/windows/worktime-session-collector.ps1 +++ b/windows/worktime-session-collector.ps1 @@ -1,7 +1,7 @@ param( [string]$ConfigPath = 'C:\ProgramData\AWatch-rus\deployment-config.json', [string]$Hostname, - [int]$PollSeconds = 30 + [int]$PollSeconds = 0 ) # Force UTF-8 for console I/O @@ -198,14 +198,46 @@ function Test-SessionIsActive { return ($s -match 'active') -or ($s -match '\u0430\u043a\u0442\u0438\u0432') } +function Get-CanonicalUserId { + param( + [pscustomobject]$Config, + [string]$HostnameValue, + [string]$Username + ) + + $normalizedUser = [string]$Username + if ([string]::IsNullOrWhiteSpace($normalizedUser)) { + return '' + } + + if ($Config -and $Config.PSObject.Properties.Name -contains 'userTasks' -and $Config.userTasks) { + foreach ($task in @($Config.userTasks)) { + try { + $taskUserId = [string]$task.userId + if ([string]::IsNullOrWhiteSpace($taskUserId)) { + continue + } + $parts = $taskUserId -split '\\', 2 + if ($parts.Count -eq 2 -and $parts[1].Equals($normalizedUser, [System.StringComparison]::OrdinalIgnoreCase)) { + return $taskUserId + } + } + catch { + } + } + } + + return "$HostnameValue\$normalizedUser" +} + # Main $cfg = Get-Config -Path $ConfigPath $hostValue = if ($Hostname -and $Hostname.Trim()) { $Hostname.Trim() } elseif ($cfg -and $cfg.PSObject.Properties.Name -contains 'awHostname' -and -not [string]::IsNullOrWhiteSpace([string]$cfg.awHostname)) { [string]$cfg.awHostname } elseif ($cfg -and $cfg.awHostname) { [string]$cfg.awHostname } else { [string]$env:COMPUTERNAME } try { $apiBase = '{0}://{1}:{2}/api/0' -f [string]$cfg.server.scheme, [string]$cfg.server.host, [string]$cfg.server.port } catch { throw 'Invalid server configuration in config file.' } $bucketId = 'aw-worktime-sessions_' + $hostValue -$pulse = 120 $sleepSec = if ($PollSeconds -gt 0) { $PollSeconds } elseif ($cfg.collector -and $cfg.collector.pollSeconds) { [int]$cfg.collector.pollSeconds } else { 30 } +$pulse = [Math]::Max($sleepSec * 3, 30) Ensure-Bucket -ApiBase $apiBase -BucketId $bucketId -HostnameValue $hostValue @@ -234,16 +266,19 @@ while ($true) { } foreach ($rec in $records) { + $canonicalUserId = Get-CanonicalUserId -Config $cfg -HostnameValue $hostValue -Username ([string]$rec.username) $payloadObj = [PSCustomObject]@{ timestamp = $now - duration = 0 + duration = $sleepSec data = [PSCustomObject]@{ username = [string]$rec.username - userId = "${env:USERDOMAIN}\$($rec.username)" + userId = $canonicalUserId sessionId = [int]$rec.sessionId sessionName = [string]$rec.sessionName state = [string]$rec.state active = Test-SessionIsActive -State ([string]$rec.state) + sampleSeconds = $sleepSec + pollSeconds = $sleepSec hostname = $hostValue source = 'worktime-session-collector' }