Merge feature/logs-query: logs query flags, json_helper module split, handshake logging (v0.5.1)

- wgctl logs --since: relative (2h/7d) and EU/ISO date formats
- wgctl logs --service: filter by service name, IP, or IP:port
- wgctl logs --event: filter wg events by type
- wgctl logs: no header when no logs found
- core/lib/util.py: shared utilities, parse_since, reverse_lookup
- core/lib/events.py: fw_events, wg_events with query params
- core/lib/peers.py: peer_data, peer_transfer
- core/lib/activity.py: activity_aggregate
- wgctl-monitor.py: handshake session poller thread with cache

commands/logs.command.sh

@@ -18,6 +18,8 @@ function cmd::logs::on_load() {
   flag::register --days
   flag::register --raw
   flag::register --detailed
+  flag::register --service
+  flag::register --event
 }
 
 function cmd::logs::help() {
@@ -35,10 +37,15 @@ Options for show:
   --name <name>         Filter by client name
   --type <type>         Filter by device type
   --limit <n>           Max results per source (default: 50)
+  --since <time>        Show events since: 2h, 7d, 23/05, 23/05/2026, 2026-05-23
+  --service <svc>       Filter by service name, IP, or IP:port
+                        e.g. pihole, proxmox:web-ui, 10.0.0.100, 10.0.0.100:8006
+  --event <type>        Filter wg events: attempt | handshake
   --fw                  Show only firewall drops
   --wg                  Show only WireGuard events
   --merged              Show all events chronologically interleaved
-  --follow, -f      Follow logs in real time (alias: wgctl watch)
+  --detailed            Show all deduplicated events (bypass hourly collapse)
+  --follow, -f          Follow logs in real time
   --raw                 Show raw IPs without service annotation
 
 Options for remove:
@@ -55,8 +62,15 @@ Options for rotate:
 
 Examples:
   wgctl logs
-  wgctl logs --name phone-nuno
-  wgctl logs --fw --limit 100
+  wgctl logs --since 2h
+  wgctl logs --since 23/05
+  wgctl logs --name phone-nuno --since 7d
+  wgctl logs --fw --service pihole
+  wgctl logs --fw --service proxmox:web-ui
+  wgctl logs --fw --service 10.0.0.100
+  wgctl logs --wg --event attempt
+  wgctl logs --wg --event handshake --since 24h
+  wgctl logs --detailed
   wgctl logs --merged
   wgctl logs --follow
   wgctl logs remove --name phone-nuno
@@ -86,21 +100,25 @@ function cmd::logs::run() {
 }
 
 function cmd::logs::show() {
-  local name="" type="" limit=50
-  local fw_only=false wg_only=false follow=false merged=false raw=false detailed=false
-  
+  local name="" type="" limit=50 since=""
+  local fw_only=false wg_only=false follow=false merged=false
+  local raw=false detailed=false
+  local filter_service="" filter_event=""
 
   while [[ $# -gt 0 ]]; do
     case "$1" in
       --name)      name="$2";            shift 2 ;;
       --type)      type="$2";            shift 2 ;;
       --limit)     limit="$2";           shift 2 ;;
+      --since)     since="$2";           shift 2 ;;
+      --service)   filter_service="$2";  shift 2 ;;
+      --event)     filter_event="$2";    shift 2 ;;
       --fw)        fw_only=true;         shift   ;;
       --wg)        wg_only=true;         shift   ;;
       --merged)    merged=true;          shift   ;;
       --follow|-f) follow=true;          shift   ;;
       --raw)       raw=true;             shift   ;;
-      --detailed)  detailed=true    shift   ;;
+      --detailed)  detailed=true;        shift   ;;
       --help)      cmd::logs::help;      return  ;;
       *)
         log::error "Unknown flag: $1"
