"""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}, ]