atlas-os/scripts/atlas_activity_export.py

#!/usr/bin/env python3
"""
atlas_activity_export.py - Query ActivityWatch API + classify into Atlas categories.
Outputs atlas-activity.json (today summary), pushed to atlas-01 dashboard.

Categories:
  sales     — Asana, WhatsApp Web, Outlook compose, Atlas Chat/Command/HUD
  delivery  — VSCode/Claude on existing-client work, atlasworks.nl wp-admin
  infra     — Claude Code, terminal, github, docker, atlas-* configs
  admin     — Inkomsten xlsx, Odoo, Knab/ING/Revolut, MEGA folder
  learning  — YouTube, BiSL/IT material, Obsidian vault read
  other     — everything else

Run daily via Atlas-Activity-Export schtask + before Friday-Wrap.
"""
import json
import sys
import subprocess
import urllib.request
import urllib.error
from datetime import datetime, timedelta
from pathlib import Path
from collections import defaultdict

AW_URL = "http://localhost:5600"
OUT_LOCAL = Path(r"C:\Users\Gebruiker\AppData\Local\Atlas\analyses\atlas-activity.json")
SSH = "root@<YOUR_VPS_IP>"
REMOTE = "/opt/atlas/dashboard/atlas-activity.json"

# Category classifiers (keyword in title or app)
CATEGORIES = {
    "sales": [
        "asana", "whatsapp", "atlas chat", "atlas command", "atlas hud",
        "outlook.com", "mail.atlasworks", "compose", "messenger",
        "instagram.com/direct", "linkedin.com/messag",
    ],
    "delivery": [
        "wp-admin", "atlasworks.nl/", "couscousbar", "charif", "bisogno",
        "le miris", "kolvari", "dushi", "al mahaba", "kapper", "barber",
        "elementor", "woocommerce",
    ],
    "infra": [
        "claude code", "claude.ai/", "cmd", "powershell", "terminal", "ms-terminal",
        "git", "github", "vscode", "code -",
        "docker", "atlas-01", "ssh ", "127.0.0.1", "tailscale",
        ".py - ", ".ps1 - ", "caddy", "ollama",
    ],
    "admin": [
        "inkomsten", "odoo", "knab", "ing.nl", "revolut", "mega",
        "vaultwarden", "factuur", "rekeningafschrift", "boekhoud",
    ],
    "learning": [
        "youtube.com/watch", "bisl", "udemy", "coursera",
        "obsidian", "atlas wiki", "00_atlas",
    ],
}


def fetch(url):
    try:
        with urllib.request.urlopen(url, timeout=5) as r:
            return json.loads(r.read().decode("utf-8", errors="replace"))
    except Exception:
        return None


def classify(text):
    t = (text or "").lower()
    for cat, keywords in CATEGORIES.items():
        for kw in keywords:
            if kw in t:
                return cat
    return "other"


def get_today_events():
    """Pull today's window events from aw-watcher-window bucket."""
    buckets = fetch(f"{AW_URL}/api/0/buckets/") or {}
    win_bucket = next((k for k in buckets if "aw-watcher-window" in k), None)
    if not win_bucket: return []
    now = datetime.now()
    start = datetime(now.year, now.month, now.day).isoformat()
    end = now.isoformat()
    url = f"{AW_URL}/api/0/buckets/{win_bucket}/events?start={start}&end={end}&limit=10000"
    return fetch(url) or []


def get_afk_events():
    """Pull today's AFK events to filter non-active time."""
    buckets = fetch(f"{AW_URL}/api/0/buckets/") or {}
    afk_bucket = next((k for k in buckets if "aw-watcher-afk" in k), None)
    if not afk_bucket: return []
    now = datetime.now()
    start = datetime(now.year, now.month, now.day).isoformat()
    end = now.isoformat()
    url = f"{AW_URL}/api/0/buckets/{afk_bucket}/events?start={start}&end={end}&limit=10000"
    return fetch(url) or []


def aggregate():
    win_events = get_today_events()
    afk_events = get_afk_events()

    # Build active timeline from AFK (only count non-afk windows)
    active_intervals = []
    for e in afk_events:
        if e.get("data", {}).get("status") == "not-afk":
            start = datetime.fromisoformat(e["timestamp"].replace("Z", "+00:00"))
            dur = e.get("duration", 0)
            active_intervals.append((start, start + timedelta(seconds=dur)))

    by_cat = defaultdict(float)
    by_app = defaultdict(float)
    by_title_top = defaultdict(float)

    for e in win_events:
        t = datetime.fromisoformat(e["timestamp"].replace("Z", "+00:00"))
        dur = e.get("duration", 0)
        if dur < 1: continue
        # Check if this event is during active (non-afk) time — approximate
        is_active = any(s <= t <= en for s, en in active_intervals)
        if not is_active: continue

        data = e.get("data", {})
        app = (data.get("app") or "?").lower()
        title = data.get("title", "")
        combined = f"{app} {title}"
        cat = classify(combined)
        by_cat[cat] += dur
        by_app[app] += dur
        by_title_top[title[:80]] += dur

    total = sum(by_cat.values())
    cat_pct = {c: round(s / total * 100, 1) if total else 0 for c, s in by_cat.items()}
    top_apps = sorted(by_app.items(), key=lambda x: x[1], reverse=True)[:5]
    top_titles = sorted(by_title_top.items(), key=lambda x: x[1], reverse=True)[:5]

    return {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "date": datetime.now().date().isoformat(),
        "active_total_hours": round(total / 3600, 2),
        "by_category_seconds": {c: int(s) for c, s in by_cat.items()},
        "by_category_pct": cat_pct,
        "top_apps": [{"app": a, "hours": round(s / 3600, 2)} for a, s in top_apps],
        "top_titles": [{"title": t, "hours": round(s / 3600, 2)} for t, s in top_titles],
        # Rule R10 check
        "r10_sales_hours": round(by_cat.get("sales", 0) / 3600, 2),
        "r10_infra_hours": round(by_cat.get("infra", 0) / 3600, 2),
        "r10_pass": by_cat.get("sales", 0) >= by_cat.get("infra", 0),
    }


def main():
    summary = aggregate()
    OUT_LOCAL.parent.mkdir(parents=True, exist_ok=True)
    OUT_LOCAL.write_text(json.dumps(summary, indent=2), encoding="utf-8")
    print(f"wrote {OUT_LOCAL}")
    print(f"  active: {summary['active_total_hours']}h today")
    print(f"  by category: {summary['by_category_pct']}")
    print(f"  R10 (sales >= infra): {'PASS' if summary['r10_pass'] else 'FAIL'} ({summary['r10_sales_hours']}h sales vs {summary['r10_infra_hours']}h infra)")
    try:
        subprocess.run(["scp", str(OUT_LOCAL), f"{SSH}:{REMOTE}"], check=True, timeout=20)
        print(f"pushed to {SSH}:{REMOTE}")
    except Exception as e:
        print(f"scp failed: {e}")


if __name__ == "__main__":
    main()