@@ -112,7 +130,6 @@ function cmd::logs::show() {
   local collapse=1
   $detailed && collapse=0
 
-
   if [[ -n "$name" && -n "$type" ]]; then
     name=$(peers::resolve_and_require "$name" "$type") || return 1
   fi
@@ -131,27 +148,71 @@ function cmd::logs::show() {
   local net_file=""
   $raw || net_file="$(ctx::net)"
 
-  log::section "WireGuard Activity Log"
-  printf "\n"
+  # Parse --service into dest_ip and dest_port
+  local filter_dest_ip="" filter_dest_port=""
+  if [[ -n "$filter_service" ]]; then
+    if [[ "$filter_service" =~ ^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+(:[0-9]+)?$ ]]; then
+      filter_dest_ip="${filter_service%%:*}"
+      local maybe_port="${filter_service##*:}"
+      [[ "$maybe_port" != "$filter_dest_ip" ]] && filter_dest_port="$maybe_port"
+    else
+      local svc_resolved
+      svc_resolved=$(net::resolve "$filter_service" 2>/dev/null | head -1)
+      if [[ -n "$svc_resolved" ]]; then
+        filter_dest_ip="${svc_resolved%%:*}"
+        local rest="${svc_resolved#*:}"
+        [[ "$rest" != "$filter_dest_ip" ]] && filter_dest_port="${rest%%:*}"
+      else
+        log::error "Service not found: ${filter_service}"
+        return 1
+      fi
+    fi
+  fi
 
   if $merged; then
-    cmd::logs::show_merged "$filter_ip" "$name" "$type" "$limit" "$net_file"
+    log::section "WireGuard Activity Log"
+    printf "\n"
+    cmd::logs::show_merged "$filter_ip" "$name" "$type" "$limit" "$net_file" "$since"
     return
   fi
 
-  $wg_only || cmd::logs::show_fw_events "$filter_ip" "$name" "$type" "$limit" "$net_file" "$collapse"
-  $fw_only || cmd::logs::show_wg_events "$filter_ip" "$name" "$type" "$limit" "$collapse"
+  # Collect output — only show header if there's data
+  local fw_output="" wg_output=""
+
+  $wg_only || fw_output=$(cmd::logs::show_fw_events \
+    "$filter_ip" "$name" "$type" "$limit" "$net_file" \
+    "$collapse" "$since" "$filter_dest_ip" "$filter_dest_port")
+
+  $fw_only || wg_output=$(cmd::logs::show_wg_events \
+    "$filter_ip" "$name" "$type" "$limit" \
+    "$collapse" "$since" "$filter_event")
+
+  if [[ -z "${fw_output// /}" && -z "${wg_output// /}" ]]; then
+    log::wg_warning "No logs found"
+    return 0
+  fi
+
+  log::section "WireGuard Activity Log"
+  printf "\n"
+  
+  if [[ -n "$fw_output" ]]; then printf "%s\n" "$fw_output"; fi;
+  if [[ -n "$wg_output" ]]; then printf "%s\n" "$wg_output"; fi;
 }
 
 function cmd::logs::show_fw_events() {
   local filter_ip="${1:-}" filter_name="${2:-}" filter_type="${3:-}" \
-        limit="${4:-50}" net_file="${5:-}" collapse="${6:-1}"
+        limit="${4:-50}" net_file="${5:-}" collapse="${6:-1}" \
+        since="${7:-}" filter_dest_ip="${8:-}" filter_dest_port="${9:-}"
 
   [[ ! -f "$FW_EVENTS_LOG" ]] && return 0
 
   local data
-  data=$(json::fw_events "$FW_EVENTS_LOG" "$filter_ip" "$filter_type" \
-    "$(ctx::clients)" "${net_file:-}" "$limit" "$collapse" 2>/dev/null)
+  data=$(json::fw_events \
+    "$FW_EVENTS_LOG" "$filter_ip" "$filter_type" \
+    "$(ctx::clients)" "${net_file:-}" \
+    "$limit" "$collapse" "$since" \
+    "$filter_dest_ip" "$filter_dest_port" \
+    2>/dev/null)
 
   [[ -z "$data" ]] && return 0
 
@@ -160,8 +221,7 @@ function cmd::logs::show_fw_events() {
   while IFS='|' read -r ts client dest_ip dest_port proto svc count; do
     [[ -z "$ts" ]] && continue
     (( ${#client} > w_client )) && w_client=${#client}
-    local dest_display
-    local host_name
+    local dest_display host_name
     host_name=$(hosts::resolve_ip "$dest_ip")
     if [[ -n "$host_name" ]]; then
       dest_display="$host_name"
@@ -186,12 +246,16 @@ function cmd::logs::show_fw_events() {
 
 function cmd::logs::show_wg_events() {
   local filter_ip="${1:-}" filter_name="${2:-}" filter_type="${3:-}" \
-        limit="${4:-50}" collapse="${5:-1}"
+        limit="${4:-50}" collapse="${5:-1}" \
+        since="${6:-}" filter_event="${7:-}"
 
   [[ ! -f "$WG_EVENTS_LOG" ]] && return 0
 
   local data
-  data=$(json::wg_events "$WG_EVENTS_LOG" "$filter_name" "$filter_type" "$limit" "$collapse" 2>/dev/null)
+  data=$(json::wg_events \
+    "$WG_EVENTS_LOG" "$filter_name" "$filter_type" \
+    "$limit" "$collapse" "$since" "$filter_event" \
+    2>/dev/null)
 
   [[ -z "$data" ]] && return 0
 
@@ -219,18 +283,21 @@ function cmd::logs::show_wg_events() {
   printf "\n"
 }
 
-
 function cmd::logs::show_merged() {
   local filter_ip="${1:-}" filter_name="${2:-}" filter_type="${3:-}" \
-        limit="${4:-50}" net_file="${5:-}"
+        limit="${4:-50}" net_file="${5:-}" since="${6:-}"
 
   local fw_data wg_data
-  fw_data=$(json::fw_events "$FW_EVENTS_LOG" "$filter_ip" "$filter_type" \
-    "$(ctx::clients)" "${net_file:-}" "$limit" 2>/dev/null)
-  wg_data=$(json::wg_events "$WG_EVENTS_LOG" "$filter_name" "$filter_type" \
-    "$limit" 2>/dev/null)
+  fw_data=$(json::fw_events \
+    "$FW_EVENTS_LOG" "$filter_ip" "$filter_type" \
+    "$(ctx::clients)" "${net_file:-}" \
+    "$limit" "1" "$since" "" "" \
+    2>/dev/null)
+  wg_data=$(json::wg_events \
+    "$WG_EVENTS_LOG" "$filter_name" "$filter_type" \
+    "$limit" "1" "$since" "" \
+    2>/dev/null)
 
-  # Measure widths across both sources
   local w_client=16 w_dest=20
   while IFS='|' read -r ts client rest; do
     [[ -z "$ts" ]] && continue
@@ -238,7 +305,6 @@ function cmd::logs::show_merged() {
   done < <(echo "$fw_data"; echo "$wg_data")
   (( w_client += 2 ))
 
-  # Tag and merge: prefix fw lines with "fw|", wg lines with "wg|"
   local merged_data
   merged_data=$(
     while IFS='|' read -r ts client dest_ip dest_port proto svc count; do
@@ -251,7 +317,6 @@ function cmd::logs::show_merged() {
     done <<< "$wg_data"
   )
 
-  # Sort by timestamp field 2
   while IFS='|' read -r source ts rest; do
     [[ -z "$source" ]] && continue
     case "$source" in
@@ -296,7 +361,6 @@ function cmd::logs::follow() {
   log::section "WireGuard Live Log (Ctrl+C to stop)"
   printf "\n"
 
-  # Delegate to watch command
   local watch_args=()
   [[ -n "$filter_name" ]] && watch_args+=(--name "$filter_name")
   [[ -n "$filter_type" ]] && watch_args+=(--type "$filter_type")

commands/test/integration.sh

@@ -187,8 +187,8 @@ function cmd::test::section_logs() {
   test::section "Logs"
   cmd::test::run_cmd "logs"                   "Activity" logs
   cmd::test::run_cmd "logs --name phone-nuno" "Activity" logs --name phone-nuno
-  cmd::test::run_cmd "logs --fw"              "Activity" logs --fw
-  cmd::test::run_cmd "logs --wg"              "Activity" logs --wg
+  cmd::test::run_cmd "logs --fw"              "Firewall Drops" logs --fw
+  cmd::test::run_cmd "logs --wg"              "WireGuard Events" logs --wg
 }
 
 function cmd::test::section_fw() {

core/lib/activity.py

@@ -0,0 +1,129 @@
+"""
+activity.py — activity aggregation for wgctl activity command.
+"""
+
+import os
+import json
+import glob
+import subprocess
+from collections import defaultdict
+from datetime import datetime, timezone, timedelta
+
+from lib.util import (
+    PROTO_MAP, build_ip_to_name, build_pubkey_to_name,
+    load_net_data, load_hosts_data,
+    reverse_lookup, resolve_display, make_dest_display,
+    ts_to_unix, parse_since,
+)
+
+
+def activity_aggregate(fw_file, wg_file, wg_interface, net_file,
+                       clients_dir, meta_dir, hours, filter_peer,
+                       filter_service_ip):
+    """
+    Aggregate activity data for wgctl activity.
+    Output:
+      peer|name|rx_bytes|tx_bytes|drop_count
+      service|peer_name|dest_display|drop_count
+    """
+    hours   = int(hours) if hours else 24
+    cutoff  = None
+    if hours > 0:
+        cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
+
+    # Preload lookups once
+    ip_to_peer     = build_ip_to_name(clients_dir)
+    pubkey_to_peer = build_pubkey_to_name(clients_dir)
+    net_data       = load_net_data(net_file)
+
+    def _reverse(dest_ip, dest_port, proto):
+        return reverse_lookup(net_data, dest_ip, dest_port, proto)
+
+    # WireGuard transfer totals
+    peer_rx = defaultdict(int)
+    peer_tx = defaultdict(int)
+    try:
+        result = subprocess.run(
+            ['wg', 'show', wg_interface, 'transfer'],
+            capture_output=True, text=True
+        )
+        for line in result.stdout.strip().splitlines():
+            parts = line.split()
+            if len(parts) >= 3:
+                pubkey, rx, tx = parts[0], int(parts[1]), int(parts[2])
+                peer = pubkey_to_peer.get(pubkey)
+                if peer:
+                    peer_rx[peer] += rx
+                    peer_tx[peer] += tx
+    except Exception:
+        pass
+
+    # Parse fw_events for drops
+    peer_drops    = defaultdict(int)
+    service_drops = defaultdict(lambda: defaultdict(int))
+
+    if os.path.exists(fw_file):
+        try:
+            with open(fw_file) as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        ev = json.loads(line)
+                        if cutoff:
+                            ts_str = ev.get('timestamp', '')
+                            try:
+                                ts = datetime.fromisoformat(ts_str)
+                                if ts.tzinfo is None:
+                                    ts = ts.replace(tzinfo=timezone.utc)
+                                if ts < cutoff:
+                                    continue
+                            except Exception:
+                                pass
+
+                        src_ip    = ev.get('src_ip', '')
+                        if not src_ip:
+                            continue
+
+                        dest_ip   = ev.get('dest_ip', '')
+                        dest_port = str(ev.get('dest_port', ''))
+                        proto_num = ev.get('ip.protocol', 0)
+                        proto     = PROTO_MAP.get(int(proto_num), str(proto_num))
+
+                        peer = ip_to_peer.get(src_ip)
+                        if not peer:
+                            continue
+                        if filter_peer and peer != filter_peer:
+                            continue
+                        if filter_service_ip and dest_ip != filter_service_ip:
+                            continue
+
+                        svc_name     = _reverse(dest_ip, dest_port, proto)
+                        dest_display = make_dest_display(dest_ip, dest_port, proto, svc_name)
+
+                        peer_drops[peer] += 1
+                        service_drops[peer][dest_display] += 1
+
+                    except Exception:
+                        continue
+        except Exception:
+            pass
+
+    # Collect peers with any activity
+    all_peers = set()
+    all_peers.update(k for k in peer_rx if peer_rx[k] > 0)
+    all_peers.update(k for k in peer_tx if peer_tx[k] > 0)
+    all_peers.update(peer_drops.keys())
+    if filter_peer:
+        all_peers = {p for p in all_peers if p == filter_peer}
+
+    for peer in sorted(all_peers):
+        rx    = peer_rx.get(peer, 0)
+        tx    = peer_tx.get(peer, 0)
+        drops = peer_drops.get(peer, 0)
+        print(f"peer|{peer}|{rx}|{tx}|{drops}")
+
+        svc_map = service_drops.get(peer, {})
+        for dest_display, count in sorted(svc_map.items(), key=lambda x: -x[1]):
+            print(f"service|{peer}|{dest_display}|{count}")

core/lib/events.py

@@ -0,0 +1,609 @@
+"""
+events.py — WireGuard and firewall event processing.
+"""
+
+import os
+import json
+import sys
+from collections import defaultdict
+from datetime import datetime
+
+from lib.util import (
+    DATETIME_FMT, PROTO_MAP,
+    build_ip_to_name, load_net_data, load_hosts_data,
+    reverse_lookup, hosts_lookup, resolve_display,
+    fmt_ts, fmt_ts_hour, ts_to_unix, parse_since,
+    make_dest_display,
+)
+
+
+# ──────────────────────────────────────────
+# fw_events
+# ──────────────────────────────────────────
+
+def fw_events(file, filter_ip, filter_type, clients_dir, net_file,
+              limit, collapse='1', since='', filter_dest_ip='', filter_dest_port=''):
+    """
+    Format firewall drop events with dedup, counts, and service annotation.
+
+    collapse='1' (default): hourly aggregation
+    collapse='0': show all deduplicated events (--detailed mode)
+    since: relative or absolute time string (e.g. '2h', '23/05', '2026-05-23')
+    filter_dest_ip: filter by destination IP (optional)
+    filter_dest_port: filter by destination port (optional)
+
+    Output per line: ts|client|dest_ip|dest_port|proto|service_name|count
+    """
+    do_collapse = str(collapse) != '0'
+    limit       = int(limit) if limit else 50
+
+    # Preload lookups once
+    ip_to_name = build_ip_to_name(clients_dir)
+    net_data   = load_net_data(net_file)
+    hosts_data = load_hosts_data(None)  # hosts lookup done in bash for now
+
+    since_dt = parse_since(since) if since else None
+
+    def _reverse(dest_ip, dest_port, proto):
+        return reverse_lookup(net_data, dest_ip, dest_port, proto)
+
+    # ── Parse and first-pass dedup (time-window per key) ──
+    events   = []
+    last_seen = {}
+
+    try:
+        with open(file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    src = e.get('src_ip', '')
+                    if not src:
+                        continue
+                    if filter_ip and src != filter_ip:
+                        continue
+
+                    proto_num = int(e.get('ip.protocol', 0))
+                    proto     = PROTO_MAP.get(proto_num, str(proto_num))
+                    dst       = e.get('dest_ip', '')
+                    port      = str(e.get('dest_port', ''))
+
+                    if filter_dest_ip and dst != filter_dest_ip:
+                        continue
+                    if filter_dest_port and port != filter_dest_port:
+                        continue
+
+                    ts_str = e.get('timestamp', '')
+                    ts     = ts_to_unix(ts_str)
+
+                    if since_dt:
+                        try:
+                            ev_dt = datetime.fromisoformat(ts_str)
+                            if ev_dt.tzinfo is None:
+                                from datetime import timezone
+                                ev_dt = ev_dt.replace(tzinfo=timezone.utc)
+                            if ev_dt < since_dt:
+                                continue
+                        except Exception:
+                            pass
+
+                    key    = (src, dst, port, proto_num)
+                    windows = {1: 5, 6: 30, 17: 10}
+                    window  = windows.get(proto_num, 10)
+                    if key in last_seen and (ts - last_seen[key]) < window:
+                        continue
+                    last_seen[key] = ts
+                    events.append(e)
+                except Exception:
+                    continue
+    except Exception:
+        pass
+
+    # ── Collapse or detailed output ──
+    if do_collapse:
+        hourly    = defaultdict(int)
+        hourly_ts = {}
+
+        for e in events:
+            src       = e.get('src_ip', '')
+            dst       = e.get('dest_ip', '')
+            port      = str(e.get('dest_port', ''))
+            proto_num = int(e.get('ip.protocol', 0))
+            proto     = PROTO_MAP.get(proto_num, str(proto_num))
+            ts_str    = e.get('timestamp', '')
+            client    = ip_to_name.get(src, src)
+            svc_name  = _reverse(dst, port, proto)
+
+            try:
+                dt       = datetime.fromisoformat(ts_str)
+                hour_key = (client, dst, port, proto, svc_name,
+                            dt.strftime('%Y-%m-%d %H'))
+                hourly[hour_key] += 1
+                if hour_key not in hourly_ts:
+                    hourly_ts[hour_key] = dt
+            except Exception:
+                continue
+
+        sorted_buckets = sorted(hourly_ts.items(), key=lambda x: x[1])
+        for hour_key, dt in sorted_buckets[-limit:]:
+            client, dst, port, proto, svc_name, _ = hour_key
+            count  = hourly[hour_key]
+            ts_fmt = fmt_ts_hour(dt.isoformat())
+            print(f"{ts_fmt}|{client}|{dst}|{port}|{proto}|{svc_name}|{count}")
+
+    else:
+        # Detailed — consecutive dedup only
+        deduped = []
+        counts  = []
+        for e in events:
+            src       = e.get('src_ip', '')
+            dst       = e.get('dest_ip', '')
+            port      = str(e.get('dest_port', ''))
+            proto_num = int(e.get('ip.protocol', 0))
+            key       = (src, dst, port, proto_num)
+
+            ts = ts_to_unix(e.get('timestamp', ''))
+
+            if deduped:
+                prev     = deduped[-1]
+                prev_ts  = ts_to_unix(prev.get('timestamp', ''))
+                prev_key = (
+                    prev.get('src_ip', ''),
+                    prev.get('dest_ip', ''),
+                    str(prev.get('dest_port', '')),
+                    int(prev.get('ip.protocol', 0))
+                )
+                if key == prev_key and (ts - prev_ts) < 300:
+                    counts[-1] += 1
+                    continue
+
+            deduped.append(e)
+            counts.append(1)
+
+        for e, count in list(zip(deduped, counts))[-limit:]:
+            src       = e.get('src_ip', '')
+            dst       = e.get('dest_ip', '')
+            port      = str(e.get('dest_port', ''))
+            proto_num = int(e.get('ip.protocol', 0))
+            proto     = PROTO_MAP.get(proto_num, str(proto_num))
+            client    = ip_to_name.get(src, src)
+            svc_name  = _reverse(dst, port, proto)
+            ts_fmt    = fmt_ts(e.get('timestamp', ''))
+            print(f"{ts_fmt}|{client}|{dst}|{port}|{proto}|{svc_name}|{count}")
+
+
+# ──────────────────────────────────────────
+# wg_events
+# ──────────────────────────────────────────
+
+def wg_events(file, filter_client, filter_type, limit, collapse='1',
+              since='', filter_event=''):
+    """
+    Format WireGuard events with dedup and counts.
+
+    collapse='1' (default): hourly aggregation for attempt events
+    collapse='0': show all deduplicated events (--detailed mode)
+    since: relative or absolute time string
+    filter_event: 'attempt' | 'handshake' | '' (all)
+
+    Output per line: ts|client|endpoint|event|count
+    """
+    do_collapse = str(collapse) != '0'
+    limit       = int(limit) if limit else 50
+    since_dt    = parse_since(since) if since else None
+
+    events = []
+    try:
+        with open(file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    client = e.get('client', '')
+                    if not client:
+                        continue
+                    if filter_client and client != filter_client:
+                        continue
+                    if filter_type and not client.startswith(filter_type + '-'):
+                        continue
+                    if filter_event and e.get('event', '') != filter_event:
+                        continue
+
+                    if since_dt:
+                        ts_str = e.get('timestamp', '')
+                        try:
+                            ev_dt = datetime.fromisoformat(ts_str)
+                            if ev_dt.tzinfo is None:
+                                from datetime import timezone
+                                ev_dt = ev_dt.replace(tzinfo=timezone.utc)
+                            if ev_dt < since_dt:
+                                continue
+                        except Exception:
+                            pass
+
+                    events.append(e)
+                except Exception:
+                    pass
+    except Exception:
+        pass
+
+    if do_collapse:
+        hourly_attempts  = defaultdict(int)
+        hourly_ts        = {}
+        handshakes       = []
+        handshake_counts = []
+
+        for e in events:
+            ts_str   = e.get('timestamp', '')
+            client   = e.get('client', '')
+            endpoint = e.get('endpoint', '')
+            event    = e.get('event', '')
+            ts       = ts_to_unix(ts_str)
+
+            try:
+                dt = datetime.fromisoformat(ts_str)
+            except Exception:
+                dt = None
+
+            if event == 'attempt':
+                if dt:
+                    hour_key = (client, endpoint, event,
+                                dt.strftime('%Y-%m-%d %H'))
+                    hourly_attempts[hour_key] += 1
+                    if hour_key not in hourly_ts:
+                        hourly_ts[hour_key] = dt
+            else:
+                key = (client, event, endpoint[:15])
+                if handshakes:
+                    prev     = handshakes[-1]
+                    prev_ts  = ts_to_unix(prev.get('timestamp', ''))
+                    prev_key = (
+                        prev.get('client', ''),
+                        prev.get('event', ''),
+                        prev.get('endpoint', '')[:15]
+                    )
+                    if key == prev_key and (ts - prev_ts) < 300:
+                        handshake_counts[-1] += 1
+                        continue
+                handshakes.append(e)
+                handshake_counts.append(1)
+
+        output = []
+        for hour_key, dt in hourly_ts.items():
+            client, endpoint, event, _ = hour_key
+            count  = hourly_attempts[hour_key]
+            ts_fmt = fmt_ts_hour(dt.isoformat())
+            output.append((dt.timestamp(), f"{ts_fmt}|{client}|{endpoint}|{event}|{count}"))
+
+        for e, count in zip(handshakes, handshake_counts):
+            ts_str   = e.get('timestamp', '')
+            client   = e.get('client', '')
+            endpoint = e.get('endpoint', '')
+            event    = e.get('event', '')
+            ts       = ts_to_unix(ts_str)
+            ts_fmt   = fmt_ts(ts_str)
+            output.append((ts, f"{ts_fmt}|{client}|{endpoint}|{event}|{count}"))
+
+        output.sort(key=lambda x: x[0])
+        for _, line in output[-limit:]:
+            print(line)
+
+    else:
+        deduped = []
+        counts  = []
+        for e in events:
+            client   = e.get('client', '')
+            event    = e.get('event', '')
+            endpoint = e.get('endpoint', '')
+            key      = (client, event, endpoint[:15])
+            ts       = ts_to_unix(e.get('timestamp', ''))
+
+            if deduped:
+                prev     = deduped[-1]
+                prev_ts  = ts_to_unix(prev.get('timestamp', ''))
+                prev_key = (
+                    prev.get('client', ''),
+                    prev.get('event', ''),
+                    prev.get('endpoint', '')[:15]
+                )
+                if key == prev_key and (ts - prev_ts) < 300:
+                    counts[-1] += 1
+                    continue
+
+            deduped.append(e)
+            counts.append(1)
+
+        for e, count in list(zip(deduped, counts))[-limit:]:
+            ts_fmt   = fmt_ts(e.get('timestamp', ''))
+            client   = e.get('client', '')
+            endpoint = e.get('endpoint', '')
+            event    = e.get('event', '')
+            print(f"{ts_fmt}|{client}|{endpoint}|{event}|{count}")
+
+
+# ──────────────────────────────────────────
+# Single event parsers (used by watch)
+# ──────────────────────────────────────────
+
+def parse_event(line):
+    """Parse a single JSON wg event line."""
+    try:
+        e = json.loads(line)
+        print(f"{e.get('timestamp','')}|{e.get('client','')}|"
+              f"{e.get('endpoint','')}|{e.get('event','')}")
+    except Exception:
+        pass
+
+
+def parse_fw_event(line):
+    """Parse a single fw_events.log JSON line."""
+    try:
+        e = json.loads(line)
+        proto_num = e.get('ip.protocol', 0)
+        proto     = PROTO_MAP.get(proto_num, str(proto_num))
+        print(f"{e.get('timestamp','')}|{e.get('src_ip','')}|"
+              f"{e.get('dest_ip','')}|{e.get('dest_port','')}|{proto}")
+    except Exception:
+        pass
+
+
+def format_fw_event(line, clients_dir):
+    """Format a single fw_event line for display."""
+    ip_to_name = build_ip_to_name(clients_dir)
+    try:
+        e         = json.loads(line.strip())
+        src       = e.get('src_ip', '')
+        if not src:
+            return None
+        dst       = e.get('dest_ip', '—')
+        port      = e.get('dest_port', '')
+        proto_num = e.get('ip.protocol', 0)
+        proto     = PROTO_MAP.get(proto_num, str(proto_num))
+        dst_str   = f"{dst}:{port}" if port else dst
+        client    = ip_to_name.get(src, src)
+        ts_fmt    = fmt_ts(e.get('timestamp', ''))
+        return f"{ts_fmt}|{client}|{dst_str}|{proto}"
+    except Exception:
+        return None
+
+
+def format_wg_event(line):
+    """Format a single wg_event line for display."""
+    try:
+        e      = json.loads(line.strip())
+        client = e.get('client', '')
+        if not client:
+            return None
+        ts_fmt   = fmt_ts(e.get('timestamp', ''))
+        endpoint = e.get('endpoint', '—')
+        event    = e.get('event', '—')
+        return f"{ts_fmt}|{client}|{endpoint}|{event}|wg"
+    except Exception:
+        return None
+
+
+# ──────────────────────────────────────────
+# Event removal
+# ──────────────────────────────────────────
+
+def remove_events(file, identifier):
+    """Remove all events for a client/ip from a JSONL file."""
+    try:
+        lines = []
+        with open(file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    if (e.get('client') == identifier or
+                            e.get('src_ip') == identifier):
+                        continue
+                    lines.append(line)
+                except Exception:
+                    lines.append(line)
+        with open(file, 'w') as f:
+            f.writelines(lines)
+    except Exception as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+def remove_events_filtered(wg_file, fw_file, filter_name, filter_ip,
+                            filter_fw, filter_wg, before_days):
+    """Remove events with filters: by name/ip, source, or age."""
+    import time
+
+    cutoff_ts = None
+    if before_days:
+        cutoff_ts = time.time() - (float(before_days) * 86400)
+
+    def should_remove_wg(e):
+        if filter_name and e.get('client') != filter_name:
+            return False
+        if cutoff_ts:
+            try:
+                return ts_to_unix(e.get('timestamp', '')) < cutoff_ts
+            except Exception:
+                return False
+        return True
+
+    def should_remove_fw(e):
+        if filter_ip and e.get('src_ip') != filter_ip:
+            return False
+        if cutoff_ts:
+            try:
+                return ts_to_unix(e.get('timestamp', '')) < cutoff_ts
+            except Exception:
+                return False
+        return True
+
+    removed_wg = removed_fw = 0
+
+    if not filter_fw and os.path.exists(wg_file):
+        lines = []
+        with open(wg_file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    if should_remove_wg(e):
+                        removed_wg += 1
+                        continue
+                except Exception:
+                    pass
+                lines.append(line)
+        with open(wg_file, 'w') as f:
+            f.writelines(lines)
+
+    if not filter_wg and os.path.exists(fw_file):
+        lines = []
+        with open(fw_file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    if should_remove_fw(e):
+                        removed_fw += 1
+                        continue
+                except Exception:
+                    pass
+                lines.append(line)
+        with open(fw_file, 'w') as f:
+            f.writelines(lines)
+
+    print(f"{removed_wg}|{removed_fw}")
+
+
+# ──────────────────────────────────────────
+# Log follower (used by old follow_logs)
+# ──────────────────────────────────────────
+
+def follow_logs(fw_file, wg_file, filter_ip, filter_type,
+                clients_dir, filter_peers=''):
+    """Follow both log files and output formatted events."""
+    import time
+    peer_filter = set(filter_peers.split(',')) if filter_peers else set()
+    ip_to_name  = build_ip_to_name(clients_dir)
+
+    files = {}
+    for label, path in [('fw', fw_file), ('wg', wg_file)]:
+        if path and os.path.exists(path):
+            f = open(path)
+            f.seek(0, 2)
+            files[label] = f
+
+    dedup = {}
+
+    try:
+        while True:
+            for label, f in files.items():
+                line = f.readline()
+                if not line:
+                    continue
+                try:
+                    e = json.loads(line.strip())
+                except Exception:
+                    continue
+
+                if label == 'fw':
+                    src = e.get('src_ip', '')
+                    if not src:
+                        continue
+                    if filter_ip and src != filter_ip:
+                        continue
+                    if peer_filter:
+                        client_name = ip_to_name.get(src, '')
+                        if client_name not in peer_filter:
+                            continue
+
+                    dst       = e.get('dest_ip', '—')
+                    port      = e.get('dest_port', '')
+                    proto_num = e.get('ip.protocol', 0)
+                    proto     = PROTO_MAP.get(proto_num, str(proto_num))
+
+                    key    = (src, dst, port, proto_num)
+                    windows = {1: 5, 6: 30, 17: 10}
+                    window  = windows.get(proto_num, 10)
+                    now    = time.time()
+                    if key in dedup and (now - dedup[key]) < window:
+                        continue
+                    dedup[key] = now
+
+                    client = ip_to_name.get(src, src)
+                    if filter_type and not client.startswith(filter_type + '-'):
+                        continue
+                    dst_str = f"{dst}:{port}" if port else dst
+                    ts      = e.get('timestamp', '')[:16].replace('T', ' ')
+                    print(f"fw|{ts}|{client}|{dst_str}|{proto}", flush=True)
+
+                elif label == 'wg':
+                    client = e.get('client', '')
+                    if not client:
+                        continue
+                    if filter_ip:
+                        ip = ip_to_name.get(filter_ip, '')
+                        if client != ip and client != filter_ip:
+                            continue
+                    if peer_filter and client not in peer_filter:
+                        continue
+                    if filter_type and not client.startswith(filter_type + '-'):
+                        continue
+                    ts       = e.get('timestamp', '')[:16].replace('T', ' ')
+                    endpoint = e.get('endpoint', '—')
+                    event    = e.get('event', '—')
+                    print(f"wg|{ts}|{client}|{endpoint}|{event}", flush=True)
+
+            time.sleep(0.1)
+    except KeyboardInterrupt:
+        pass
+
+
+# ──────────────────────────────────────────
+# Misc
+# ──────────────────────────────────────────
+
+def last_event(file, key, field, client):
+    """Get last event field for a client."""
+    try:
+        last = None
+        with open(file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    if e.get(key) == client:
+                        last = e
+                except Exception:
+                    pass
+        if last:
+            print(last.get(field, ''))
+    except Exception:
+        pass
+
+
+def events_for(file, ip, limit):
+    """Format events for a given IP."""
+    try:
+        events = []
+        with open(file) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    if e.get('ip') == ip:
+                        events.append(e)
+                except Exception:
+                    pass
+        for e in events[-int(limit):]:
+            ts_fmt   = fmt_ts(e.get('timestamp', ''))
+            endpoint = e.get('endpoint', '—')
+            client   = e.get('client', '—')
+            event    = e.get('event', '—')
+            print(f'  {ts_fmt}  {client:<20} {endpoint:<20} {event}')
+    except Exception:
+        pass
+
+
+def iso_to_ts(iso_str):
+    """Convert ISO timestamp to unix timestamp."""
+    try:
+        from datetime import timezone
+        dt = datetime.fromisoformat(iso_str)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        print(int(dt.timestamp()))
+    except Exception:
+        print(0)

core/lib/peers.py

@@ -0,0 +1,180 @@
+"""
+peers.py — peer data, transfer stats, group lookups.
+"""
+
+import os
+import json
+import sys
+import glob
+
+from lib.util import DATETIME_FMT, build_ip_to_name, build_pubkey_to_name, fmt_ts
+
+
+def peer_data(clients_dir, meta_dir, events_log):
+    """
+    Output: name|ip|rule|type|last_ts|last_evt|main_group
+    """
+    meta = {}
+    for f in glob.glob(f"{meta_dir}/*.meta"):
+        name = os.path.basename(f).replace('.meta', '')
+        try:
+            with open(f) as mf:
+                meta[name] = json.load(mf)
+        except Exception:
+            meta[name] = {}
+
+    last_events = {}
+    try:
+        with open(events_log) as f:
+            for line in f:
+                try:
+                    e = json.loads(line.strip())
+                    client = e.get('client', '')
+                    if client:
+                        last_events[client] = e
+                except Exception:
+                    pass
+    except Exception:
+        pass
+
+    for conf in sorted(glob.glob(f"{clients_dir}/*.conf")):
+        name = os.path.basename(conf).replace('.conf', '')
+        ip = ''
+        try:
+            with open(conf) as f:
+                for line in f:
+                    if line.startswith('Address'):
+                        ip = line.split('=')[1].strip().split('/')[0]
+                        break
+        except Exception:
+            pass
+
+        m          = meta.get(name, {})
+        rule       = m.get('rule', '')
+        peer_type  = m.get('type', '')
+        main_group = m.get('main_group', '')
+
+        last_event = last_events.get(name, {})
+        last_ts    = last_event.get('timestamp', '')
+        last_evt   = last_event.get('event', '')
+
+        print(f"{name}|{ip}|{rule}|{peer_type}|{last_ts}|{last_evt}|{main_group}")
+
+
+def peer_transfer(wg_interface):
+    """Get total transfer bytes per peer."""
+    import subprocess
+    low  = int(os.environ.get('ACTIVITY_TOTAL_LOW_BYTES',  '1000000'))
+    med  = int(os.environ.get('ACTIVITY_TOTAL_MED_BYTES',  '10000000'))
+    high = int(os.environ.get('ACTIVITY_TOTAL_HIGH_BYTES', '100000000'))
+    try:
+        result = subprocess.run(
+            ['wg', 'show', wg_interface, 'transfer'],
+            capture_output=True, text=True
+        )
+        for line in result.stdout.strip().split('\n'):
+            if not line:
+                continue
+            parts = line.split('\t')
+            if len(parts) == 3:
+                pubkey, rx, tx = parts
+                total = int(rx) + int(tx)
+                if total == 0:        level = 'none'
+                elif total < low:     level = 'low'
+                elif total < med:     level = 'medium'
+                elif total < high:    level = 'high'
+                else:                 level = 'very high'
+                print(f"{pubkey}|{rx}|{tx}|{level}")
+    except Exception:
+        pass
+
+
+def peer_transfer_delta(wg_interface, cache_file):
+    """Calculate current transfer rate using delta from previous sample."""
+    import subprocess
+    import time
+
+    low  = int(os.environ.get('ACTIVITY_CURRENT_LOW_BYTES',  '10000'))
+    med  = int(os.environ.get('ACTIVITY_CURRENT_MED_BYTES',  '100000'))
+    high = int(os.environ.get('ACTIVITY_CURRENT_HIGH_BYTES', '1000000'))
+
+    current = {}
+    now = time.time()
+    try:
+        result = subprocess.run(
+            ['wg', 'show', wg_interface, 'transfer'],
+            capture_output=True, text=True
+        )
+        for line in result.stdout.strip().split('\n'):
+            if not line:
+                continue
+            parts = line.split('\t')
+            if len(parts) == 3:
+                pubkey, rx, tx = parts
+                current[pubkey] = {'rx': int(rx), 'tx': int(tx), 'ts': now}
+    except Exception:
+        pass
+
+    prev = {}
+    if os.path.exists(cache_file):
+        try:
+            with open(cache_file) as f:
+                prev = json.load(f)
+        except Exception:
+            pass
+
+    try:
+        with open(cache_file, 'w') as f:
+            json.dump(current, f)
+    except Exception:
+        pass
+
+    for pubkey, data in current.items():
+        if pubkey in prev:
+            dt = data['ts'] - prev[pubkey].get('ts', data['ts'])
+            if dt > 0:
+                rx_rate = max(0, (data['rx'] - prev[pubkey]['rx']) / dt)
+                tx_rate = max(0, (data['tx'] - prev[pubkey]['tx']) / dt)
+                total   = rx_rate + tx_rate
+                if total <= 0:        level = 'idle'
+                elif total < low:     level = 'low'
+                elif total < med:     level = 'medium'
+                elif total < high:    level = 'high'
+                else:                 level = 'very high'
+                print(f"{pubkey}|{int(rx_rate)}|{int(tx_rate)}|{level}")
+            else:
+                print(f"{pubkey}|0|0|idle")
+        else:
+            print(f"{pubkey}|0|0|unknown")
+
+
+def peer_group_map(groups_dir):
+    """Return peer:group pairs for all groups."""
+    try:
+        for group_file in glob.glob(f"{groups_dir}/*.group"):
+            try:
+                with open(group_file) as f:
+                    g = json.load(f)
+                name = g.get('name', '')
+                for peer in g.get('peers', []):
+                    if peer:
+                        print(f"{peer}:{name}")
+            except Exception:
+                pass
+    except Exception:
+        pass
+
+
+def peer_groups(groups_dir, peer_name):
+    """Find all groups containing a peer."""
+    try:
+        for group_file in glob.glob(f"{groups_dir}/*.group"):
+            try:
+                with open(group_file) as f:
+                    g = json.load(f)
+                if peer_name in g.get('peers', []):
+                    print(g.get('name', ''))
+            except Exception:
+                pass
+    except Exception:
+        pass

core/lib/util.py

@@ -0,0 +1,255 @@
+"""
+util.py — shared utilities for wgctl json_helper modules.
+Imported by all other lib modules.
+"""
+
+import os
+import json
+import sys
+from datetime import datetime, timezone, timedelta
+
+# ──────────────────────────────────────────
+# Global config (read from environment)
+# ──────────────────────────────────────────
+
+DATETIME_FMT = os.environ.get('WGCTL_DATETIME_FMT', '%Y-%m-%d %H:%M')
+DATE_FORMAT   = os.environ.get('WGCTL_DATE_FORMAT', 'eu')  # eu | iso
+
+PROTO_MAP = {1: 'icmp', 6: 'tcp', 17: 'udp'}
+
+# ──────────────────────────────────────────
+# IP → Peer name map
+# ──────────────────────────────────────────
+
+def build_ip_to_name(clients_dir):
+    """
+    Build a dict mapping peer IP -> peer name from .conf files.
+    Cached per process — call once, reuse.
+    """
+    import glob
+    ip_to_name = {}
+    for conf in glob.glob(f"{clients_dir}/*.conf"):
+        name = os.path.basename(conf).replace('.conf', '')
+        try:
+            with open(conf) as f:
+                for line in f:
+                    if line.startswith('Address'):
+                        ip = line.split('=')[1].strip().split('/')[0]
+                        ip_to_name[ip] = name
+                        break
+        except Exception:
+            pass
+    return ip_to_name
+
+
+def build_pubkey_to_name(clients_dir):
+    """
+    Build a dict mapping public key -> peer name from *_public.key files.
+    """
+    import glob
+    pubkey_to_peer = {}
+    for kf in glob.glob(f"{clients_dir}/*_public.key"):
+        name = os.path.basename(kf).replace('_public.key', '')
+        try:
+            with open(kf) as f:
+                key = f.read().strip()
+                if key:
+                    pubkey_to_peer[key] = name
+        except Exception:
+            pass
+    return pubkey_to_peer
+
+
+# ──────────────────────────────────────────
+# Service reverse lookup
+# ──────────────────────────────────────────
+
+def load_net_data(net_file):
+    """Load services.json into a dict. Returns {} on failure."""
+    if not net_file or not os.path.exists(net_file):
+        return {}
+    try:
+        with open(net_file) as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def reverse_lookup(net_data, dest_ip, dest_port='', proto=''):
+    """
+    Resolve dest_ip[:port] to a service name using services.json data.
+    Returns '' if no match found.
+    """
+    for svc_name, svc in net_data.items():
+        if not isinstance(svc, dict):
+            continue
+        if svc.get('ip', '') != dest_ip:
+            continue
+        ports = svc.get('ports', {})
+        if dest_port:
+            for port_name, port_def in ports.items():
+                if not isinstance(port_def, dict):
+                    continue
+                if (str(port_def.get('port', '')) == str(dest_port) and
+                        port_def.get('proto', 'tcp') == proto):
+                    return f"{svc_name}:{port_name}"
+            # IP matched but no port match — return service name
+            return svc_name
+        return svc_name
+    return ''
+
+
+def load_hosts_data(hosts_file):
+    """Load hosts.json into a dict. Returns empty structure on failure."""
+    if not hosts_file or not os.path.exists(hosts_file):
+        return {"hosts": {}, "subnets": {}, "ports": {}}
+    try:
+        with open(hosts_file) as f:
+            data = json.load(f)
+        data.setdefault("hosts", {})
+        data.setdefault("subnets", {})
+        data.setdefault("ports", {})
+        return data
+    except Exception:
+        return {"hosts": {}, "subnets": {}, "ports": {}}
+
+
+def hosts_lookup(hosts_data, ip):
+    """
+    Resolve IP to display name using hosts.json data.
+    Returns '' if no match.
+    """
+    entry = hosts_data.get("hosts", {}).get(ip)
+    if not entry:
+        return ''
+    if isinstance(entry, dict):
+        return entry.get('name', '')
+    return str(entry)
+
+
+def resolve_display(net_data, hosts_data, dest_ip, dest_port='', proto=''):
+    """
+    Full resolution chain:
+      1. hosts.json exact IP match
+      2. services.json match
+      3. raw IP fallback (returns dest_ip)
+    """
+    # 1. hosts.json
+    name = hosts_lookup(hosts_data, dest_ip)
+    if name:
+        return name
+    # 2. services.json
+    name = reverse_lookup(net_data, dest_ip, dest_port, proto)
+    if name:
+        return name
+    # 3. raw fallback
+    return dest_ip
+
+
+# ──────────────────────────────────────────
+# Timestamp utilities
+# ──────────────────────────────────────────
+
+def fmt_ts(ts_str, fmt=None):
+    """
+    Format an ISO timestamp string using DATETIME_FMT (or override fmt).
+    Returns ts_str unchanged on failure.
+    """
+    fmt = fmt or DATETIME_FMT
+    try:
+        dt = datetime.fromisoformat(ts_str)
+        return dt.strftime(fmt)
+    except Exception:
+        return ts_str
+
+
+def fmt_ts_hour(ts_str, fmt=None):
+    """
+    Format an ISO timestamp to hour precision (minutes replaced with 00).
+    """
+    fmt = fmt or DATETIME_FMT
+    hour_fmt = fmt.replace('%M', '00')
+    try:
+        dt = datetime.fromisoformat(ts_str)
+        return dt.strftime(hour_fmt)
+    except Exception:
+        return ts_str
+
+
+def ts_to_unix(ts_str):
+    """Convert ISO timestamp to unix float. Returns 0.0 on failure."""
+    try:
+        dt = datetime.fromisoformat(ts_str)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt.timestamp()
+    except Exception:
+        return 0.0
+
+
+def parse_since(value, date_format=None):
+    """
+    Parse a --since value to a datetime (UTC-aware).
+    Accepts:
+      Relative:  2h, 30m, 7d
+      EU date:   23/05, 23/05/2026, 23-05, 23-05-2026
+      ISO date:  2026-05-23, 2026-05-23 03:00
+    Returns None on failure.
+    """
+    import re
+    date_format = date_format or DATE_FORMAT
+    value = value.strip()
+
+    # Relative: e.g. 2h, 30m, 7d
+    m = re.fullmatch(r'(\d+)([mhd])', value)
+    if m:
+        n, unit = int(m.group(1)), m.group(2)
+        delta = {'m': timedelta(minutes=n),
+                 'h': timedelta(hours=n),
+                 'd': timedelta(days=n)}[unit]
+        return datetime.now(timezone.utc) - delta
+
+    now_year = datetime.now().year
+
+    # EU formats: 23/05, 23/05/2026, 23-05, 23-05-2026
+    for pattern, fmt in [
+        (r'(\d{1,2})/(\d{1,2})$',          f'%d/%m/{now_year}'),
+        (r'(\d{1,2})/(\d{1,2})/(\d{4})$',  '%d/%m/%Y'),
+        (r'(\d{1,2})-(\d{1,2})$',           f'%d-%m-{now_year}'),
+        (r'(\d{1,2})-(\d{1,2})-(\d{4})$',  '%d-%m-%Y'),
+    ]:
+        if re.fullmatch(pattern, value):
+            try:
+                if f'/{now_year}' in fmt or f'-{now_year}' in fmt:
+                    dt = datetime.strptime(f"{value}/{now_year}" if '/' in value
+                                           else f"{value}-{now_year}", fmt)
+                else:
+                    dt = datetime.strptime(value, fmt)
+                return dt.replace(tzinfo=timezone.utc)
+            except Exception:
+                pass
+
+    # ISO formats: 2026-05-23, 2026-05-23 03:00
+    for fmt in ('%Y-%m-%d', '%Y-%m-%d %H:%M'):
+        try:
+            dt = datetime.strptime(value, fmt)
+            return dt.replace(tzinfo=timezone.utc)
+        except Exception:
+            pass
+
+    return None
+
+
+# ──────────────────────────────────────────
+# Dest display formatting
+# ──────────────────────────────────────────
+
+def make_dest_display(dest_ip, dest_port, proto, svc_name):
+    """Build a human-readable destination string."""
+    if svc_name and svc_name != dest_ip:
+        return svc_name
+    if dest_port:
+        return f"{dest_ip}:{dest_port}/{proto}"
+    if proto and proto not in ('tcp',):
+        return f"{dest_ip} ({proto})"
+    return dest_ip