awatch-rus/AWatch-rus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(aw): harden collectors and grafana exports

Browse Source
This commit is contained in:
igor04091968
2026-05-28 10:54:46 +03:00
parent 524ab33f63
commit cc9e4a0669
29 changed files with 1181 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aw-server/aw-worktime-ui-bridge.py
+122 -21
View File
@@ -12,7 +12,7 @@ STATE_PATH = os.environ.get(
"AW_WORKTIME_UI_BRIDGE_STATE",
"/var/lib/activitywatch/aw-worktime-ui-bridge-state.json",
)
TIMEOUT = float(os.environ.get("AW_WORKTIME_UI_BRIDGE_TIMEOUT", "20"))
TIMEOUT = float(os.environ.get("AW_WORKTIME_UI_BRIDGE_TIMEOUT", "60"))
WATCHER_FALLBACK_ENABLED = os.environ.get("AW_WORKTIME_UI_BRIDGE_WATCHER_FALLBACK", "1").strip().lower() not in {
"0",
"false",
@@ -29,6 +29,8 @@ WATCHER_AFK_BUCKET = f"aw-watcher-afk_{HOST}"
WATCHER_WINDOW_BUCKET = f"aw-watcher-window_{HOST}"
WEB_CATEGORY_BUCKET = f"aw-detmir-web-category_{HOST}"
COLLECTOR_HEALTH_MAX_AGE_SECONDS = float(os.environ.get("AW_WORKTIME_UI_BRIDGE_COLLECTOR_HEALTH_MAX_AGE_SECONDS", "300"))
COLLECTOR_HEALTH_QUERY_LIMIT = int(os.environ.get("AW_WORKTIME_UI_BRIDGE_COLLECTOR_HEALTH_QUERY_LIMIT", "200"))
FOREGROUND_CONTEXT_CACHE_SECONDS = float(os.environ.get("AW_WORKTIME_UI_BRIDGE_FOREGROUND_CACHE_SECONDS", "900"))
def _req(method: str, path: str, payload=None):
@@ -100,6 +102,8 @@ def watcher_window_needs_bridge_sync(now_utc: datetime):
source = str(data.get("source", "")).strip().lower()
app = str(data.get("app", "")).strip()
title = str(data.get("title", "")).strip()
if source == "aw-worktime-ui-bridge":
return True
if source != "aw-worktime-ui-bridge":
return False
if app.upper() == "RDP":
@@ -148,22 +152,54 @@ def build_window_title(users, active_count):
return f"RDP active ({active_count}): " + ", ".join(users)
def get_latest_foreground_context(now_utc: datetime):
def get_latest_active_session_ids(events):
grouped = {}
for event in events:
ts = event.get("timestamp")
if not ts:
continue
grouped.setdefault(ts, []).append(event)
if not grouped:
return set()
latest_ts = max(grouped.keys(), key=lambda item: parse_iso_utc(item))
active_session_ids = set()
for event in grouped.get(latest_ts, []):
data = event.get("data") or {}
if not _is_session_active(data):
continue
try:
active_session_ids.add(int(data.get("sessionId")))
except Exception:
continue
return active_session_ids
def _normalize_foreground_context(data):
foreground_process = str(data.get("foregroundProcess", "")).strip()
foreground_title = str(data.get("foregroundTitle", "")).strip()
if not foreground_process and not foreground_title:
return None
return {
"app": foreground_process if foreground_process.endswith(".exe") else f"{foreground_process}.exe",
"title": foreground_title or foreground_process,
}
def get_latest_foreground_context(now_utc: datetime, active_session_ids=None, state=None):
try:
events = _req("GET", f"/api/0/buckets/{WEB_CATEGORY_BUCKET}/events?limit=20") or []
events = _req("GET", f"/api/0/buckets/{WEB_CATEGORY_BUCKET}/events?limit={COLLECTOR_HEALTH_QUERY_LIMIT}") or []
except urllib.error.HTTPError as e:
if e.code == 404:
return None
raise
events = []
else:
raise
for event in events:
active_session_ids = set(active_session_ids or [])
recent_candidates = []
for event in reversed(events):
data = event.get("data") or {}
if str(data.get("signalType", "")).strip().lower() != "collector_health":
continue
foreground_process = str(data.get("foregroundProcess", "")).strip()
foreground_title = str(data.get("foregroundTitle", "")).strip()
if not foreground_process and not foreground_title:
continue
ts = event.get("timestamp")
if not ts:
continue
@@ -173,11 +209,43 @@ def get_latest_foreground_context(now_utc: datetime):
continue
if (now_utc - event_dt).total_seconds() > COLLECTOR_HEALTH_MAX_AGE_SECONDS:
continue
return {
"app": foreground_process if foreground_process.endswith(".exe") else f"{foreground_process}.exe",
"title": foreground_title or foreground_process,
}
normalized = _normalize_foreground_context(data)
if not normalized:
continue
session_id = data.get("sessionId")
try:
session_id = int(session_id)
except Exception:
session_id = None
recent_candidates.append((session_id, event_dt, normalized))
for session_id, event_dt, normalized in recent_candidates:
if active_session_ids and session_id in active_session_ids:
normalized["timestamp"] = to_iso_utc(event_dt.isoformat())
return normalized
if recent_candidates:
session_id, event_dt, normalized = recent_candidates[0]
normalized["timestamp"] = to_iso_utc(event_dt.isoformat())
return normalized
cached = (state or {}).get("last_foreground_context")
if isinstance(cached, dict):
cached_ts = str(cached.get("timestamp", "")).strip()
if cached_ts:
try:
cached_dt = parse_iso_utc(cached_ts)
except Exception:
cached_dt = None
if cached_dt and (now_utc - cached_dt).total_seconds() <= FOREGROUND_CONTEXT_CACHE_SECONDS:
app = str(cached.get("app", "")).strip()
title = str(cached.get("title", "")).strip()
if app or title:
return {
"app": app or "RDP",
"title": title or app or "RDP",
"timestamp": cached_ts,
}
return None
@@ -229,13 +297,19 @@ def transform(events, foreground_context=None):
rows = grouped[ts]
src_duration = max(float(r.get("duration", 0.0)) for r in rows)
duration = src_duration
cur_dt = parsed_ts.get(ts)
next_dt = parsed_ts.get(ordered_ts[idx + 1]) if idx + 1 < len(ordered_ts) else None
next_gap = None
if cur_dt and next_dt:
next_gap = max(0.0, (next_dt - cur_dt).total_seconds())
if duration <= 0:
cur_dt = parsed_ts.get(ts)
next_dt = parsed_ts.get(ordered_ts[idx + 1]) if idx + 1 < len(ordered_ts) else None
if cur_dt and next_dt:
duration = max(0.0, (next_dt - cur_dt).total_seconds())
duration = next_gap or 0.0
if duration <= 0:
duration = 10.0
elif next_gap is not None and next_gap > 0:
# Do not let a sampled session interval extend past the next sample.
duration = min(duration, next_gap)
duration = min(duration, 30.0)
active_users = []
for r in rows:
@@ -271,6 +345,22 @@ def transform(events, foreground_context=None):
return out_afk, out_win, last_ts
def normalize_watcher_window_events(win_events):
normalized = []
for event in win_events:
cloned = dict(event)
data = dict(event.get("data") or {})
app = str(data.get("app") or "").strip()
title = str(data.get("title") or "").strip()
if not app or app.upper() == "RDP":
continue
if app and app.upper() != "RDP" and " | RDP active (" in title:
data["title"] = title.split(" | RDP active (", 1)[0].strip()
cloned["data"] = data
normalized.append(cloned)
return normalized
def main():
state = load_state()
last_ts = state.get("last_ts", "1970-01-01T00:00:00Z")
@@ -301,10 +391,12 @@ def main():
if not events:
return
foreground_context = get_latest_foreground_context(now_utc)
active_session_ids = get_latest_active_session_ids(events)
foreground_context = get_latest_foreground_context(now_utc, active_session_ids=active_session_ids, state=state)
afk_events, win_events, new_last_ts = transform(events, foreground_context=foreground_context)
if not afk_events or not win_events or not new_last_ts:
return
watcher_win_events = normalize_watcher_window_events(win_events)
_req("POST", f"/api/0/buckets/{AFK_BUCKET}/events", afk_events)
_req("POST", f"/api/0/buckets/{WINDOW_BUCKET}/events", win_events)
@@ -312,10 +404,19 @@ def main():
if bucket_needs_fallback(WATCHER_AFK_BUCKET, now_utc, WATCHER_FALLBACK_STALE_SECONDS):
ensure_bucket(WATCHER_AFK_BUCKET, "afkstatus", "aw-watcher-afk")
_req("POST", f"/api/0/buckets/{WATCHER_AFK_BUCKET}/events", afk_events)
if watcher_window_needs_bridge_sync(now_utc):
if watcher_win_events and watcher_window_needs_bridge_sync(now_utc):
ensure_bucket(WATCHER_WINDOW_BUCKET, "currentwindow", "aw-watcher-window")
_req("POST", f"/api/0/buckets/{WATCHER_WINDOW_BUCKET}/events", win_events)
save_state({"last_ts": new_last_ts})
_req("POST", f"/api/0/buckets/{WATCHER_WINDOW_BUCKET}/events", watcher_win_events)
next_state = {"last_ts": new_last_ts}
if foreground_context:
next_state["last_foreground_context"] = {
"app": str(foreground_context.get("app") or ""),
"title": str(foreground_context.get("title") or ""),
"timestamp": str(foreground_context.get("timestamp") or to_iso_utc(now_utc.isoformat())),
}
elif isinstance(state.get("last_foreground_context"), dict):
next_state["last_foreground_context"] = state["last_foreground_context"]
save_state(next_state)
print(f"posted_afk={len(afk_events)} posted_win={len(win_events)} last_ts={new_last_ts}")