naps62
bf1dc76125
A markdown table carries no borders of its own; whether any get drawn is up to the renderer, and terminal renderers mostly draw none. Emitting the grid directly makes the output look the same everywhere it lands. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
353 lines
12 KiB
Python
Executable File
353 lines
12 KiB
Python
Executable File
#!/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 a table: one row per day, one column per project, each cell holding
|
|
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.
|
|
|
|
Default rendering is a drawn grid, because a markdown table has no borders of
|
|
its own and terminal renderers vary in whether they draw any. --markdown emits
|
|
pipes instead, for pasting somewhere that does render them.
|
|
"""
|
|
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 fmt_hour_list(hours):
|
|
"""[0,7,20,21,22] -> '00h, 07h, 20-22h'."""
|
|
out, run = [], []
|
|
for h in sorted(set(hours)) + [None]:
|
|
if run and h == run[-1] + 1:
|
|
run.append(h)
|
|
continue
|
|
if run:
|
|
out.append(f"{run[0]:02d}h" if len(run) == 1
|
|
else f"{run[0]:02d}-{run[-1]:02d}h")
|
|
run = [h] if h is not None else []
|
|
return ", ".join(out)
|
|
|
|
|
|
def widths(rows):
|
|
return [max(len(r[i]) for r in rows) for i in range(len(rows[0]))]
|
|
|
|
|
|
def _cells(row, w, align_right):
|
|
return "| " + " | ".join(
|
|
c.rjust(w[i]) if i in align_right else c.ljust(w[i])
|
|
for i, c in enumerate(row)) + " |"
|
|
|
|
|
|
def box_table(rows, align_right=()):
|
|
"""Render rows (first is the header) as a grid with every cell bordered."""
|
|
w = widths(rows)
|
|
rule = "+" + "+".join("-" * (n + 2) for n in w) + "+"
|
|
out = [rule]
|
|
for row in rows:
|
|
out += [_cells(row, w, align_right), rule]
|
|
return "\n".join(out)
|
|
|
|
|
|
def md_table(rows, align_right=()):
|
|
"""Render rows (first is the header) as a padded markdown table."""
|
|
w = widths(rows)
|
|
sep = "|" + "|".join(
|
|
("-" * (w[i] + 1) + ":") if i in align_right else ("-" * (w[i] + 2))
|
|
for i in range(len(w))) + "|"
|
|
return "\n".join(
|
|
[_cells(rows[0], w, align_right), sep] +
|
|
[_cells(r, w, align_right) for r in rows[1:]])
|
|
|
|
|
|
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")
|
|
ap.add_argument("--markdown", action="store_true",
|
|
help="pipe table for pasting, instead of a drawn grid")
|
|
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
|
|
hit = False
|
|
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
|
|
hit = True
|
|
grid[t.date().isoformat()][proj][t.hour].add(
|
|
int(t.timestamp()) // slot)
|
|
nfiles += hit
|
|
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
|
|
|
|
print(f"{since} .. {until} — {nfiles} sessions, {mins}-min slots, "
|
|
f"{cfg['timezone']}, {cfg['nominal_day_hours']}h day")
|
|
print("hours are a suggested split of a full day, not measured time\n")
|
|
|
|
cols = [p for p in dict.fromkeys(
|
|
x["project"] for e in report for x in e["projects"])
|
|
if not p.startswith("unmapped:")]
|
|
if not cols:
|
|
print("no mapped project activity in this range")
|
|
return 0
|
|
|
|
rows = [["day"] + cols + ["total", "flags"]]
|
|
week = collections.Counter()
|
|
for e in report:
|
|
if not e["projects"]:
|
|
continue
|
|
by = {p["project"]: p for p in e["projects"]}
|
|
cells = []
|
|
for c in cols:
|
|
p = by.get(c)
|
|
cells.append("—" if not p or not p["suggested_hours"] else
|
|
f"{fmt_hours(p['suggested_hours'])} · "
|
|
f"{p['share']*100:.0f}% · {p['active_minutes']}min")
|
|
if p:
|
|
week[c] += p["suggested_hours"]
|
|
day_sum = sum(by[c]["suggested_hours"] for c in cols if c in by)
|
|
flags = []
|
|
if e["thin"]:
|
|
flags.append("under 30min, treat as empty")
|
|
outside = sorted({h for p in e["projects"] for h in p["outside_workday"]})
|
|
if outside:
|
|
flags.append("outside " + fmt_hour_list(outside))
|
|
if any(p["project"].startswith("unmapped:") for p in e["projects"]):
|
|
flags.append("unmapped time excluded")
|
|
rows.append([f"{e['date']} {e['weekday']}"] + cells +
|
|
[fmt_hours(day_sum) if day_sum else "—", "; ".join(flags)])
|
|
|
|
bold = (lambda s: f"**{s}**") if a.markdown else (lambda s: s)
|
|
rows.append([bold("total")] +
|
|
[bold(fmt_hours(week[c])) for c in cols] +
|
|
[bold(fmt_hours(sum(week.values()))), ""])
|
|
right = set(range(1, len(cols) + 2))
|
|
render = md_table if a.markdown else box_table
|
|
print(render(rows, align_right=right))
|
|
|
|
if unmapped:
|
|
print("\nunmapped paths — add them to the config or the exclude list:")
|
|
for u in unmapped:
|
|
print(" ", u)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|