agent-skills/skills/hourlog/scripts/scan-activity.py

#!/usr/bin/env python3
"""Bucket Claude Code and Codex session activity into days per project.

Usage: scan-activity.py --since YYYY-MM-DD [--until YYYY-MM-DD] [--json]
       scan-activity.py --week last|this

Reads ~/.config/hourlog/projects.json for the path-prefix -> project mapping
(see config.example.json). The mapping is machine-local on purpose: this repo
is public and client names are not.

The unit is an "active slot" — a 5-minute window in which at least one message
was written. Slots are a set per (day, project), so a swarm of 40 subagent
transcripts on one project counts once, and two projects worked in parallel
each keep their own slots. Message counts would make one overnight autonomous
run outweigh a real morning; wall-clock presence does not.

Output is one line per project per day: suggested hours, share of the day, and
the active minutes behind it. The suggestion splits a nominal working day by
share, in 15-minute steps — active minutes are a floor on real work, never a
measure of it, so the day's length comes from the calendar and only the split
between projects comes from the sessions. See SKILL.md.
"""
import argparse
import collections
import datetime as dt
import glob
import json
import os
import sys
from zoneinfo import ZoneInfo

CONFIG = os.path.expanduser("~/.config/hourlog/projects.json")
DEFAULTS = {
    "timezone": "Europe/Lisbon",
    "slot_minutes": 5,
    "day_start_hour": 8,
    "day_end_hour": 20,
    "nominal_day_hours": 8,
    "projects": [],
    "exclude": [],
}


def load_config(path):
    cfg = dict(DEFAULTS)
    if os.path.exists(path):
        with open(path) as fh:
            cfg.update(json.load(fh))
    else:
        print(f"no config at {path} — every path will land in 'unmapped'",
              file=sys.stderr)
    # Longest prefix wins, so a worktree dir can override its parent.
    rules = []
    for p in cfg["projects"]:
        for m in p["match"]:
            rules.append((os.path.expanduser(m).rstrip("/"), p["name"]))
    for m in cfg["exclude"]:
        rules.append((os.path.expanduser(m).rstrip("/"), None))
    cfg["_rules"] = sorted(rules, key=lambda r: -len(r[0]))
    return cfg


def classify(cwd, rules):
    """-> project name, or None if excluded, or 'unmapped:<top-3-dirs>'."""
    if not cwd:
        return "unmapped:?"
    for prefix, name in rules:
        if cwd == prefix or cwd.startswith(prefix + "/"):
            return name
    short = cwd.replace(os.path.expanduser("~"), "~")
    return "unmapped:" + "/".join(short.split("/")[:3])


def claude_files(root, cutoff):
    for f in glob.glob(os.path.join(root, "*", "*.jsonl")):
        try:
            if os.stat(f).st_mtime >= cutoff:
                yield f
        except OSError:
            continue


def codex_files(root, cutoff):
    for f in glob.glob(os.path.join(root, "*", "*", "*", "*.jsonl")):
        try:
            if os.stat(f).st_mtime >= cutoff:
                yield f
        except OSError:
            continue


def read_session(path):
    """-> (cwd, [timestamp strings]). Handles both Claude and Codex layouts."""
    cwd, stamps = None, []
    try:
        fh = open(path, errors="replace")
    except OSError:
        return None, []
    with fh:
        for line in fh:
            try:
                d = json.loads(line)
            except ValueError:
                continue
            if cwd is None:
                cwd = d.get("cwd") or (d.get("payload") or {}).get("cwd")
            ts = d.get("timestamp")
            if ts:
                stamps.append(ts)
    return cwd, stamps


def split_day(entry, nominal_hours):
    """Split a nominal working day across the day's mapped projects by share.

    Largest-remainder over 15-minute steps, so the parts add back up to the
    whole day exactly. Unmapped paths get nothing: they are not known to be
    client work, and silently handing them hours would hide a missing rule.
    """
    for p in entry["projects"]:
        p["suggested_hours"] = 0.0
    entry["thin"] = entry["total_active_minutes"] < 30
    mapped = [p for p in entry["projects"]
              if not p["project"].startswith("unmapped:")]
    weight = sum(p["active_minutes"] for p in mapped)
    if not weight or entry["thin"]:
        return
    quarters = int(round(nominal_hours * 4))
    exact = [(p, quarters * p["active_minutes"] / weight) for p in mapped]
    given = [(p, int(v)) for p, v in exact]
    left = quarters - sum(v for _, v in given)
    order = sorted(range(len(exact)), key=lambda i: -(exact[i][1] % 1))
    for i in order[:left]:
        given[i] = (given[i][0], given[i][1] + 1)
    for p, q in given:
        p["suggested_hours"] = q / 4


