TKfA/outils/temps-session.py

#!/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()