init

outils/temps-session.py

@@ -0,0 +1,226 @@
+#!/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. 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.0"
+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])
+    return [(t, p and humain(d, questions)) for t, p, d in ev]
+
+
+def plages(ev, pause):
+    """Une plage s'arrete quand l'agent attend l'humain, 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()