def fmt_hours(h):
    whole, rem = int(h), round((h - int(h)) * 60)
    return f"{whole}h" if not rem else f"{whole}h{rem:02d}"


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--since")
    ap.add_argument("--until")
    ap.add_argument("--week", choices=["last", "this"])
    ap.add_argument("--config", default=CONFIG)
    ap.add_argument("--json", action="store_true")
    a = ap.parse_args()

    cfg = load_config(a.config)
    tz = ZoneInfo(cfg["timezone"])
    today = dt.datetime.now(tz).date()

    if a.week:
        monday = today - dt.timedelta(days=today.weekday())
        if a.week == "last":
            monday -= dt.timedelta(days=7)
        since, until = monday, monday + dt.timedelta(days=6)
    elif a.since:
        since = dt.date.fromisoformat(a.since)
        until = dt.date.fromisoformat(a.until) if a.until else today
    else:
        ap.error("need --since or --week")

    start = dt.datetime.combine(since, dt.time(0), tz)
    end = dt.datetime.combine(until, dt.time(23, 59, 59), tz)
    slot = cfg["slot_minutes"] * 60

    # date -> project -> hour -> set of slot indices
    grid = collections.defaultdict(
        lambda: collections.defaultdict(lambda: collections.defaultdict(set)))

    sources = [
        (os.path.expanduser("~/.claude/projects"), claude_files),
        (os.path.expanduser("~/.codex/sessions"), codex_files),
    ]
    nfiles = 0
    for root, lister in sources:
        if not os.path.isdir(root):
            continue
        for f in lister(root, start.timestamp()):
            cwd, stamps = read_session(f)
            proj = classify(cwd, cfg["_rules"])
            if proj is None:
                continue
            nfiles += 1
            for ts in stamps:
                try:
                    t = dt.datetime.fromisoformat(
                        ts.replace("Z", "+00:00")).astimezone(tz)
                except ValueError:
                    continue
                if not (start <= t <= end):
                    continue
                grid[t.date().isoformat()][proj][t.hour].add(
                    int(t.timestamp()) // slot)
            del stamps

    mins = cfg["slot_minutes"]
    report = []
    d = since
    while d <= until:
        per = grid.get(d.isoformat(), {})
        totals = {p: sum(len(s) for s in hrs.values()) for p, hrs in per.items()}
        day_total = sum(totals.values())
        entry = {
            "date": d.isoformat(),
            "weekday": d.strftime("%a"),
            "total_active_minutes": day_total * mins,
            "nominal_day_hours": cfg["nominal_day_hours"],
            "projects": [],
        }
        for proj, n in sorted(totals.items(), key=lambda kv: -kv[1]):
            hrs = sorted(per[proj])
            entry["projects"].append({
                "project": proj,
                "active_minutes": n * mins,
                "share": round(n / day_total, 3) if day_total else 0,
                "first_hour": hrs[0],
                "last_hour": hrs[-1],
                "outside_workday": [
                    h for h in hrs
                    if h < cfg["day_start_hour"] or h >= cfg["day_end_hour"]
                ],
            })
        split_day(entry, cfg["nominal_day_hours"])
        report.append(entry)
        d += dt.timedelta(days=1)

    unmapped = sorted(
        {p for e in report for p in
         (x["project"] for x in e["projects"]) if p.startswith("unmapped:")})

    if a.json:
        json.dump({"since": since.isoformat(), "until": until.isoformat(),
                   "slot_minutes": mins, "sessions": nfiles,
                   "nominal_day_hours": cfg["nominal_day_hours"],
                   "unmapped": unmapped, "report": report}, sys.stdout, indent=1)
        print()
        return 0

    cols = [p for p in dict.fromkeys(
        x["project"] for e in report for x in e["projects"])
        if not p.startswith("unmapped:")]
    widths = [max(len(c), 6) for c in cols]
    head = "day         " + "  ".join(
        f"{c:>{w}}" for c, w in zip(cols, widths)) + "   total"
    print(head)
    print("-" * len(head))

    totals = collections.Counter()
    flagged = False
    for e in report:
        hours = {p["project"]: p["suggested_hours"] for p in e["projects"]}
        day_sum = sum(hours.get(c, 0) for c in cols)
        cells = "  ".join(
            f"{fmt_hours(hours[c]) if hours.get(c) else '—':>{w}}"
            for c, w in zip(cols, widths))
        mark = " *" if e["thin"] and e["total_active_minutes"] else ""
        flagged = flagged or bool(mark)
        print(f"{e['date']} {e['weekday']}  {cells}  "
              f"{fmt_hours(day_sum) if day_sum else '—':>5}{mark}")
        for c in cols:
            totals[c] += hours.get(c, 0)

    print("-" * len(head))
    print("total       " + "  ".join(
        f"{fmt_hours(totals[c]):>{w}}" for c, w in zip(cols, widths)) +
        f"  {fmt_hours(sum(totals.values())):>5}")

    if flagged:
        print("\n* under 30min of activity — treat as empty unless you know "
              "otherwise")
    if unmapped:
        print("\nunmapped paths, excluded from the split — add them to the "
              "config or the exclude list:")
        for u in unmapped:
            print("   ", u)
    return 0


if __name__ == "__main__":
    sys.exit(main())