#!/usr/bin/env python3
import csv
import html
import io
import importlib.util
import json
import os
import sys
import tempfile
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, urlencode, urlparse
from zoneinfo import ZoneInfo
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"))
WORKDAY_START_HOUR = int(os.environ.get("AW_WORKTIME_MANAGER_START_HOUR", "9"))
WORKDAY_END_HOUR = int(os.environ.get("AW_WORKTIME_MANAGER_END_HOUR", "18"))
MANAGER_TARGET_COVERAGE_PCT = max(1, min(100, int(os.environ.get("AW_WORKTIME_MANAGER_TARGET_COVERAGE_PCT", "75"))))
MANAGER_LOW_COVERAGE_PCT = max(1, min(100, int(os.environ.get("AW_WORKTIME_MANAGER_LOW_COVERAGE_PCT", "35"))))
MANAGER_LATE_START_GRACE_MINUTES = max(0, int(os.environ.get("AW_WORKTIME_MANAGER_LATE_START_GRACE_MINUTES", "60")))
MANAGER_EARLY_FINISH_GRACE_MINUTES = max(0, int(os.environ.get("AW_WORKTIME_MANAGER_EARLY_FINISH_GRACE_MINUTES", "90")))
MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS = max(60, int(os.environ.get("AW_WORKTIME_MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS", "900")))
MANAGER_WEB_SOURCE_MAX_AGE_SECONDS = max(3600, int(os.environ.get("AW_WORKTIME_MANAGER_WEB_SOURCE_MAX_AGE_SECONDS", "259200")))
MANAGER_SESSION_SOURCE_MAX_AGE_SECONDS = max(3600, int(os.environ.get("AW_WORKTIME_MANAGER_SESSION_SOURCE_MAX_AGE_SECONDS", "604800")))
MANAGER_INFRA_SOURCE_MAX_AGE_SECONDS = max(3600, int(os.environ.get("AW_WORKTIME_MANAGER_INFRA_SOURCE_MAX_AGE_SECONDS", "172800")))
MANAGER_TREND_DAYS = max(3, min(31, int(os.environ.get("AW_WORKTIME_MANAGER_TREND_DAYS", "7"))))
MANAGER_CACHE_TTL_SECONDS = max(0, int(os.environ.get("AW_WORKTIME_MANAGER_CACHE_TTL_SECONDS", "120")))
MANAGER_CACHE_DIR = Path(os.environ.get("AW_WORKTIME_MANAGER_CACHE_DIR", "/var/lib/activitywatch/worktime-cache"))
MANAGER_ALIASES_JSON = Path(os.environ.get("AW_WORKTIME_MANAGER_ALIASES_JSON", "/etc/activitywatch/worktime-manager-aliases.json"))
MANAGER_EXCLUDE_USERS = {item.strip().lower() for item in os.environ.get("AW_WORKTIME_MANAGER_EXCLUDE_USERS", "").split(",") if item.strip()}
MODULE_PATH = Path(__file__).resolve()
_ALIASES_CACHE = {"mtime": None, "payload": {}}
def get(u):
with urllib.request.urlopen(u, timeout=30) as r:
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 to_iso_utc(dt):
return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
def parse_iso_utc(value):
if not value:
return None
return pts(value)
def hhmm(total_seconds):
total_seconds = max(0, int(total_seconds))
return "%02d:%02d" % (total_seconds // 3600, (total_seconds % 3600) // 60)
def now_utc():
return datetime.now(timezone.utc)
def age_seconds(ts, now=None):
if ts is None:
return None
ref = now or now_utc()
return max(0, int((ref - ts).total_seconds()))
def write_atomic_json(path, payload):
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
json.dump(payload, handle, ensure_ascii=False, indent=2)
handle.write("\n")
tmp_name = handle.name
os.replace(tmp_name, path)
def safe_slug(value):
text = str(value or "").strip().lower()
slug = []
for char in text:
if char.isalnum():
slug.append(char)
else:
slug.append("-")
normalized = "".join(slug).strip("-")
while "--" in normalized:
normalized = normalized.replace("--", "-")
return normalized or "user"
def _normalize_identity_key(value):
return str(value or "").strip().lower()
def _default_display_name(user, user_id):
base = str(user_id or user or "").strip()
if "\\" in base:
base = base.split("\\", 1)[1]
if not base:
base = str(user or "").strip()
if base and base.isascii() and base.lower() == base and any(ch.isalpha() for ch in base):
return base.upper()
return base or str(user or "").strip()
def load_manager_aliases():
try:
stat = MANAGER_ALIASES_JSON.stat()
except FileNotFoundError:
_ALIASES_CACHE["mtime"] = None
_ALIASES_CACHE["payload"] = {}
return {}
mtime = stat.st_mtime
if _ALIASES_CACHE["mtime"] == mtime:
return _ALIASES_CACHE["payload"]
try:
raw = json.loads(MANAGER_ALIASES_JSON.read_text(encoding="utf-8"))
except Exception as exc:
log_warning(f"failed to load manager aliases from {MANAGER_ALIASES_JSON}: {exc}")
raw = {}
payload = {}
if isinstance(raw, dict):
users = raw.get("users", raw)
if isinstance(users, dict):
for key, value in users.items():
norm_key = _normalize_identity_key(key)
if not norm_key:
continue
if isinstance(value, str):
payload[norm_key] = {"display_name": value}
elif isinstance(value, dict):
payload[norm_key] = dict(value)
_ALIASES_CACHE["mtime"] = mtime
_ALIASES_CACHE["payload"] = payload
return payload
def resolve_user_alias(user, user_id, host):
aliases = load_manager_aliases()
candidates = [
_normalize_identity_key(user_id),
_normalize_identity_key(f"{resolve_host(host)}\\{user}"),
_normalize_identity_key(user),
]
alias = {}
for candidate in candidates:
if candidate and candidate in aliases:
alias = dict(aliases[candidate])
break
display_name = str(alias.get("display_name") or alias.get("name") or _default_display_name(user, user_id)).strip()
manager_owner = str(alias.get("manager") or alias.get("owner") or display_name).strip() or display_name
department = str(alias.get("department") or "").strip()
role = str(alias.get("role") or "").strip()
notes = str(alias.get("notes") or "").strip()
canonical_user_id = str(alias.get("canonical_user_id") or user_id or "").strip()
exclude = bool(alias.get("exclude")) or _normalize_identity_key(user) in MANAGER_EXCLUDE_USERS or _normalize_identity_key(display_name) in MANAGER_EXCLUDE_USERS
return {
"display_name": display_name,
"manager_owner": manager_owner,
"department": department,
"role": role,
"notes": notes,
"canonical_user_id": canonical_user_id,
"exclude": exclude,
}
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 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 get_workday_bounds(report_date):
start_local = datetime(report_date.year, report_date.month, report_date.day, WORKDAY_START_HOUR, 0, 0, tzinfo=REPORT_TZ)
end_local = datetime(report_date.year, report_date.month, report_date.day, WORKDAY_END_HOUR, 0, 0, tzinfo=REPORT_TZ)
if end_local <= start_local:
end_local = start_local + timedelta(hours=8)
return {
"start_local": start_local,
"end_local": end_local,
"duration_seconds": int((end_local - start_local).total_seconds()),
}
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):
state = str(data.get("state") or "").strip().lower()
if isinstance(data.get("active"), bool) and data.get("active"):
return True
if ("актив" in state) or (state == "active"):
return True
if state == "unknown":
try:
sid = int(data.get("sessionId"))
except Exception:
sid = -1
user = str(data.get("username") or "").strip()
session_name = str(data.get("sessionName") or "").strip().lower()
if sid > 0 and user and (not _is_machine_user(user)) and (session_name.startswith("rdp-") or session_name == "console"):
return True
return False
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 []
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 _collect_user_rows(events, start, end, host):
end_exclusive = end + timedelta(seconds=1)
by_user = {}
by_identity = {}
for event in events:
ts = pts(event.get("timestamp"))
if ts < start or ts > end:
continue
data = event.get("data") or {}
username = str(data.get("username") or "").strip()
if not username:
continue
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_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)
return _build_rows_from_user_map(by_user, start, end, include_intervals=False)
def aggregate_rows_with_intervals(events, start, end, host):
by_user = _collect_user_rows(events, start, end, host)
return _build_rows_from_user_map(by_user, start, end, include_intervals=True)
def _build_rows_from_user_map(by_user, start, end, include_intervals):
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"],
}
)
if include_intervals:
rows[-1]["_intervals"] = merged
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 {
"users_count": 0,
"total_active_seconds": 0,
"total_active_hhmm": "00:00",
"first_activity": "",
"last_activity": "",
"top_user": "",
"top_user_active_hhmm": "00:00",
}
total_active_seconds = sum(int(row.get("active_seconds", 0) or 0) for row in rows)
first_values = [row.get("first_activity") for row in rows if row.get("first_activity")]
last_values = [row.get("last_activity") for row in rows if row.get("last_activity")]
top_row = max(rows, key=lambda row: int(row.get("active_seconds", 0) or 0))
return {
"users_count": len(rows),
"total_active_seconds": total_active_seconds,
"total_active_hhmm": hhmm(total_active_seconds),
"first_activity": min(first_values) if first_values else "",
"last_activity": max(last_values) if last_values else "",
"top_user": top_row.get("user", ""),
"top_user_active_hhmm": top_row.get("active_hhmm", "00:00"),
}
def latest_bucket_event(bucket_id):
try:
events = get(f"{AW}/buckets/{bucket_id}/events?limit=20")
except Exception:
return None
if not isinstance(events, list) or not events:
return None
valid = [item for item in events if isinstance(item, dict)]
if not valid:
return None
valid.sort(key=lambda item: item.get("timestamp") or "", reverse=True)
return valid[0]
def _priority_rank(priority):
return {"critical": 0, "high": 1, "medium": 2, "low": 3}.get(priority, 9)
def _clamp_pct(value):
return round(min(100.0, max(0.0, float(value))), 2)
def _action(action_id, priority, owner, deadline_hint, reason, recommended_action, *, user_id="", evidence=None):
return {
"action_id": action_id,
"priority": priority,
"owner": owner,
"user_id": user_id,
"deadline_hint": deadline_hint,
"reason": reason,
"recommended_action": recommended_action,
"evidence": evidence or {},
}
def build_executive_summary(summary, actions, sources):
critical = [action for action in actions if action.get("priority") == "critical"]
high = [action for action in actions if action.get("priority") == "high"]
stale_sources = [source for source in sources if source.get("status") != "ok"]
if critical:
portfolio_state = "critical"
headline = f"Есть {len(critical)} критичных вопроса, требующих решения сегодня."
elif high:
portfolio_state = "attention"
headline = f"Критичных провалов нет, но есть {len(high)} вопроса повышенного внимания."
elif summary.get("portfolio_coverage_pct", 0.0) < MANAGER_TARGET_COVERAGE_PCT:
portfolio_state = "attention"
headline = "Покрытие ниже целевого порога, но явных критичных кейсов не найдено."
else:
portfolio_state = "stable"
headline = "Критичных отклонений не найдено, рабочий день идёт в пределах нормы."
message_parts = [
f"Активны {summary.get('active_users', 0)} из {summary.get('users_count', 0)} сотрудников.",
f"Покрытие рабочего окна {summary.get('portfolio_coverage_pct', 0.0)}%.",
]
if stale_sources:
message_parts.append(f"Есть {len(stale_sources)} проблем(ы) со свежестью источников.")
message = " ".join(message_parts)
focus_items = []
for action in actions[:5]:
focus_items.append(
{
"priority": action["priority"],
"owner": action["owner"],
"title": action["action_id"],
"reason": action["reason"],
"recommended_action": action["recommended_action"],
}
)
stale_items = []
for source in stale_sources[:3]:
stale_items.append(
{
"source_id": source["source_id"],
"label": source["label"],
"status": source["status_label"],
"summary": source["summary"],
}
)
return {
"portfolio_state": portfolio_state,
"headline": headline,
"message": message,
"focus_items": focus_items,
"stale_sources": stale_items,
}
def _interval_overlap_seconds(intervals, start, end):
total = 0
first = None
last = None
for interval_start, interval_end in intervals or []:
overlap_start = max(interval_start, start)
overlap_end = min(interval_end, end)
if overlap_end <= overlap_start:
continue
seconds = int((overlap_end - overlap_start).total_seconds())
if seconds <= 0:
continue
total += seconds
if first is None or overlap_start < first:
first = overlap_start
if last is None or overlap_end > last:
last = overlap_end
return total, first, last
def _source_status_label(status):
return {
"ok": "fresh",
"warn": "stale",
"fail": "missing",
}.get(status, status)
def _source_summary(event):
data = (event or {}).get("data") or {}
signal_type = str(data.get("signalType") or "").strip()
if signal_type == "collector_health":
return (
f"queue={data.get('queueDepth', 0)} "
f"failures={data.get('sendFailures', 0)} "
f"flushed={data.get('eventsFlushed', 0)}"
)
if data.get("domain"):
return f"{data.get('domain')} ({data.get('category', 'uncategorized')})"
if data.get("eventType"):
return f"{data.get('eventType')} {data.get('username', '')}".strip()
if data.get("action"):
return f"{data.get('action')} {data.get('result', '')}".strip()
if data.get("title"):
return str(data.get("title"))[:120]
if data.get("status"):
return str(data.get("status"))
if data.get("app"):
return str(data.get("app"))
return ""
def management_cache_path(host, report_date):
host_slug = safe_slug(host)
return MANAGER_CACHE_DIR / f"{host_slug}-{report_date.isoformat()}.json"
def load_management_cache(host, report_date):
if MANAGER_CACHE_TTL_SECONDS <= 0:
return None
path = management_cache_path(host, report_date)
if not path.exists():
return None
age = age_seconds(datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc))
if age is None or age > MANAGER_CACHE_TTL_SECONDS:
return None
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
return None
def save_management_cache(host, report_date, payload):
if MANAGER_CACHE_TTL_SECONDS <= 0:
return
write_atomic_json(management_cache_path(host, report_date), payload)
def build_source_freshness(host):
source_specs = [
{
"source_id": "worktime_sessions",
"label": "RDP worktime sessions",
"bucket_candidates": [f"aw-worktime-sessions_{host}"],
"max_age_seconds": MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS,
"required": True,
"owner": "ops",
},
{
"source_id": "rdp_window",
"label": "RDP current window",
"bucket_candidates": [f"aw-rdp-window_{host}"],
"max_age_seconds": MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS,
"required": True,
"owner": "ops",
},
{
"source_id": "rdp_afk",
"label": "RDP AFK",
"bucket_candidates": [f"aw-rdp-afk_{host}"],
"max_age_seconds": MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS,
"required": True,
"owner": "ops",
},
{
"source_id": "watcher_window",
"label": "Local watcher window",
"bucket_candidates": [f"aw-watcher-window_{host}"],
"max_age_seconds": MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS,
"required": True,
"owner": "ops",
},
{
"source_id": "watcher_afk",
"label": "Local watcher AFK",
"bucket_candidates": [f"aw-watcher-afk_{host}"],
"max_age_seconds": MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS,
"required": True,
"owner": "ops",
},
{
"source_id": "file_operations",
"label": "File operations collector",
"bucket_candidates": [f"aw-file-operations_{host}", "aw-file-operations_10.10.10.13"],
"max_age_seconds": MANAGER_CRITICAL_SOURCE_MAX_AGE_SECONDS,
"required": True,
"owner": "ops",
},
{
"source_id": "web_categories",
"label": "Browser/web categories",
"bucket_candidates": [f"aw-detmir-web-category_{host}"],
"max_age_seconds": MANAGER_WEB_SOURCE_MAX_AGE_SECONDS,
"required": False,
"owner": "ops",
},
{
"source_id": "session_events",
"label": "Windows session events",
"bucket_candidates": [f"aw-session-events_{host}"],
"max_age_seconds": MANAGER_SESSION_SOURCE_MAX_AGE_SECONDS,
"required": False,
"owner": "ops",
},
{
"source_id": "pve_tasks",
"label": "PVE task feed",
"bucket_candidates": ["aw-pve-task-events_pve-detmir"],
"max_age_seconds": MANAGER_INFRA_SOURCE_MAX_AGE_SECONDS,
"required": False,
"owner": "ops",
},
]
now = now_utc()
sources = []
actions = []
for spec in source_specs:
matched_bucket = ""
matched_event = None
matched_age = None
for candidate in spec["bucket_candidates"]:
event = latest_bucket_event(candidate)
if not event:
continue
ts = parse_iso_utc(event.get("timestamp"))
candidate_age = age_seconds(ts, now=now)
if matched_event is None or (candidate_age is not None and (matched_age is None or candidate_age < matched_age)):
matched_bucket = candidate
matched_event = event
matched_age = candidate_age
ts = parse_iso_utc((matched_event or {}).get("timestamp"))
age = matched_age if matched_event is not None else age_seconds(ts, now=now)
if matched_event is None:
status = "fail" if spec["required"] else "warn"
summary = "bucket missing or empty"
elif age is None:
status = "warn"
summary = "timestamp parse failed"
elif age > spec["max_age_seconds"]:
status = "fail" if spec["required"] else "warn"
summary = f"stale ({age}s)"
else:
status = "ok"
summary = f"fresh ({age}s)"
detail = {
"source_id": spec["source_id"],
"label": spec["label"],
"status": status,
"status_label": _source_status_label(status),
"bucket_id": matched_bucket or spec["bucket_candidates"][0],
"timestamp": (matched_event or {}).get("timestamp", ""),
"age_seconds": age,
"required": spec["required"],
"max_age_seconds": spec["max_age_seconds"],
"summary": summary,
"event_summary": _source_summary(matched_event),
}
sources.append(detail)
if status == "ok":
continue
priority = "critical" if spec["required"] else "medium"
actions.append(
_action(
"source_freshness_review",
priority,
spec["owner"],
"today" if spec["required"] else "3d",
f"Источник '{spec['label']}' в состоянии {detail['status_label']}: {summary}.",
"Проверить collector/service, причину отставания и подтвердить, что управленческие выводы по данным ещё надёжны.",
evidence={
"source_id": spec["source_id"],
"bucket_id": detail["bucket_id"],
"age_seconds": age,
"required": spec["required"],
},
)
)
return sources, actions
def _build_management_core(rows, host, report_date):
calendar_summary = build_report_summary(rows)
report_bounds = get_report_bounds(report_date)
workday = get_workday_bounds(report_date)
now_local = datetime.now(REPORT_TZ)
is_today = report_date == now_local.date()
effective_end_local = min(now_local, workday["end_local"]) if is_today else workday["end_local"]
elapsed_seconds = max(0, int((effective_end_local - workday["start_local"]).total_seconds()))
if not is_today:
elapsed_seconds = workday["duration_seconds"]
expected_seconds_per_user = min(workday["duration_seconds"], max(0, elapsed_seconds))
target_seconds = int(expected_seconds_per_user * (MANAGER_TARGET_COVERAGE_PCT / 100.0))
low_seconds = int(expected_seconds_per_user * (MANAGER_LOW_COVERAGE_PCT / 100.0))
late_start_local = workday["start_local"] + timedelta(minutes=MANAGER_LATE_START_GRACE_MINUTES)
early_finish_local = workday["end_local"] - timedelta(minutes=MANAGER_EARLY_FINISH_GRACE_MINUTES)
roster = []
actions = []
active_users = 0
on_target_users = 0
below_target_users = 0
workday_total_active_seconds = 0
workday_first_values = []
workday_last_values = []
top_workday_user = ""
top_workday_seconds = -1
for row in rows:
alias = resolve_user_alias(row.get("user", ""), row.get("user_id", ""), host)
if alias["exclude"]:
continue
public_row = {key: value for key, value in row.items() if key != "_intervals"}
calendar_active_seconds = int(row.get("active_seconds", 0) or 0)
intervals = row.get("_intervals") or []
workday_active_seconds, workday_first, workday_last = _interval_overlap_seconds(
intervals,
workday["start_local"].astimezone(timezone.utc),
effective_end_local.astimezone(timezone.utc),
)
first_activity = parse_iso_utc(row.get("first_activity"))
last_activity = parse_iso_utc(row.get("last_activity"))
first_local = first_activity.astimezone(REPORT_TZ) if first_activity else None
last_local = last_activity.astimezone(REPORT_TZ) if last_activity else None
workday_first_local = workday_first.astimezone(REPORT_TZ) if workday_first else None
workday_last_local = workday_last.astimezone(REPORT_TZ) if workday_last else None
coverage_pct = _clamp_pct((workday_active_seconds / expected_seconds_per_user) * 100.0) if expected_seconds_per_user > 0 else 0.0
status = "ok"
if workday_active_seconds <= 0:
status = "inactive"
elif workday_active_seconds < target_seconds:
status = "below_target"
if workday_active_seconds > 0:
active_users += 1
workday_total_active_seconds += workday_active_seconds
if workday_first_local:
workday_first_values.append(workday_first_local.isoformat())
if workday_last_local:
workday_last_values.append(workday_last_local.isoformat())
if workday_active_seconds >= target_seconds and workday_active_seconds > 0:
on_target_users += 1
elif workday_active_seconds > 0:
below_target_users += 1
if workday_active_seconds > top_workday_seconds:
top_workday_seconds = workday_active_seconds
top_workday_user = alias["display_name"]
roster.append(
{
**public_row,
"user": alias["display_name"],
"user_original": row.get("user", ""),
"manager_owner": alias["manager_owner"],
"department": alias["department"],
"role": alias["role"],
"notes": alias["notes"],
"canonical_user_id": alias["canonical_user_id"] or row.get("user_id", ""),
"calendar_active_seconds": calendar_active_seconds,
"calendar_active_hhmm": row.get("active_hhmm", "00:00"),
"workday_active_seconds": workday_active_seconds,
"workday_active_hhmm": hhmm(workday_active_seconds),
"coverage_pct": coverage_pct,
"status": status,
"first_activity_local": first_local.isoformat() if first_local else "",
"last_activity_local": last_local.isoformat() if last_local else "",
"workday_first_activity_local": workday_first_local.isoformat() if workday_first_local else "",
"workday_last_activity_local": workday_last_local.isoformat() if workday_last_local else "",
}
)
owner = alias["display_name"]
user_id = alias["canonical_user_id"] or row.get("user_id", "")
evidence = {
"calendar_active_hhmm": row.get("active_hhmm", "00:00"),
"workday_active_hhmm": hhmm(workday_active_seconds),
"coverage_pct": coverage_pct,
"first_activity": row.get("first_activity", ""),
"last_activity": row.get("last_activity", ""),
"sessions_count": row.get("sessions_count", 0),
"manager_owner": alias["manager_owner"],
"department": alias["department"],
"role": alias["role"],
}
if workday_active_seconds <= 0:
actions.append(
_action(
"missing_activity",
"critical",
alias["manager_owner"],
"today",
f"За {report_date.isoformat()} у сотрудника {owner} нет подтверждённой активности в рабочем окне RDP.",
f"Проверить сотрудника {owner}: работал ли он в рабочее время, была ли потеря сбора данных или отсутствие входа в систему.",
user_id=user_id,
evidence=evidence,
)
)
continue
if expected_seconds_per_user > 0 and workday_active_seconds < low_seconds:
actions.append(
_action(
"low_activity_review",
"high",
alias["manager_owner"],
"24h",
f"У сотрудника {owner} активное время в рабочем окне {hhmm(workday_active_seconds)} ниже {MANAGER_LOW_COVERAGE_PCT}% от ожидаемого окна.",
f"Проверить загрузку сотрудника {owner}, задачи и фактическое присутствие в рабочем процессе.",
user_id=user_id,
evidence=evidence,
)
)
elif expected_seconds_per_user > 0 and workday_active_seconds < target_seconds:
actions.append(
_action(
"target_gap_review",
"medium",
alias["manager_owner"],
"24h",
f"У сотрудника {owner} активное время в рабочем окне {hhmm(workday_active_seconds)} ниже управленческого целевого порога {MANAGER_TARGET_COVERAGE_PCT}%.",
f"Уточнить причину отклонения по сотруднику {owner} и подтвердить план работ.",
user_id=user_id,
evidence=evidence,
)
)
if workday_first_local and workday_first_local > late_start_local:
actions.append(
_action(
"late_start_review",
"medium",
alias["manager_owner"],
"24h",
f"У сотрудника {owner} первая активность в рабочем окне зафиксирована поздно: {workday_first_local.strftime('%H:%M')}.",
f"Проверить причину позднего старта сотрудника {owner} и подтвердить, что это не проблема доступа или дисциплины.",
user_id=user_id,
evidence=evidence,
)
)
if (not is_today) and workday_last_local and workday_last_local < early_finish_local:
actions.append(
_action(
"early_finish_review",
"medium",
alias["manager_owner"],
"24h",
f"У сотрудника {owner} последняя активность в рабочем окне завершилась рано: {workday_last_local.strftime('%H:%M')}.",
f"Проверить, было ли досрочное завершение рабочего дня сотрудника {owner} согласовано и чем оно объясняется.",
user_id=user_id,
evidence=evidence,
)
)
actions.sort(key=lambda item: (_priority_rank(item["priority"]), item["owner"].lower(), item["action_id"]))
inactive_users = sum(1 for row in roster if row["status"] == "inactive")
portfolio_coverage_pct = _clamp_pct((workday_total_active_seconds / (expected_seconds_per_user * len(rows))) * 100.0) if rows and expected_seconds_per_user > 0 else 0.0
calendar_first = calendar_summary.get("first_activity", "")
calendar_last = calendar_summary.get("last_activity", "")
return {
"generated_at_utc": now_utc().isoformat().replace("+00:00", "Z"),
"host": resolve_host(host),
"report_date": report_date.isoformat(),
"report_timezone": str(REPORT_TZ),
"workday": {
"start_local": workday["start_local"].isoformat(),
"end_local": workday["end_local"].isoformat(),
"expected_seconds_per_user": expected_seconds_per_user,
"expected_hhmm_per_user": hhmm(expected_seconds_per_user),
"target_coverage_pct": MANAGER_TARGET_COVERAGE_PCT,
"low_coverage_pct": MANAGER_LOW_COVERAGE_PCT,
},
"summary": {
**calendar_summary,
"calendar_total_active_seconds": calendar_summary["total_active_seconds"],
"calendar_total_active_hhmm": calendar_summary["total_active_hhmm"],
"calendar_first_activity": calendar_first,
"calendar_last_activity": calendar_last,
"workday_total_active_seconds": workday_total_active_seconds,
"workday_total_active_hhmm": hhmm(workday_total_active_seconds),
"workday_first_activity": min(workday_first_values) if workday_first_values else "",
"workday_last_activity": max(workday_last_values) if workday_last_values else "",
"total_active_seconds": workday_total_active_seconds,
"total_active_hhmm": hhmm(workday_total_active_seconds),
"first_activity": min(workday_first_values) if workday_first_values else "",
"last_activity": max(workday_last_values) if workday_last_values else "",
"top_user": top_workday_user,
"top_user_active_hhmm": hhmm(top_workday_seconds if top_workday_seconds > 0 else 0),
"active_users": active_users,
"inactive_users": inactive_users,
"on_target_users": on_target_users,
"below_target_users": below_target_users,
"portfolio_coverage_pct": portfolio_coverage_pct,
"actions_count": len(actions),
"critical_actions_count": sum(1 for action in actions if action["priority"] == "critical"),
"high_actions_count": sum(1 for action in actions if action["priority"] == "high"),
},
"actions": actions,
"rows": roster,
"bucket_id": get_sessions_bucket_id(host),
"report_bounds": {
"start_utc": to_iso_utc(report_bounds["start"]),
"end_utc": to_iso_utc(report_bounds["end"]),
},
}
def build_management_trend(host, anchor_date):
trend = []
for offset in range(MANAGER_TREND_DAYS - 1, -1, -1):
current_date = anchor_date - timedelta(days=offset)
bounds, events = fetch_events_for_date(host, current_date)
rows = aggregate_rows_with_intervals(events, bounds["start"], bounds["end"], host)
payload = _build_management_core(rows, host, current_date)
summary = payload["summary"]
trend.append(
{
"report_date": current_date.isoformat(),
"users_count": summary["users_count"],
"active_users": summary["active_users"],
"inactive_users": summary["inactive_users"],
"workday_total_active_seconds": summary["workday_total_active_seconds"],
"workday_total_active_hhmm": summary["workday_total_active_hhmm"],
"portfolio_coverage_pct": summary["portfolio_coverage_pct"],
"actions_count": summary["actions_count"],
"critical_actions_count": summary["critical_actions_count"],
}
)
return trend
def build_management_payload(rows, host, report_date):
payload = _build_management_core(rows, host, report_date)
source_freshness, source_actions = build_source_freshness(resolve_host(host))
payload["sources"] = source_freshness
payload["trend"] = build_management_trend(resolve_host(host), report_date)
if source_actions:
payload["actions"].extend(source_actions)
payload["actions"].sort(key=lambda item: (_priority_rank(item["priority"]), item["owner"].lower(), item["action_id"]))
payload["summary"]["actions_count"] = len(payload["actions"])
payload["summary"]["critical_actions_count"] = sum(1 for action in payload["actions"] if action["priority"] == "critical")
payload["summary"]["high_actions_count"] = sum(1 for action in payload["actions"] if action["priority"] == "high")
payload["executive"] = build_executive_summary(payload["summary"], payload["actions"], payload["sources"])
return payload
def report_for_date(host, report_date):
bounds, events = fetch_events_for_date(host, report_date)
return aggregate_rows(events, bounds["start"], bounds["end"], host)
def management_report_for_date(host, report_date):
cached = load_management_cache(host, report_date)
if cached is not None:
return cached
bounds, events = fetch_events_for_date(host, report_date)
rows = aggregate_rows_with_intervals(events, bounds["start"], bounds["end"], host)
payload = build_management_payload(rows, host, report_date)
save_management_cache(host, report_date, payload)
return payload
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 = 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 ""
summary = build_report_summary(rows)
today_url = "/reports/worktime/today?" + urlencode({"format": "html", "host": resolve_host(host), "day": "today"})
yesterday_url = "/reports/worktime/today?" + urlencode({"format": "html", "host": resolve_host(host), "day": "yesterday"})
csv_url = "/reports/worktime/today?" + urlencode({"format": "csv", "host": resolve_host(host), **({"day": selected_day} if selected_day in {"today", "yesterday"} else {"date": date_local})})
json_url = "/reports/worktime/today?" + urlencode({"host": resolve_host(host), **({"day": selected_day} if selected_day in {"today", "yesterday"} else {"date": date_local})})
form_action = "/reports/worktime/today"
cards = [
("Пользователи", str(summary["users_count"])),
("Активное время", summary["total_active_hhmm"]),
("Лидер дня", f"{summary['top_user']} · {summary['top_user_active_hhmm']}" if summary["top_user"] else "н/д"),
("Диапазон", f"{summary['first_activity']} -> {summary['last_activity']}" if summary["first_activity"] else "нет активности"),
]
trs = []
detail_cards = []
for row in rows:
user_slug = safe_slug(row["user"])
active_seconds = int(row.get("active_seconds", 0) or 0)
utilization = 0.0
day_total = 24 * 3600
if day_total > 0:
utilization = round((active_seconds / day_total) * 100.0, 2)
trs.append(
""
f"| {html.escape(row['user'])} | "
f"{html.escape(row['user_id'])} | "
f"{row['active_hhmm']} | "
f"{row['active_seconds']} | "
f"{html.escape(row['first_activity'])} | "
f"{html.escape(row['last_activity'])} | "
f"{row['idle_seconds']} | "
f"{row['sessions_count']} | "
f"{row['samples_count']} | "
"
"
)
detail_cards.append(
""
""
"
{user}
"
"{active}"
""
""
"
Пользователь{user_id}
"
"
Загрузка{utilization}%
"
"
Начало активности{first_activity}
"
"
Конец активности{last_activity}
"
"
Сессии{sessions}
"
"
Активные сэмплы{active_samples} / {samples}
"
"
| За выбранную дату данных пока нет. |
')
detail_cards.append("За выбранную дату нет активности пользователей.
")
return f"""
AW-rus Отчёт по работе в RDP
Отчёт по работе в RDP
Хост: {resolve_host(host)} · Дата: {date_local} · Часовой пояс: {REPORT_TZ} · Сформировано UTC: {generated}
{''.join(f"
{html.escape(label)}{html.escape(value)}
" for label, value in cards)}
Таблица по пользователям
| Пользователь |
Учётная запись |
Активно |
Активно, сек |
Начало активности |
Конец активности |
Простой, сек |
Сессии |
Сэмплы |
{''.join(trs)}
Детали по пользователям
{''.join(detail_cards)}
"""
def render_management_html(payload, selected_day=None):
summary = payload["summary"]
workday = payload["workday"]
report_date = payload["report_date"]
host = payload["host"]
executive = payload.get("executive") or {}
today_url = "/reports/worktime/management?" + urlencode({"format": "html", "host": host, "day": "today"})
yesterday_url = "/reports/worktime/management?" + urlencode({"format": "html", "host": host, "day": "yesterday"})
json_url = "/reports/worktime/management?" + urlencode({"host": host, **({"day": selected_day} if selected_day in {"today", "yesterday"} else {"date": report_date})})
classic_url = "/reports/worktime/today?" + urlencode({"format": "html", "host": host, **({"day": selected_day} if selected_day in {"today", "yesterday"} else {"date": report_date})})
actions_html = []
for action in payload["actions"]:
actions_html.append(
""
f"| {html.escape(action['priority'])} | "
f"{html.escape(action['owner'])} | "
f"{html.escape(action['action_id'])} | "
f"{html.escape(action['deadline_hint'])} | "
f"{html.escape(action['reason'])} | "
f"{html.escape(action['recommended_action'])} | "
"
"
)
if not actions_html:
actions_html.append("| Отклонений по текущим правилам не найдено. |
")
roster_html = []
for row in payload["rows"]:
roster_html.append(
""
f"| {html.escape(row['user'])} | "
f"{html.escape(row.get('canonical_user_id') or row['user_id'])} | "
f"{html.escape(row.get('manager_owner') or row['user'])} | "
f"{html.escape(row.get('department') or '-')} | "
f"{html.escape(row['workday_active_hhmm'])} | "
f"{html.escape(row['calendar_active_hhmm'])} | "
f"{row['coverage_pct']} | "
f"{html.escape(row['status'])} | "
f"{html.escape(row.get('workday_first_activity_local') or '-')} | "
f"{html.escape(row.get('workday_last_activity_local') or '-')} | "
f"{row['sessions_count']} | "
"
"
)
if not roster_html:
roster_html.append("| За выбранную дату данных нет. |
")
trend_html = []
for row in payload.get("trend", []):
trend_html.append(
""
f"| {html.escape(row['report_date'])} | "
f"{row['users_count']} | "
f"{row['active_users']} | "
f"{row['inactive_users']} | "
f"{html.escape(row['workday_total_active_hhmm'])} | "
f"{row['portfolio_coverage_pct']} | "
f"{row['actions_count']} | "
f"{row['critical_actions_count']} | "
"
"
)
if not trend_html:
trend_html.append("| Тренд пока недоступен. |
")
sources_html = []
for source in payload.get("sources", []):
sources_html.append(
""
f"| {html.escape(source['label'])} | "
f"{html.escape(source['status_label'])} | "
f"{html.escape(source['bucket_id'])} | "
f"{html.escape(source.get('timestamp') or '-')} | "
f"{html.escape(str(source.get('age_seconds')) if source.get('age_seconds') is not None else '-')} | "
f"{html.escape(source.get('event_summary') or source.get('summary') or '-')} | "
"
"
)
if not sources_html:
sources_html.append("| Статусы источников недоступны. |
")
focus_html = []
for item in executive.get("focus_items", []):
focus_html.append(
""
f"{html.escape(item['priority'])}
"
f"{html.escape(item['title'])}
"
f"Ответственный: {html.escape(item['owner'])}
"
f"{html.escape(item['reason'])}
"
f"{html.escape(item['recommended_action'])}"
""
)
if not focus_html:
focus_html.append("Критичных действий нет
На текущий момент менеджерских отклонений по активным правилам не найдено.
")
stale_html = []
for item in executive.get("stale_sources", []):
stale_html.append(
""
f"{html.escape(item['label'])}: {html.escape(item['status'])} · {html.escape(item['summary'])}"
""
)
stale_block = (
"Проблемы со свежестью источников:"
+ "".join(stale_html)
+ "
"
) if stale_html else ""
cards = [
("Пользователи", str(summary["users_count"])),
("Активны", str(summary["active_users"])),
("Без активности", str(summary["inactive_users"])),
("Ниже цели", str(summary["below_target_users"])),
("Покрытие", f"{summary['portfolio_coverage_pct']}%"),
("Действия", str(summary["actions_count"])),
("Рабочее окно", summary["workday_total_active_hhmm"]),
("Календарный день", summary["calendar_total_active_hhmm"]),
]
return f"""
AW-rus Управленческий отчёт по работе в RDP
Управленческий отчёт по работе в RDP
Хост: {html.escape(host)} · Дата: {html.escape(report_date)} · Рабочее окно: {html.escape(workday['start_local'])} -> {html.escape(workday['end_local'])} · Сформировано UTC: {html.escape(payload['generated_at_utc'])}
{''.join(f"
{html.escape(label)}{html.escape(value)}
" for label, value in cards)}
Целевое покрытие: {MANAGER_TARGET_COVERAGE_PCT}% от ожидаемого рабочего окна на пользователя.
Критический провал: ниже {MANAGER_LOW_COVERAGE_PCT}% или полное отсутствие активности.
Рабочее окно считается отдельно от календарной активности, чтобы ночная работа не маскировала дневной провал.
Что делать сегодня
{html.escape(executive.get('headline') or 'Сводка недоступна')}
{html.escape(executive.get('message') or '')}
{stale_block}
{''.join(focus_html)}
Тренд за {MANAGER_TREND_DAYS} дней
| Дата |
Пользователи |
Активны |
Без активности |
Рабочее окно |
Покрытие, % |
Действия |
Critical |
{''.join(trend_html)}
Очередь действий руководителя
| Приоритет |
Сотрудник |
Тип |
Срок |
Почему это важно |
Что сделать |
{''.join(actions_html)}
Покрытие по сотрудникам
| Сотрудник |
Учётная запись |
Ответственный |
Подразделение |
Активно в окне |
Активно за день |
Покрытие, % |
Статус |
Первая активность в окне |
Последняя активность в окне |
Сессии |
{''.join(roster_html)}
Свежесть источников данных
| Источник |
Статус |
Бакет |
Последнее событие UTC |
Возраст, сек |
Контекст |
{''.join(sources_html)}
"""
class H(BaseHTTPRequestHandler):
def do_GET(self):
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()
return
path = os.path.join(IOC_DIR, name)
if not os.path.isfile(path):
self.send_response(404)
self.end_headers()
return
with open(path, "rb") as f:
data = f.read()
if name.endswith(".json"):
ctype = "application/json; charset=utf-8"
elif name.endswith(".csv"):
ctype = "text/csv; charset=utf-8"
else:
ctype = "text/plain; charset=utf-8"
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
return
if parsed.path not in {"/reports/worktime/today", "/reports/worktime/management"}:
self.send_response(404)
self.end_headers()
return
params = parse_qs(parsed.query, keep_blank_values=False)
fmt = "json"
if params.get("format", ["json"])[0] == "csv":
fmt = "csv"
elif params.get("format", ["json"])[0] == "html":
fmt = "html"
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)
is_management = parsed.path == "/reports/worktime/management"
management_payload = management_report_for_date(host, report_date) if is_management else None
rows = report_for_date_fresh(host, report_date) if not is_management else management_payload["rows"]
if fmt == "csv":
if is_management:
out = io.StringIO()
writer = csv.DictWriter(
out,
fieldnames=["priority", "owner", "user_id", "action_id", "deadline_hint", "reason", "recommended_action"],
extrasaction="ignore",
)
writer.writeheader()
writer.writerows(management_payload["actions"])
data = out.getvalue().encode()
self.send_response(200)
self.send_header("Content-Type", "text/csv; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
return
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()
writer.writerows(rows)
data = out.getvalue().encode()
self.send_response(200)
self.send_header("Content-Type", "text/csv; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
return
if fmt == "html":
if is_management:
data = render_management_html(management_payload, selected_day=day if day in {"today", "yesterday"} else None).encode("utf-8")
else:
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
if is_management:
obj = management_payload
data = json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; 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")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def main():
HTTPServer((LISTEN_HOST, LISTEN_PORT), H).serve_forever()
if __name__ == "__main__":
main()