572 lines
20 KiB
Python
Executable File
572 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Pi Session Launcher
|
|
===================
|
|
GUI (GTK3) che elenca tutte le sessioni salvate di Pi Agent
|
|
(~/.pi/agent/sessions/**/*.jsonl) e, selezionandone una:
|
|
|
|
* apre un terminale (Konsole) nella cartella di lavoro della sessione (cwd)
|
|
* riprende la sessione con: pi --session <file di sessione>
|
|
|
|
Requisiti: python3-gi (GTK3), konsole (o xterm come fallback), pi installato.
|
|
"""
|
|
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gtk", "3.0")
|
|
gi.require_version("Gdk", "3.0")
|
|
from gi.repository import Gdk, GLib, Gtk, Pango
|
|
|
|
APP_TITLE = "Pi Session Launcher"
|
|
SESSIONS_DIR = os.path.expanduser("~/.pi/agent/sessions/")
|
|
MAX_NAME_SCAN = 16 * 1024 * 1024 # oltre questa dimensione, scandisci solo inizio/fine
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Lettura sessioni
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _decode(data: bytes) -> str:
|
|
return data.decode("utf-8", "replace")
|
|
|
|
|
|
def _first_user_message(text: str) -> str:
|
|
"""Cerca il primo messaggio con role=user nei primi chunk del file."""
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
entry = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if entry.get("type") != "message":
|
|
continue
|
|
msg = entry.get("message") or {}
|
|
if msg.get("role") != "user":
|
|
continue
|
|
content = msg.get("content", "")
|
|
if isinstance(content, list):
|
|
parts = []
|
|
for block in content:
|
|
if isinstance(block, dict) and block.get("type") == "text":
|
|
parts.append(block.get("text", ""))
|
|
return " ".join(parts).strip()
|
|
if isinstance(content, str):
|
|
return content.strip()
|
|
return ""
|
|
|
|
|
|
def _session_name(full_text: str) -> str:
|
|
"""Nome mostrato della sessione: come pi, l'ULTIMA entry session_info vince
|
|
(nome vuoto cancella il titolo)."""
|
|
name = ""
|
|
for line in reversed(full_text.splitlines()):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if '"type":"session_info"' not in line and '"type": "session_info"' not in line:
|
|
continue
|
|
try:
|
|
entry = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if entry.get("type") == "session_info":
|
|
return entry.get("name") or ""
|
|
return name
|
|
|
|
|
|
def parse_session(path: str):
|
|
"""Legge un file .jsonl di sessione e ne estrae le info utili."""
|
|
try:
|
|
stat = os.stat(path)
|
|
except OSError:
|
|
return None
|
|
if stat.st_size == 0:
|
|
return None
|
|
|
|
info = {
|
|
"path": path,
|
|
"cwd": "",
|
|
"sid": "",
|
|
"name": "",
|
|
"first": "",
|
|
"mtime": stat.st_mtime,
|
|
"size": stat.st_size,
|
|
}
|
|
|
|
try:
|
|
with open(path, "rb") as f:
|
|
header_raw = f.readline().decode("utf-8", "replace")
|
|
header = json.loads(header_raw)
|
|
if header.get("type") != "session":
|
|
return None
|
|
info["cwd"] = header.get("cwd", "") or ""
|
|
info["sid"] = header.get("id", "") or ""
|
|
|
|
# primo messaggio utente: leggiamo i primi ~64 KB (righe iniziali)
|
|
f.seek(0)
|
|
head = _decode(f.read(65536))
|
|
info["first"] = _first_user_message(head)
|
|
|
|
# nome sessione: scandisci tutto il file (con limite prudente)
|
|
if stat.st_size <= MAX_NAME_SCAN:
|
|
full = head + _decode(f.read())
|
|
else:
|
|
f.seek(max(0, stat.st_size - 1024 * 1024))
|
|
tail = _decode(f.read())
|
|
full = head + tail
|
|
info["name"] = _session_name(full)
|
|
except (OSError, json.JSONDecodeError, ValueError):
|
|
return None
|
|
|
|
return info
|
|
|
|
|
|
def rename_session(path, new_name):
|
|
"""Imposta (o cancella, con '') il nome visualizzato di una sessione,
|
|
appendendo una entry session_info al file JSONL (stesso meccanismo di pi).
|
|
pi usa il nome dell'ultima entry session_info presente nel file."""
|
|
last_id = None
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
e = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if e.get("type") != "session" and e.get("id"):
|
|
last_id = e["id"]
|
|
entry = {
|
|
"type": "session_info",
|
|
"id": os.urandom(4).hex(),
|
|
"parentId": last_id,
|
|
"timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
|
|
"name": new_name,
|
|
}
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(entry, ensure_ascii=False, separators=(",", ":")) + "\n")
|
|
|
|
|
|
def scan_sessions():
|
|
"""Elenca tutte le sessioni, ordinate per ultima modifica (più recente prima)."""
|
|
sessions = []
|
|
if not os.path.isdir(SESSIONS_DIR):
|
|
return sessions
|
|
for root, _dirs, files in os.walk(SESSIONS_DIR):
|
|
for fn in files:
|
|
if not fn.endswith(".jsonl"):
|
|
continue
|
|
info = parse_session(os.path.join(root, fn))
|
|
if info:
|
|
sessions.append(info)
|
|
sessions.sort(key=lambda s: s["mtime"], reverse=True)
|
|
return sessions
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Formattazione
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def fmt_relative(ts: float) -> str:
|
|
d = time.time() - ts
|
|
if d < 0:
|
|
return "adesso"
|
|
if d < 60:
|
|
return f"{int(d)}s fa"
|
|
if d < 3600:
|
|
return f"{int(d // 60)}min fa"
|
|
if d < 86400:
|
|
return f"{int(d // 3600)}h fa"
|
|
if d < 172800:
|
|
return "ieri"
|
|
return f"{int(d // 86400)}g fa"
|
|
|
|
|
|
def fmt_absolute(ts: float) -> str:
|
|
return time.strftime("%d/%m/%Y %H:%M", time.localtime(ts))
|
|
|
|
|
|
def fmt_size(n: int) -> str:
|
|
if n < 1024:
|
|
return f"{n} B"
|
|
if n < 1024 * 1024:
|
|
return f"{n / 1024:.0f} KB"
|
|
return f"{n / (1024 * 1024):.1f} MB"
|
|
|
|
|
|
def display_text(info: dict, limit: int = 100) -> str:
|
|
"""Testo principale mostrato per la sessione: nome se presente, altrimenti primo messaggio."""
|
|
if info["name"]:
|
|
return info["name"]
|
|
first = info["first"].replace("\n", " ")
|
|
if len(first) > limit:
|
|
first = first[: limit - 1] + "…"
|
|
return first or "(sessione senza messaggi)"
|
|
|
|
|
|
def session_markup(info: dict) -> str:
|
|
"""Titolo leggibile, con anteprima secondaria per le sessioni rinominate."""
|
|
title = GLib.markup_escape_text(display_text(info))
|
|
if not info["name"]:
|
|
return title
|
|
preview = info["first"].replace("\n", " ").strip()
|
|
if not preview or preview == info["name"]:
|
|
return f"<b>{title}</b>"
|
|
if len(preview) > 96:
|
|
preview = preview[:95] + "…"
|
|
return (
|
|
f"<b>{title}</b>\n"
|
|
f"<span size='small' foreground='#9aa0aa'>{GLib.markup_escape_text(preview)}</span>"
|
|
)
|
|
|
|
|
|
def icon_path():
|
|
"""Percorso dell'icona: dati bundlati (PyInstaller) o cartella assets del sorgente."""
|
|
base = getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__)))
|
|
for rel in ("assets/pi-session-launcher.png", "pi-session-launcher.png"):
|
|
p = os.path.join(base, rel)
|
|
if os.path.isfile(p):
|
|
return p
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GUI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class LauncherWindow(Gtk.ApplicationWindow):
|
|
def __init__(self, app):
|
|
super().__init__(application=app, title=APP_TITLE)
|
|
self.set_default_size(1000, 600)
|
|
icon = icon_path()
|
|
if icon:
|
|
try:
|
|
self.set_icon_from_file(icon)
|
|
except Exception:
|
|
pass
|
|
|
|
# modello: markup_sessione, cwd, rel_time, abs_time, mtime(float), path, size(int), search_text
|
|
self.store = Gtk.ListStore(str, str, str, str, float, str, int, str)
|
|
self.by_path = {}
|
|
|
|
self.search_entry = Gtk.SearchEntry(
|
|
placeholder_text="Cerca per nome, messaggio o cartella…"
|
|
)
|
|
self.search_entry.set_tooltip_text("Filtra le sessioni mentre scrivi")
|
|
self.search_entry.connect("search-changed", self._on_search_changed)
|
|
|
|
# vista lista: righe ariose, griglia orizzontale e metadati attenuati.
|
|
self.filter = self.store.filter_new()
|
|
self.filter.set_visible_func(self._filter_visible)
|
|
self.tree = Gtk.TreeView(model=self.filter)
|
|
self.tree.set_search_column(0)
|
|
self.tree.set_headers_clickable(True)
|
|
self.tree.set_grid_lines(Gtk.TreeViewGridLines.HORIZONTAL)
|
|
|
|
rend = Gtk.CellRendererText(ellipsize="end", ypad=4)
|
|
col = Gtk.TreeViewColumn("Sessione", rend, markup=0)
|
|
col.set_expand(True)
|
|
col.set_min_width(470)
|
|
col.set_sort_column_id(7)
|
|
self.tree.append_column(col)
|
|
|
|
rend2 = Gtk.CellRendererText(ellipsize="middle", ypad=4)
|
|
rend2.set_property("foreground", "#9aa0aa")
|
|
col2 = Gtk.TreeViewColumn("Cartella di lavoro", rend2, text=1)
|
|
col2.set_expand(True)
|
|
col2.set_min_width(280)
|
|
col2.set_sort_column_id(1)
|
|
self.tree.append_column(col2)
|
|
|
|
rend3 = Gtk.CellRendererText(ypad=4)
|
|
rend3.set_property("foreground", "#9aa0aa")
|
|
col3 = Gtk.TreeViewColumn("Modificata", rend3, text=2)
|
|
col3.set_sort_column_id(4)
|
|
col3.set_alignment(1.0)
|
|
self.tree.append_column(col3)
|
|
|
|
selection = self.tree.get_selection()
|
|
selection.set_mode(Gtk.SelectionMode.SINGLE)
|
|
selection.connect("changed", lambda *_: self._update_status())
|
|
self.tree.connect("row-activated", self._on_row_activated)
|
|
self.tree.connect("key-press-event", self._on_key_press)
|
|
|
|
scrolled = Gtk.ScrolledWindow()
|
|
scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
|
|
scrolled.add(self.tree)
|
|
|
|
# toolbar: azioni secondarie a sinistra, ripresa evidenziata a destra.
|
|
def button(label, icon_name, tooltip):
|
|
b = Gtk.Button(label=label, tooltip_text=tooltip)
|
|
b.set_image(Gtk.Image.new_from_icon_name(icon_name, Gtk.IconSize.BUTTON))
|
|
b.set_always_show_image(True)
|
|
return b
|
|
|
|
btn_refresh = button("Aggiorna", "view-refresh-symbolic", "Ricarica l'elenco (Ctrl+R)")
|
|
btn_refresh.connect("clicked", lambda *_: self.refresh())
|
|
btn_open_dir = button("Apri cartella", "folder-open-symbolic", "Apre la cartella della sessione in Dolphin")
|
|
btn_open_dir.connect("clicked", lambda *_: self.open_folder())
|
|
btn_rename = button("Rinomina", "edit-rename-symbolic", "Rinomina la sessione selezionata (F2)")
|
|
btn_rename.connect("clicked", lambda *_: self.rename())
|
|
btn_launch = button("Riprendi sessione", "media-playback-start-symbolic", "Apre Konsole nella cartella e riprende la sessione (Invio)")
|
|
btn_launch.get_style_context().add_class("suggested-action")
|
|
btn_launch.connect("clicked", lambda *_: self.launch())
|
|
btn_quit = button("Esci", "application-exit-symbolic", "Chiude il launcher")
|
|
btn_quit.connect("clicked", lambda *_: self.close())
|
|
|
|
hbox = Gtk.Box(spacing=8)
|
|
hbox.pack_start(btn_refresh, False, False, 0)
|
|
hbox.pack_start(btn_open_dir, False, False, 0)
|
|
hbox.pack_start(btn_rename, False, False, 0)
|
|
hbox.pack_start(Gtk.Box(), True, True, 0) # spaziatore
|
|
hbox.pack_start(btn_launch, False, False, 0)
|
|
hbox.pack_start(btn_quit, False, False, 0)
|
|
|
|
# status bar
|
|
self.status = Gtk.Label(label="", xalign=0.0, halign=Gtk.Align.START)
|
|
self.status.set_ellipsize(Pango.EllipsizeMode.END)
|
|
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
|
|
box.set_border_width(8)
|
|
box.pack_start(self.search_entry, False, False, 0)
|
|
box.pack_start(scrolled, True, True, 0)
|
|
box.pack_start(hbox, False, False, 0)
|
|
sep = Gtk.Separator()
|
|
box.pack_start(sep, False, False, 0)
|
|
box.pack_start(self.status, False, False, 0)
|
|
self.add(box)
|
|
|
|
self.refresh()
|
|
self.show_all()
|
|
self.search_entry.grab_focus()
|
|
|
|
# -- popolamento ------------------------------------------------------
|
|
|
|
def refresh(self):
|
|
sessions = scan_sessions()
|
|
selected_path = self.selected_path()
|
|
self.store.clear()
|
|
self.by_path.clear()
|
|
for s in sessions:
|
|
text = display_text(s)
|
|
self.store.append([session_markup(s), s["cwd"], fmt_relative(s["mtime"]),
|
|
fmt_absolute(s["mtime"]), s["mtime"], s["path"], s["size"], text.lower()])
|
|
self.by_path[s["path"]] = s
|
|
if selected_path in self.by_path:
|
|
self.select_path(selected_path)
|
|
self._update_status()
|
|
|
|
def _update_status(self):
|
|
n = len(self.store)
|
|
selected = self.selected_path()
|
|
if selected:
|
|
info = self.by_path.get(selected, {})
|
|
label = display_text(info) if info else os.path.basename(selected)
|
|
self.status.set_text(
|
|
f"{n} session{'i' if n != 1 else 'e'} · Selezionata: {label}"
|
|
)
|
|
else:
|
|
self.status.set_text(
|
|
f"{n} session{'i' if n != 1 else 'e'} · {SESSIONS_DIR}"
|
|
)
|
|
|
|
# -- selezione --------------------------------------------------------
|
|
|
|
def selected_iter(self):
|
|
model, it = self.tree.get_selection().get_selected()
|
|
return model, it
|
|
|
|
def selected_path(self):
|
|
model, it = self.selected_iter()
|
|
if it is None:
|
|
return None
|
|
return model[it][5]
|
|
|
|
def select_path(self, path):
|
|
it = self.store.get_iter_first()
|
|
while it is not None:
|
|
if self.store[it][5] == path:
|
|
self.tree.get_selection().select_iter(it)
|
|
self.tree.scroll_to_cell(self.store.get_path(it))
|
|
return
|
|
it = self.store.iter_next(it)
|
|
|
|
# -- filtri -----------------------------------------------------------
|
|
|
|
def _filter_visible(self, model, it, _data=None):
|
|
q = self.search_entry.get_text().strip().lower()
|
|
if not q:
|
|
return True
|
|
return q in model[it][7] or q in model[it][1].lower()
|
|
|
|
def _on_search_changed(self, _entry):
|
|
self.filter.refilter()
|
|
|
|
# -- azioni -----------------------------------------------------------
|
|
|
|
def _on_row_activated(self, _tree, _path, _col):
|
|
self.launch()
|
|
|
|
def _on_key_press(self, _widget, event):
|
|
if event.keyval in (Gdk.KEY_Return, Gdk.KEY_KP_Enter):
|
|
self.launch()
|
|
return True
|
|
if event.keyval == Gdk.KEY_F2:
|
|
self.rename()
|
|
return True
|
|
if event.keyval == Gdk.KEY_r and event.state & Gdk.ModifierType.CONTROL_MASK:
|
|
self.refresh()
|
|
return True
|
|
return False
|
|
|
|
def launch(self):
|
|
path = self.selected_path()
|
|
if not path:
|
|
self._flash_status("Seleziona prima una sessione dalla lista.")
|
|
return
|
|
info = self.by_path.get(path, {})
|
|
cwd = info.get("cwd") or os.path.expanduser("~")
|
|
if not os.path.isdir(cwd):
|
|
cwd = os.path.expanduser("~")
|
|
|
|
# Le app avviate dal menu KDE non ereditano ~/.bashrc: individua Pi e
|
|
# antepone la sua directory a PATH, così il suo shebang usa il Node.js
|
|
# compatibile incluso in ~/.local/share/pi-node/ invece del Node di sistema.
|
|
pi_bin = self._find_pi()
|
|
if not pi_bin:
|
|
self._flash_status("Pi Agent non trovato: installalo oppure aggiungilo al PATH.")
|
|
return
|
|
child_env = os.environ.copy()
|
|
pi_dir = os.path.dirname(pi_bin)
|
|
child_env["PATH"] = pi_dir + os.pathsep + child_env.get("PATH", "")
|
|
|
|
konsole = shutil.which("konsole")
|
|
if konsole:
|
|
# wrapper bash: più affidabile del passaggio diretto di -e a Konsole
|
|
# ("exec bash" mantiene la shell aperta quando Pi esce)
|
|
cmd = [konsole, "--workdir", cwd, "--hold", "--separate",
|
|
"-e", "bash", "-c", '"$1" --session "$2"; exec bash', "_", pi_bin, path]
|
|
else:
|
|
term = shutil.which("xterm")
|
|
if not term:
|
|
self._flash_status("Nessun terminale trovato (installa konsole o xterm).")
|
|
return
|
|
cmd = [term, "-hold", "-e", pi_bin, "--session", path]
|
|
|
|
try:
|
|
subprocess.Popen(cmd, start_new_session=True, env=child_env)
|
|
except OSError as e:
|
|
self._flash_status(f"Impossibile aprire il terminale: {e}")
|
|
return
|
|
self._flash_status(f"Avviato: pi --session {os.path.basename(path)} in {cwd}")
|
|
|
|
@staticmethod
|
|
def _find_pi():
|
|
"""Trova Pi privilegiando il runtime portable con il suo Node.js."""
|
|
candidates = sorted(
|
|
glob.glob(os.path.expanduser("~/.local/share/pi-node/node-*/bin/pi")),
|
|
reverse=True,
|
|
)
|
|
found_on_path = shutil.which("pi")
|
|
if found_on_path:
|
|
candidates.append(found_on_path)
|
|
for candidate in candidates:
|
|
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
|
|
# Conserva il symlink bin/pi: il suo parent contiene anche node.
|
|
return os.path.abspath(candidate)
|
|
return None
|
|
|
|
def rename(self):
|
|
path = self.selected_path()
|
|
if not path:
|
|
self._flash_status("Seleziona prima una sessione dalla lista.")
|
|
return
|
|
info = self.by_path.get(path, {})
|
|
current = info.get("name", "")
|
|
|
|
dialog = Gtk.Dialog(title="Rinomina sessione", transient_for=self, modal=True)
|
|
dialog.add_buttons(Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL,
|
|
Gtk.STOCK_OK, Gtk.ResponseType.OK)
|
|
entry = Gtk.Entry(text=current)
|
|
entry.set_activates_default(True)
|
|
lbl = Gtk.Label(label="Nome mostrato (vuoto = cancella il nome):")
|
|
lbl.set_xalign(0.0)
|
|
box = dialog.get_content_area()
|
|
box.set_border_width(12)
|
|
box.set_spacing(8)
|
|
box.pack_start(lbl, False, False, 0)
|
|
box.pack_start(entry, False, False, 0)
|
|
dialog.set_default_response(Gtk.ResponseType.OK)
|
|
dialog.show_all()
|
|
response = dialog.run()
|
|
name = entry.get_text().strip()
|
|
dialog.destroy()
|
|
|
|
if response != Gtk.ResponseType.OK:
|
|
return
|
|
try:
|
|
rename_session(path, name)
|
|
except OSError as e:
|
|
self._flash_status(f"Errore nella rinomina: {e}")
|
|
return
|
|
self.refresh()
|
|
self._flash_status(f"Sessione rinominata: {name or '(nome cancellato)'}")
|
|
|
|
def open_folder(self):
|
|
path = self.selected_path()
|
|
if not path:
|
|
return
|
|
info = self.by_path.get(path, {})
|
|
cwd = info.get("cwd") or os.path.expanduser("~")
|
|
if not os.path.isdir(cwd):
|
|
cwd = os.path.expanduser("~")
|
|
dolphin = shutil.which("dolphin")
|
|
if dolphin:
|
|
subprocess.Popen([dolphin, cwd], start_new_session=True)
|
|
|
|
def _flash_status(self, text):
|
|
self.status.set_text(text)
|
|
GLib.timeout_add(5000, lambda: self._update_status() or False)
|
|
|
|
|
|
class LauncherApp(Gtk.Application):
|
|
def __init__(self):
|
|
# deve combaciare con il nome del file .desktop (senza estensione)
|
|
super().__init__(application_id=APP_ID)
|
|
self.win = None
|
|
|
|
def do_activate(self):
|
|
if self.win is None:
|
|
self.win = LauncherWindow(self)
|
|
self.win.present()
|
|
|
|
|
|
APP_ID = "it.enne2.pi-session-launcher"
|
|
|
|
|
|
def main():
|
|
# KDE Plasma/Wayland: l'icona di finestra viene associata via app-id
|
|
# + file .desktop con lo stesso nome stem (vedi ricerca Perplexity).
|
|
# GLib richiede un punto nell'application_id (reverse-DNS).
|
|
GLib.set_prgname(APP_ID)
|
|
Gtk.Window.set_default_icon_name(APP_ID)
|
|
app = LauncherApp()
|
|
app.run(sys.argv)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|