From 140a09efbfb4e7f49f531fe4de82e88d07dd574c Mon Sep 17 00:00:00 2001 From: luciano Date: Tue, 29 Sep 2026 17:44:35 +0200 Subject: [PATCH] Rispondere come l'utente (stile dai suoi messaggi WhatsApp/Telegram) e stima dell'umore oraria opzionale Co-Authored-By: Claude Fable 5.1 --- avatar/memory_tools.py | 13 +- avatar/monitor.py | 11 ++ avatar/persona.md | 5 + avatar/settings.py | 2 + avatar/settings_dialog.py | 5 + plugins/come_me.py | 298 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 333 insertions(+), 1 deletion(-) create mode 100644 plugins/come_me.py diff --git a/avatar/memory_tools.py b/avatar/memory_tools.py index 9dbb4cc..475c8ea 100644 --- a/avatar/memory_tools.py +++ b/avatar/memory_tools.py @@ -119,7 +119,18 @@ def memory_prompt() -> str: block = mm.format_memory_for_prompt(mm.load_memory()) except Exception: block = "" - return block or "Nessuna memoria salvata finora." + block = block or "Nessuna memoria salvata finora." + try: + import sys as _s + from avatar.settings import BASE_DIR as _B + _s.path.insert(0, str(_B / "plugins")) + import importlib + riga = importlib.import_module("come_me").riga_prompt() + if riga: + block += "\n\n" + riga + except Exception: + pass + return block def parse_args(raw) -> dict: diff --git a/avatar/monitor.py b/avatar/monitor.py index fea1c2d..6022b7d 100644 --- a/avatar/monitor.py +++ b/avatar/monitor.py @@ -56,12 +56,23 @@ class Monitor: def _loop(self) -> None: time.sleep(20) + last_mood = 0.0 while not self._stop.is_set(): if self.settings.get("monitor_enabled"): try: self.scan() except Exception as err: print(f"[monitor] {err}") + if self.settings.get("umore_enabled") and time.time() - last_mood > 3600: + last_mood = time.time() + try: + import sys as _s; _s.path.insert(0, str(Path(__file__).resolve().parent.parent / "plugins")) + import importlib + r = importlib.import_module("come_me").analizza_umore(6) + if r.get("tono"): + self.ui.write_log(f"SYS: Umore stimato: {r['tono']} ({r.get('nota', '')[:80]})") + except Exception as err: + print(f"[umore] {err}") self._stop.wait(int(self.settings.get("monitor_intervallo") or 60)) # ── raccolta ───────────────────────────────────────────────────────── diff --git a/avatar/persona.md b/avatar/persona.md index fcb3340..2b5c19d 100644 --- a/avatar/persona.md +++ b/avatar/persona.md @@ -77,3 +77,8 @@ Sei **Ava**, l'assistente personale di chi ti parla. Vivi in un'app sul suo Mac ## Newsletter - Per "disiscrivimi", "troppa pubblicità", "posta indesiderata": prima mail_newsletter, poi leggi all'utente i mittenti principali e chiedi da quali disiscriversi (o tutti); esegui con mail_disiscrivi solo dopo la conferma sullo schermo. Per i mittenti senza disiscrizione standard proponi mail_blocca invece di cliccare link sospetti: con lo spam vero rispondere conferma che l'indirizzo è attivo. + +## Come me + +- Se l'utente ti chiede di rispondere a qualcuno al posto suo ("rispondi a Valentina come me", "digli che arrivo tardi, come lo direi io"), usa come_me_contesto, scrivi il messaggio in prima persona imitando il suo stile (parole, lunghezza, emoji, tono con quella persona) e leggiglielo prima di inviarlo con lo strumento di invio, che chiede conferma. Mai inviare senza conferma. +- L'umore stimato dai suoi messaggi serve solo a calibrare il tuo tono: non commentarlo se non te lo chiede, e se noti un momento pesante chiedi come sta, con tatto. diff --git a/avatar/settings.py b/avatar/settings.py index 7110e83..5da9c9c 100644 --- a/avatar/settings.py +++ b/avatar/settings.py @@ -54,6 +54,8 @@ DEFAULTS: dict[str, Any] = { "homeassistant_url": "http://homeassistant.local:8123", # Home Assistant (token nel portachiavi) "immagini_famiglia": "z-image-turbo", # z-image-turbo | schnell | dev | qwen (comando mflux e passi predefiniti) "immagini_modello": "", # repo Hugging Face o cartella; vuoto = predefinito della famiglia + "umore_enabled": False, # stima oraria dell'umore dai messaggi scritti dall'utente + "io_nomi": "Luciano", # come compare l'utente come mittente nelle chat esportate "remote_enabled": True, # app web per il telefono (HTTPS sulla rete locale) "remote_token": "", # chiave di abbinamento (generata al primo avvio) } diff --git a/avatar/settings_dialog.py b/avatar/settings_dialog.py index 2d19826..18a8312 100644 --- a/avatar/settings_dialog.py +++ b/avatar/settings_dialog.py @@ -253,6 +253,10 @@ class SettingsDialog(QDialog): form5.addRow("Cosa merita un avviso", self.mon_rules) self.mon_excl = QLineEdit(s.get("monitor_escludi") or ""); self.mon_excl.setPlaceholderText("es. newsletter, offerta, gruppo Calcetto, Amazon — parole separate da virgola, cercate in mittente, chat e testo") form5.addRow("Escludi dagli avvisi", self.mon_excl) + self.umore_on = QCheckBox("Stima il mio umore ogni ora dai messaggi che scrivo (WhatsApp, Telegram) e adatta il tono; dati solo sul Mac") + self.umore_on.setChecked(bool(s.get("umore_enabled"))); form5.addRow("Umore", self.umore_on) + self.io_nomi = QLineEdit(str(s.get("io_nomi") or "Luciano")); self.io_nomi.setPlaceholderText("nomi con cui compaio nelle chat esportate, separati da virgola") + form5.addRow("Il mio nome nelle chat", self.io_nomi) add(pg_mon, form5) # ── Casa (Home Assistant) ───────────────────────────────────────── @@ -583,6 +587,7 @@ class SettingsDialog(QDialog): "monitor_intervallo": int(self.mon_int.currentData() or 60), "monitor_regole": self.mon_rules.text().strip(), "monitor_escludi": self.mon_excl.text().strip(), + "umore_enabled": self.umore_on.isChecked(), "io_nomi": self.io_nomi.text().strip() or "Luciano", "server_assist_ssh": self.srv_ssh.text().strip(), "server_assist_dir": self.srv_dir.text().strip() or "/opt/aiserverassistance", "server_assist_bot": self.srv_bot.text().strip().lstrip("@") or "luzaserver_bot", diff --git a/plugins/come_me.py b/plugins/come_me.py new file mode 100644 index 0000000..a10b465 --- /dev/null +++ b/plugins/come_me.py @@ -0,0 +1,298 @@ +"""Rispondere come l'utente e stimare il suo umore dai messaggi che scrive (WhatsApp e Telegram). + +- stile: profilo del modo di scrivere dell'utente (globale e per contatto) ricavato dai suoi messaggi, con esempi. +- come_me_contesto: coda della conversazione + profilo di stile, perché il modello scriva la risposta in prima persona + come l'utente e la invii con whatsapp_invia / telegram_invia (che chiedono conferma). +- umore: analisi periodica dei messaggi scritti dall'utente (tono, energia, temi) salvata in data/umore.json; + la riga "umore recente" entra nel prompt di LuZa solo se l'opzione è attiva nelle impostazioni. +""" +from __future__ import annotations + +import json +import random +import re +import sqlite3 +import sys +import time +from datetime import datetime, timedelta +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from avatar.settings import DATA_DIR, Settings # noqa: E402 +from avatar import quick_llm # noqa: E402 + +WA_DB = DATA_DIR / "whatsapp" / "index.sqlite" +STILE = DATA_DIR / "stile.json" +UMORE = DATA_DIR / "umore.json" + + +def _names() -> list[str]: + s = Settings() + raw = str(s.get("io_nomi") or "Luciano") + return [n.strip().lower() for n in raw.split(",") if n.strip()] + + +def _load(p: Path) -> dict: + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception: + return {} + + +def _save(p: Path, d: dict) -> None: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps(d, indent=2, ensure_ascii=False), encoding="utf-8") + + +# ── messaggi dell'utente ───────────────────────────────────────────────────── +def _chat_names(query: str) -> list[str]: + """Nomi di chat reali che corrispondono a uno dei nomi in `query` (separati da |), confrontati senza emoji/punteggiatura.""" + if not WA_DB.exists(): + return [] + keys = [re.sub(r"[^a-z0-9]", "", q.lower()) for q in query.split("|") if q.strip()] + keys = [k for k in keys if k] + con = sqlite3.connect(f"file:{WA_DB}?mode=ro", uri=True, timeout=5) + try: + names = [r[0] for r in con.execute("SELECT DISTINCT chat FROM messages WHERE chat IS NOT NULL").fetchall()] + finally: + con.close() + out = [] + for n in names: + nk = re.sub(r"[^a-z0-9]", "", n.lower()) + if any(k == nk or (len(k) >= 3 and k in nk) for k in keys): + out.append(n) + return out + + +def _wa_mine(chat: str = "", since: float | None = None, limit: int = 400) -> list[dict]: + if not WA_DB.exists(): + return [] + names = _names() + con = sqlite3.connect(f"file:{WA_DB}?mode=ro", uri=True, timeout=5) + try: + q = "SELECT chat, ts, sender, text, from_me FROM messages WHERE text != '' AND (from_me = 1 OR lower(sender) IN (%s))" % ",".join("?" * len(names)) + args: list = list(names) + if chat: + names_chat = _chat_names(chat) + if not names_chat: + return [] + q += " AND chat IN (" + ",".join("?" * len(names_chat)) + ")"; args += names_chat + if since: + q += " AND ts >= ?"; args.append(datetime.fromtimestamp(since).strftime("%Y-%m-%d %H:%M")) + q += " ORDER BY ts DESC LIMIT ?"; args.append(limit) + rows = con.execute(q, args).fetchall() + finally: + con.close() + return [{"canale": "whatsapp", "chat": c or "", "ts": ts or "", "testo": t} for c, ts, s, t, fm in rows if not str(t).startswith(("<", "[")) and len(str(t)) > 1] + + +def _wa_tail(chat: str, n: int = 20) -> list[dict]: + if not WA_DB.exists(): + return [] + names = _names() + con = sqlite3.connect(f"file:{WA_DB}?mode=ro", uri=True, timeout=5) + try: + names_chat = _chat_names(chat) + if not names_chat: + return [] + rows = con.execute("SELECT chat, ts, sender, text, from_me FROM messages WHERE chat IN (" + ",".join("?" * len(names_chat)) + ") AND text != '' ORDER BY ts DESC LIMIT ?", (*names_chat, n)).fetchall() + finally: + con.close() + out = [] + for c, ts, s, t, fm in reversed(rows): + mine = bool(fm) or (s or "").lower() in names + out.append({"chi": "io" if mine else (s or c), "ts": ts, "testo": t}) + return out + + +def _tg_mine(since: float | None, limit_dialogs: int = 25) -> list[dict]: + try: + from avatar import telegram_client as tg + except Exception: + return [] + cutoff = datetime.fromtimestamp(since) if since else None + + async def go(client): + out = [] + async for d in client.iter_dialogs(limit=limit_dialogs): + if d.is_channel and not d.is_group: + continue + async for m in client.iter_messages(d.entity, limit=40): + if not m.out or not (m.message or "").strip(): + continue + if cutoff and m.date.replace(tzinfo=None) < cutoff - timedelta(hours=2): + break + out.append({"canale": "telegram", "chat": d.name or "", "ts": m.date.strftime("%Y-%m-%d %H:%M"), "testo": m.message}) + return out + try: + return tg.run(go) + except Exception: + return [] + + +def _tg_tail(chat: str, n: int = 20) -> list[dict]: + try: + from avatar import telegram_client as tg + import plugins.telegram as tplug # noqa + except Exception: + tplug = None + + async def go(client): + d = None + name_l = chat.lower() + async for x in client.iter_dialogs(limit=300): + if name_l in (x.name or "").lower(): + d = x; break + if d is None: + return [] + out = [] + async for m in client.iter_messages(d.entity, limit=n): + if (m.message or "").strip(): + out.append({"chi": "io" if m.out else d.name, "ts": m.date.strftime("%Y-%m-%d %H:%M"), "testo": m.message}) + return list(reversed(out)) + try: + from avatar import telegram_client as tg + return tg.run(go) + except Exception as err: + return [{"chi": "errore", "ts": "", "testo": str(err)}] + + +# ── stile ──────────────────────────────────────────────────────────────────── +def _profilo_stile(msgs: list[dict], contatto: str = "") -> dict: + sample = random.sample(msgs, min(len(msgs), 120)) + testo = "\n".join(f"- {m['testo'][:200]}" for m in sample) + prompt = (f"Questi sono messaggi scritti da Luciano{' a ' + contatto if contatto else ' a varie persone'} su WhatsApp/Telegram:\n{testo}\n\n" + "Descrivi in italiano, in 6-10 righe, il suo modo di scrivere: registro (tu/lei, formale/informale), lunghezza tipica, punteggiatura e maiuscole, " + "uso di emoji e faccine, parole ed espressioni ricorrenti, dialetto o intercalari, come saluta e come chiude, tono abituale" + + (", come chiama questa persona e che rapporto traspare" if contatto else "") + ". Solo la descrizione, niente premesse.") + desc = quick_llm.ask(prompt, system="Sei un linguista che descrive lo stile di scrittura di una persona.", max_tokens=500) + esempi = [m["testo"][:160] for m in random.sample(msgs, min(len(msgs), 25))] + return {"descrizione": desc.strip(), "esempi": esempi, "n": len(msgs), "quando": time.strftime("%Y-%m-%d")} + + +def stile_impara(params: dict, ctx: dict) -> str: + contatto = str(params.get("contatto") or "").strip() + alias = str(params.get("alias") or "").strip() + if ctx.get("player") is not None and hasattr(ctx["player"], "set_state"): + ctx["player"].set_state("PROCESSING · studio il tuo stile") + msgs = _wa_mine("|".join(x for x in (contatto, alias) if x), limit=600) + if not contatto: + msgs += _tg_mine(None) + if len(msgs) < 15: + return f"Troppo pochi messaggi tuoi {('a ' + contatto) if contatto else ''} ({len(msgs)}): servono almeno 15 per ricavare uno stile." + prof = _profilo_stile(msgs, contatto) + if alias: + prof["alias"] = alias + d = _load(STILE); d[contatto.lower() or "_globale"] = prof; _save(STILE, d) + return f"Stile {'con ' + contatto if contatto else 'generale'} appreso da {len(msgs)} messaggi:\n{prof['descrizione']}" + + +def come_me_contesto(params: dict, ctx: dict) -> str: + chat = str(params.get("chat") or "").strip() + canale = str(params.get("canale") or "whatsapp").lower() + if not chat: + return "Errore: serve il nome della chat o del contatto." + d = _load(STILE) + prof = next((v for k, v in d.items() if k != "_globale" and k in chat.lower()), None) + chat_q = "|".join(x for x in (chat, (prof or {}).get("alias", "")) if x) + tail = _wa_tail(chat_q, 20) if canale == "whatsapp" else _tg_tail(chat, 20) + if not tail: + return f"Non trovo la conversazione '{chat}' su {canale}." + if prof is None: + mine = _wa_mine(chat, limit=300) if canale == "whatsapp" else [] + if len(mine) >= 15: + prof = _profilo_stile(mine, chat); d[chat.lower()] = prof; _save(STILE, d) + glob = d.get("_globale") + if glob is None: + mine = _wa_mine("", limit=600) + _tg_mine(None) + if len(mine) >= 15: + glob = _profilo_stile(mine); d["_globale"] = glob; _save(STILE, d) + stile = (prof or glob or {}).get("descrizione", "stile non ancora appreso: informale, frasi brevi") + esempi = (prof or glob or {}).get("esempi", [])[:12] + conv = "\n".join(f"[{m['ts'][-5:] if m['ts'] else ''}] {m['chi']}: {m['testo'][:300]}" for m in tail) + umore = _load(UMORE).get("ultimo", {}) + out = [f"Conversazione recente con {chat} ({canale}), 'io' = Luciano:", conv, "", + f"STILE DI LUCIANO{' con questa persona' if prof else ''}: {stile}", + "ESEMPI DI SUOI MESSAGGI: " + " | ".join(esempi)] + if umore: + out.append(f"UMORE RECENTE DI LUCIANO: {umore.get('tono', '')} ({umore.get('nota', '')})") + istr = str(params.get("istruzioni") or "").strip() + out.append("ISTRUZIONI: scrivi la risposta come se fossi Luciano, in prima persona, con il suo stile, lunghezza e tono (niente stile da assistente, niente firma)." + + (f" Indicazioni dell'utente: {istr}." if istr else "") + + f" Poi inviala con {'whatsapp_invia' if canale == 'whatsapp' else 'telegram_invia'} a «{chat}»: prima della conferma leggi il testo all'utente.") + return "\n".join(out) + + +# ── umore ──────────────────────────────────────────────────────────────────── +def analizza_umore(ore: int = 24) -> dict: + since = time.time() - ore * 3600 + msgs = _wa_mine("", since=since, limit=300) + _tg_mine(since) + msgs.sort(key=lambda m: m["ts"]) + if len(msgs) < 3: + return {"tono": "", "nota": f"troppo pochi messaggi ({len(msgs)}) nelle ultime {ore} ore", "n": len(msgs), "quando": time.strftime("%Y-%m-%d %H:%M")} + testo = "\n".join(f"[{m['ts'][5:]} {m['canale']} → {m['chat'][:18]}] {m['testo'][:220]}" for m in msgs[-120:]) + prompt = (f"Messaggi scritti da Luciano nelle ultime {ore} ore (destinatari indicati tra parentesi):\n{testo}\n\n" + "Stima il suo stato d'animo. Rispondi SOLO con un JSON: {\"tono\": una o due parole (es. sereno, allegro, stanco, teso, giù, concentrato, irritato), " + "\"energia\": \"bassa|media|alta\", \"nota\": una frase breve con gli indizi (temi, ritmo, orari), \"attenzione\": true se emerge malessere o stress marcato, altrimenti false}") + raw = quick_llm.ask(prompt, system="Analizzi con delicatezza lo stato d'animo dai messaggi. Rispondi solo con il JSON richiesto.", max_tokens=300) + m = re.search(r"\{.*\}", raw, re.S) + try: + d = json.loads(m.group(0)) if m else {} + except Exception: + d = {} + res = {"tono": str(d.get("tono", "")).strip(), "energia": str(d.get("energia", "")), "nota": str(d.get("nota", "")).strip(), + "attenzione": bool(d.get("attenzione")), "n": len(msgs), "quando": time.strftime("%Y-%m-%d %H:%M")} + st = _load(UMORE) + st["ultimo"] = res + giorno = time.strftime("%Y-%m-%d") + st.setdefault("giorni", {})[giorno] = res + st["giorni"] = dict(sorted(st["giorni"].items())[-60:]) + _save(UMORE, st) + return res + + +def umore_analizza(params: dict, ctx: dict) -> str: + ore = max(3, min(int(params.get("ore") or 24), 168)) + if ctx.get("player") is not None and hasattr(ctx["player"], "set_state"): + ctx["player"].set_state("PROCESSING · leggo i tuoi messaggi") + r = analizza_umore(ore) + if not r["tono"]: + return r["nota"] + return f"Umore stimato dalle ultime {ore} ore ({r['n']} messaggi tuoi): {r['tono']}, energia {r['energia']}. {r['nota']}" + (" Sembra un momento pesante: valuta se chiederglielo con tatto." if r["attenzione"] else "") + + +def umore_oggi(params: dict, ctx: dict) -> str: + st = _load(UMORE) + giorni = max(1, min(int(params.get("giorni") or 7), 60)) + if not st.get("giorni"): + return "Nessuna stima dell'umore ancora: usa umore_analizza, oppure attiva l'opzione nelle impostazioni (scheda Monitor) per farla ogni ora." + rows = sorted(st["giorni"].items())[-giorni:] + lines = [f"- {g}: {r.get('tono') or '?'} (energia {r.get('energia') or '?'}) — {r.get('nota', '')}" for g, r in rows] + return f"Umore degli ultimi {len(rows)} giorni (dai messaggi scritti da Luciano):\n" + "\n".join(lines) + + +def riga_prompt() -> str: + """Riga per il prompt di LuZa, solo se l'opzione è attiva e la stima è recente (meno di 12 ore).""" + try: + if not Settings().get("umore_enabled"): + return "" + u = _load(UMORE).get("ultimo") or {} + if not u.get("tono"): + return "" + if time.time() - datetime.strptime(u["quando"], "%Y-%m-%d %H:%M").timestamp() > 12 * 3600: + return "" + return f"Umore recente dell'utente (stimato dai suoi messaggi alle {u['quando'][-5:]}): {u['tono']}, energia {u.get('energia', '?')}. {u.get('nota', '')} Adatta il tono di conseguenza senza citare l'analisi, a meno che non te lo chieda." + except Exception: + return "" + + +TOOLS = [ + {"name": "stile_impara", "description": "Studia il modo di scrivere dell'utente dai suoi messaggi WhatsApp/Telegram (in generale, o con un contatto specifico) e lo memorizza, per poter rispondere come lui.", + "parameters": {"type": "object", "properties": {"contatto": {"type": "string"}, "alias": {"type": "string", "description": "altro nome della stessa chat nell'archivio (es. 'My ❤️' per Valentina)"}}}, "run": stile_impara}, + {"name": "come_me_contesto", "description": "Quando l'utente chiede di rispondere a qualcuno 'come me' / 'al posto mio' su WhatsApp o Telegram: restituisce la conversazione recente con quella persona, lo stile dell'utente ed eventuali istruzioni. Poi scrivi tu la risposta in prima persona come l'utente e inviala con whatsapp_invia o telegram_invia (che chiedono conferma), leggendola prima a voce.", + "parameters": {"type": "object", "properties": {"chat": {"type": "string"}, "canale": {"type": "string", "enum": ["whatsapp", "telegram"]}, "istruzioni": {"type": "string", "description": "cosa vuole dire l'utente, se lo ha specificato"}}, "required": ["chat"]}, "run": come_me_contesto}, + {"name": "umore_analizza", "description": "Stima lo stato d'animo dell'utente dai messaggi che ha scritto su WhatsApp e Telegram nelle ultime ore (default 24): tono, energia, indizi.", + "parameters": {"type": "object", "properties": {"ore": {"type": "integer"}}}, "run": umore_analizza}, + {"name": "umore_oggi", "description": "Diario dell'umore stimato negli ultimi giorni (default 7): per 'come mi vedi oggi?', 'com'è andata la settimana?'.", + "parameters": {"type": "object", "properties": {"giorni": {"type": "integer"}}}, "run": umore_oggi}, +]