george-claudeandClaude Opus 5.5
0b70aad64a
Une entrée user arrivée après la fin d'un tour de l'agent (dernière réponse du fil principal avec stop_reason autre que tool_use) ouvre une nouvelle plage, même écrite par Claude Code (isMeta) : message ou avis d'une autre session, commande slash, renommage. Une entrée isMeta écrite pendant un tour, ou après la fin d'un tour de sous-agent, ne coupe rien. Ajoute le fichier VERSION (1.0.1), vérifié égal à la constante de l'outil. Refs #2 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KT2yBsd7nPivLnrEtcvuhX
250 lines
10 KiB
Python
250 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Temps actif d'une session Claude Code, pret a saisir sur un ticket Gitea.
|
|
|
|
Usage :
|
|
python3 temps-session.py --depot organisation/depot --ticket 12 \
|
|
[--depuis 2026-10-10T09:00:00+02:00] [--jusqua 2026-10-10T11:30:00+02:00] \
|
|
[--session ID] [--pause 1800] [--seance 3600] [--jeton FICHIER] \
|
|
[--instance NOM] [--config FICHIER]
|
|
|
|
--depuis et --jusqua bornent le travail sur le ticket. --jusqua sert a saisir
|
|
apres coup un ticket quitte pour un autre dans la meme session : sans lui, le
|
|
travail fait ensuite sur l'autre ticket serait compte sur celui-ci.
|
|
|
|
Lit le transcript de la session (~/.claude/projects/*/<session>.jsonl et ses
|
|
sous-agents) et ne compte que le temps ou l'agent travaille : du message de
|
|
l'humain (ou de sa reponse a une question) jusqu'au dernier evenement avant
|
|
qu'il reprenne la main. L'attente de l'humain n'est jamais comptee, qu'elle
|
|
dure une minute ou trois jours ; celle d'un message d'une autre session non
|
|
plus (entree user, meme ecrite par Claude Code, arrivee apres la fin du
|
|
tour). Au sein d'un tour, un trou de plus de --pause secondes (autorisation
|
|
restee sans reponse) n'est pas compte non plus.
|
|
|
|
Une ligne par seance de travail (tours separes de moins de --seance secondes),
|
|
pour ne pas couvrir le ticket de saisies : sa duree est le temps actif cumule,
|
|
datee de la fin reelle de la seance. La date de la derniere saisie sert de
|
|
repere a la suivante : tout ce qui la precede est deja compte, relancer ne
|
|
double rien. Gitea ignore la date envoyee et date la saisie de son envoi,
|
|
posterieur a la fin de la seance : le repere reste juste dans les deux cas.
|
|
|
|
Sortie : une ligne JSON par seance restant a saisir, au format attendu par
|
|
POST /repos/{depot}/issues/{ticket}/times ({"created": ..., "time": secondes}).
|
|
Les seances de moins d'une minute sont ignorees.
|
|
|
|
L'instance Gitea interrogee se choisit dans un fichier de configuration TOML
|
|
(voir tkfa.exemple.toml) : --config, sinon TKFA_CONFIG, sinon
|
|
~/.config/tkfa/config.toml s'il existe. L'instance est --instance, sinon la
|
|
cle `defaut` du fichier. Sans instance, l'outil n'interroge rien. Le jeton
|
|
est --jeton, sinon le fichier de la cle `jeton` de l'instance, sinon celui
|
|
que nomme la variable de sa cle `jeton_env` : une variable ne sert que
|
|
l'instance qui la nomme, le jeton d'une instance ne part pas vers une autre.
|
|
|
|
Ne lit que le transcript et l'API Gitea (GET). N'ecrit rien, n'envoie rien.
|
|
"""
|
|
import argparse, datetime, glob, json, os, sys, tomllib, urllib.request
|
|
|
|
VERSION = "1.0.1"
|
|
MIN = 60
|
|
|
|
|
|
def iso(s):
|
|
return datetime.datetime.fromisoformat(s.replace("Z", "+00:00"))
|
|
|
|
|
|
def borne(s):
|
|
"""Date de --depuis ou --jusqua ; sans fuseau, celui de la machine."""
|
|
d = iso(s)
|
|
return d if d.tzinfo else d.astimezone()
|
|
|
|
|
|
def humain(d, questions):
|
|
"""Vrai si l'entree est une reponse de l'humain : message tape, ou
|
|
reponse a une question posee par l'agent (AskUserQuestion)."""
|
|
if d.get("type") != "user" or d.get("isMeta"):
|
|
return False
|
|
c = (d.get("message") or {}).get("content")
|
|
if isinstance(c, str):
|
|
return True
|
|
return any(b.get("type") == "text" or (b.get("type") == "tool_result"
|
|
and b.get("tool_use_id") in questions) for b in c or [])
|
|
|
|
|
|
def evenements(session):
|
|
base = os.path.expanduser("~/.claude/projects/*/")
|
|
principal = glob.glob(base + session + ".jsonl")
|
|
if not principal:
|
|
sys.exit("transcript introuvable pour la session %s" % session)
|
|
sous = glob.glob(base + session + "/**/*.jsonl", recursive=True)
|
|
ev, questions = [], set()
|
|
for f in principal + sous:
|
|
for line in open(f, errors="ignore"):
|
|
try:
|
|
d = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
# Seuls les echanges comptent : les entrees de service (titre,
|
|
# mode, cout...) s'ecrivent parfois pendant l'attente.
|
|
if d.get("type") not in ("user", "assistant") or not d.get("timestamp"):
|
|
continue
|
|
for b in (d.get("message") or {}).get("content") or []:
|
|
if isinstance(b, dict) and b.get("name") == "AskUserQuestion":
|
|
questions.add(b.get("id"))
|
|
ev.append((iso(d["timestamp"]), f in principal, d))
|
|
ev.sort(key=lambda e: e[0])
|
|
out, rendu = [], False
|
|
for t, p, d in ev:
|
|
fil = p and not d.get("isSidechain")
|
|
ouvre = p and (humain(d, questions) or (rendu and fil and d["type"] == "user"))
|
|
out.append((t, ouvre))
|
|
if fil:
|
|
rendu = rendu_la_main(d)
|
|
return out
|
|
|
|
|
|
def rendu_la_main(d):
|
|
"""Vrai si, apres l'entree d du transcript principal, l'agent attend : sa
|
|
reponse a fini son tour (stop_reason autre que tool_use). Toute entree
|
|
user qui suit relance alors le travail, meme ecrite par Claude Code
|
|
(message ou avis d'une autre session) : l'attente qui la precede n'est
|
|
pas comptee. Une entree user ecrite pendant un tour (rappel, sortie de
|
|
hook) ne change rien."""
|
|
if d["type"] == "assistant":
|
|
fin = (d.get("message") or {}).get("stop_reason")
|
|
return fin is not None and fin != "tool_use"
|
|
return False
|
|
|
|
|
|
def plages(ev, pause):
|
|
"""Une plage s'arrete quand l'agent attend (l'humain ou une autre
|
|
session : voir rendu_la_main), quelle que soit la duree de l'attente ; au
|
|
sein d'un tour, un trou de plus de --pause (une autorisation restee sans
|
|
reponse) l'arrete aussi."""
|
|
out = []
|
|
for t, h in ev:
|
|
if out and not h and (t - out[-1][1]).total_seconds() <= pause:
|
|
out[-1][1] = t
|
|
else:
|
|
out.append([t, t])
|
|
return out
|
|
|
|
|
|
def configuration(chemin):
|
|
"""Contenu du fichier de configuration ; {} s'il n'y en a pas. Un fichier
|
|
nomme par --config ou TKFA_CONFIG doit exister, le chemin par defaut non."""
|
|
explicite = chemin or os.environ.get("TKFA_CONFIG")
|
|
chemin = os.path.expanduser(explicite or os.path.join(
|
|
os.environ.get("XDG_CONFIG_HOME") or "~/.config", "tkfa", "config.toml"))
|
|
if not explicite and not os.path.exists(chemin):
|
|
return {}, None
|
|
try:
|
|
with open(chemin, "rb") as f:
|
|
return tomllib.load(f), os.path.dirname(os.path.abspath(chemin))
|
|
except (OSError, tomllib.TOMLDecodeError) as e:
|
|
sys.exit("configuration illisible (%s) : %s" % (chemin, e))
|
|
|
|
|
|
def instance(a):
|
|
"""(nom, api, fichier du jeton) de l'instance Gitea a interroger."""
|
|
conf, dossier = configuration(a.config)
|
|
nom = a.instance or conf.get("defaut")
|
|
if not nom:
|
|
sys.exit("aucune instance Gitea : passer --instance, ou poser `defaut` "
|
|
"dans la configuration (voir tkfa.exemple.toml)")
|
|
i = (conf.get("instances") or {}).get(nom)
|
|
if not isinstance(i, dict) or not i.get("api"):
|
|
sys.exit("instance %s absente de la configuration, ou sans `api`" % nom)
|
|
api = i["api"].rstrip("/")
|
|
if not api.startswith("https://"):
|
|
sys.exit("instance %s : l'api doit etre en https (le jeton y passe)" % nom)
|
|
jeton = a.jeton or i.get("jeton")
|
|
# Un chemin relatif dans le fichier se lit depuis le dossier du fichier.
|
|
if jeton and not a.jeton and dossier:
|
|
jeton = os.path.join(dossier, os.path.expanduser(jeton))
|
|
if not jeton and i.get("jeton_env"):
|
|
jeton = os.environ.get(i["jeton_env"])
|
|
return nom, api, jeton
|
|
|
|
|
|
def gitea(api, chemin, jeton):
|
|
req = urllib.request.Request(api + chemin,
|
|
headers={"Authorization": "token " + jeton})
|
|
with urllib.request.urlopen(req) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def deja_saisi(api, depot, ticket, jeton):
|
|
"""Fin de la derniere seance deja saisie sur le ticket par ce compte.
|
|
Chaque saisie est datee de la fin de sa seance (voir main), ou de son
|
|
envoi quand Gitea ignore la date : tout ce qui precede est deja compte.
|
|
Ajouter la duree a cette date sauterait le travail qui suit l'envoi."""
|
|
moi = gitea(api, "/user", jeton)["login"]
|
|
repere, page = None, 1
|
|
while True:
|
|
lot = gitea(api, "/repos/%s/issues/%d/times?user=%s&limit=50&page=%d"
|
|
% (depot, ticket, moi, page), jeton)
|
|
for e in lot:
|
|
fin = iso(e["created"])
|
|
repere = max(repere or fin, fin)
|
|
if len(lot) < 50:
|
|
return moi, repere
|
|
page += 1
|
|
|
|
|
|
def seances(blocs, ecart):
|
|
"""Regroupe les plages separees de moins de --seance secondes."""
|
|
out = []
|
|
for d, f in blocs:
|
|
if out and (d - out[-1][-1][1]).total_seconds() <= ecart:
|
|
out[-1].append((d, f))
|
|
else:
|
|
out.append([(d, f)])
|
|
return out
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--version", action="version", version="%(prog)s " + VERSION)
|
|
p.add_argument("--depot", required=True)
|
|
p.add_argument("--ticket", type=int, required=True)
|
|
p.add_argument("--depuis")
|
|
p.add_argument("--jusqua")
|
|
p.add_argument("--session", default=os.environ.get("CLAUDE_CODE_SESSION_ID"))
|
|
p.add_argument("--pause", type=int, default=1800)
|
|
p.add_argument("--seance", type=int, default=3600)
|
|
p.add_argument("--jeton")
|
|
p.add_argument("--instance")
|
|
p.add_argument("--config")
|
|
a = p.parse_args()
|
|
if not a.session:
|
|
sys.exit("session inconnue : passer --session")
|
|
nom, api, fichier = instance(a)
|
|
if not fichier:
|
|
sys.exit("jeton inconnu pour l'instance %s : passer --jeton, ou lui donner "
|
|
"`jeton` ou `jeton_env` (variable definie)" % nom)
|
|
jeton = open(os.path.expanduser(fichier)).read().strip()
|
|
|
|
ev = evenements(a.session)
|
|
if a.depuis:
|
|
ev = [e for e in ev if e[0] >= borne(a.depuis)]
|
|
if a.jusqua:
|
|
ev = [e for e in ev if e[0] <= borne(a.jusqua)]
|
|
moi, repere = deja_saisi(api, a.depot, a.ticket, jeton)
|
|
|
|
blocs = [(max(d, repere) if repere else d, f)
|
|
for d, f in plages(ev, a.pause) if not repere or f > repere]
|
|
total = 0
|
|
for groupe in seances(blocs, a.seance):
|
|
s = int(sum((f - d).total_seconds() for d, f in groupe))
|
|
if s >= MIN:
|
|
total += s
|
|
# Une ligne par seance : la duree active, datee de la fin reelle
|
|
# de la seance (c'est le repere de la saisie suivante).
|
|
fin = groupe[-1][1]
|
|
print(json.dumps({"created": fin.isoformat(), "time": s}))
|
|
print("compte %s, %s#%d : %d min a saisir sur %s/repos/%s/issues/%d/times"
|
|
% (moi, a.depot, a.ticket, total // 60, api, a.depot, a.ticket), file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|