Plugin sms: lettura SMS/iMessage da Messaggi, invio con scelta del servizio, fonte per il monitor

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

plugins/messaggi.py

@@ -8,8 +8,13 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from avatar.macos import osa, confirm_or_param, CONFERMATO  # noqa: E402
 
 SEND = '''on run argv
+  set kind to item 3 of argv
   tell application "Messages"
-    set svc to 1st account whose service type = iMessage
+    if kind is "sms" then
+      set svc to 1st account whose service type = SMS
+    else
+      set svc to 1st account whose service type = iMessage
+    end if
     set b to participant (item 1 of argv) of svc
     send (item 2 of argv) to b
   end tell
@@ -25,9 +30,20 @@ def invia(params: dict, ctx: dict) -> str:
     if "@" not in a and not a.replace("+", "").replace(" ", "").isdigit():
         return "Il destinatario deve essere un numero di telefono o un'email: cerca il contatto con contatti_cerca."
 
+    servizio = str(params.get("servizio", "auto")).lower()
+
     def do() -> str:
-        osa(SEND, a.replace(" ", ""), testo, timeout=30)
-        msg = f"Messaggio inviato a {a}."
+        dest = a.replace(" ", "")
+        try:
+            osa(SEND, dest, testo, "sms" if servizio == "sms" else "imessage", timeout=30)
+            via = "SMS" if servizio == "sms" else "iMessage"
+        except Exception as err:
+            if servizio == "auto" and "@" not in dest:
+                osa(SEND, dest, testo, "sms", timeout=30)
+                via = "SMS"
+            else:
+                raise
+        msg = f"Messaggio inviato a {a} via {via}."
         if ctx.get("say"):
             ctx["say"](msg)
         return msg
@@ -36,6 +52,6 @@ def invia(params: dict, ctx: dict) -> str:
 
 
 TOOLS = [
-    {"name": "messaggi_invia", "description": "Invia un iMessage a un numero di telefono (con prefisso) o a un'email. Azione irreversibile con conferma sullo schermo; con [CONFIRMATION_PENDING] di' all'utente di confermare e non dire che è inviato. Per i nomi usa prima contatti_cerca.",
-     "parameters": {"type": "object", "properties": {"a": {"type": "string"}, "testo": {"type": "string"}, "confermato": CONFERMATO}, "required": ["a", "testo"]}, "run": invia},
+    {"name": "messaggi_invia", "description": "Invia un iMessage o un SMS a un numero di telefono (con prefisso) o a un'email. Azione irreversibile con conferma sullo schermo; con [CONFIRMATION_PENDING] di' all'utente di confermare e non dire che è inviato. Per i nomi usa prima contatti_cerca.",
+     "parameters": {"type": "object", "properties": {"a": {"type": "string"}, "testo": {"type": "string"}, "servizio": {"type": "string", "enum": ["auto", "imessage", "sms"], "description": "auto = iMessage con ripiego su SMS."}, "confermato": CONFERMATO}, "required": ["a", "testo"]}, "run": invia},
 ]

plugins/sms.py

@@ -0,0 +1,115 @@
+"""SMS e iMessage ricevuti sul Mac (app Messaggi): non letti, conversazioni, ricerca."""
+from __future__ import annotations
+
+import re
+import sqlite3
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from avatar.phonebook import name_for  # noqa: E402
+
+DB = Path.home() / "Library" / "Messages" / "chat.db"
+EPOCH = datetime(2001, 1, 1)
+
+
+def _ts(raw) -> datetime:
+    if raw is None:
+        return EPOCH
+    raw = int(raw)
+    if raw > 10**12:          # nanosecondi
+        raw //= 10**9
+    return EPOCH + timedelta(seconds=raw)
+
+
+def decode_body(blob: bytes | None) -> str:
+    """Estrae il testo da attributedBody (archivio NSAttributedString)."""
+    if not blob:
+        return ""
+    i = blob.find(b"NSString")
+    if i < 0:
+        return ""
+    j = blob.find(b"+", i)
+    if j < 0:
+        return ""
+    k = j + 1
+    length = blob[k]
+    k += 1
+    if length == 0x81:
+        length = int.from_bytes(blob[k:k + 2], "little"); k += 2
+    elif length == 0x82:
+        length = int.from_bytes(blob[k:k + 3], "little"); k += 3
+    return blob[k:k + length].decode("utf-8", errors="replace")
+
+
+def _query(where: str, params: tuple, limit: int):
+    if not DB.exists():
+        raise RuntimeError("Database di Messaggi non trovato.")
+    try:
+        con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True, timeout=5)
+    except sqlite3.OperationalError as err:
+        raise RuntimeError("Non posso leggere Messaggi: concedi «Accesso completo al disco» all'app in Impostazioni di Sistema > Privacy e sicurezza.") from err
+    try:
+        rows = con.execute(f"""SELECT m.ROWID, m.text, m.attributedBody, m.date, m.is_from_me, m.is_read, m.service, COALESCE(h.id, ''), COALESCE(m.cache_roomnames, '')
+            FROM message m LEFT JOIN handle h ON h.ROWID = m.handle_id WHERE 1=1 {where} ORDER BY m.date DESC LIMIT ?""", (*params, limit)).fetchall()
+    except sqlite3.OperationalError as err:
+        if "unable to open" in str(err) or "authorization" in str(err):
+            raise RuntimeError("Non posso leggere Messaggi: concedi «Accesso completo al disco» all'app in Impostazioni di Sistema > Privacy e sicurezza.") from err
+        raise
+    finally:
+        con.close()
+    out = []
+    for rid, text, body, date, from_me, is_read, service, handle, room in rows:
+        t = (text or "").strip() or decode_body(body).strip()
+        if not t:
+            continue
+        out.append({"id": rid, "testo": t, "data": _ts(date), "mio": bool(from_me), "letto": bool(is_read), "servizio": service or "?",
+                    "chi": name_for(handle), "handle": handle, "gruppo": bool(room)})
+    return out
+
+
+def _fmt(rows) -> str:
+    return "\n".join(f"- [{r['data']:%d/%m %H:%M}] {'io → ' + r['chi'] if r['mio'] else r['chi']} ({r['servizio']}): {r['testo'][:220]}" for r in rows)
+
+
+def non_letti(params: dict, ctx: dict) -> str:
+    lim = max(1, min(int(params.get("limite") or 15), 50))
+    rows = _query("AND m.is_from_me = 0 AND m.is_read = 0", (), lim)
+    return f"Messaggi non letti ({len(rows)}):\n{_fmt(rows)}" if rows else "Nessun SMS o iMessage non letto."
+
+
+def recenti(params: dict, ctx: dict) -> str:
+    lim = max(1, min(int(params.get("limite") or 15), 60))
+    chi = str(params.get("contatto", "")).strip()
+    ore = int(params.get("ore") or 0)
+    where, args = "", []
+    if ore:
+        since = int((datetime.now() - timedelta(hours=ore) - EPOCH).total_seconds()) * 10**9
+        where += " AND m.date >= ?"; args.append(since)
+    rows = _query(where, tuple(args), 400 if chi else lim)
+    if chi:
+        q = chi.lower()
+        rows = [r for r in rows if q in r["chi"].lower() or q in r["handle"].lower()][:lim]
+    rows = rows[:lim]
+    if not rows:
+        return "Nessun messaggio trovato."
+    return f"Messaggi{' con ' + chi if chi else ''} (dal più recente):\n{_fmt(rows)}"
+
+
+def cerca(params: dict, ctx: dict) -> str:
+    q = str(params.get("query", "")).strip().lower()
+    if not q:
+        return "Errore: serve la query."
+    rows = [r for r in _query("", (), 3000) if q in r["testo"].lower()][:20]
+    return f"Messaggi con '{q}':\n{_fmt(rows)}" if rows else f"Nessun messaggio contiene '{q}'."
+
+
+TOOLS = [
+    {"name": "sms_non_letti", "description": "SMS e iMessage non letti ricevuti sul Mac (app Messaggi), con mittente risolto dalla rubrica.",
+     "parameters": {"type": "object", "properties": {"limite": {"type": "integer"}}}, "run": non_letti},
+    {"name": "sms_recenti", "description": "Ultimi SMS/iMessage, di tutti o di un contatto (nome o numero), opzionalmente nelle ultime N ore.",
+     "parameters": {"type": "object", "properties": {"contatto": {"type": "string"}, "limite": {"type": "integer"}, "ore": {"type": "integer"}}}, "run": recenti},
+    {"name": "sms_cerca", "description": "Cerca una parola negli SMS/iMessage.",
+     "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}, "run": cerca},
+]