AvatarPy: assistente personale con volto 3D, voce, memoria e plugin
Riscrittura in Python dell'assistente Avatar con interfaccia HUD (derivata da Mark LIV, CC BY-NC 4.0, vedi NOTICE.md). Tre motori (Claude API, server locale OpenAI-compatibile, Claude Code), voce Kokoro/macOS, Whisper MLX, avatar 3D con sincronizzazione labiale, memoria per categorie, allegati con OCR, monitor con avvisi, plugin per Calendario, Mail, Promemoria, Note, Musica, app, Mac, timer, meteo, contatti, Messaggi, file, Comandi Rapidi, browser, Telegram, WhatsApp (archivio e tempo reale). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
lucianoandClaude Fable 5.1 committed
commit
ff79832c30
76 files changed
+82468
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,55 @@
|
||||
"""Apre e chiude applicazioni, apre siti e file."""
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import osa # noqa: E402
|
||||
|
||||
|
||||
def apri(params: dict, ctx: dict) -> str:
|
||||
nome = str(params.get("nome", "")).strip()
|
||||
url = str(params.get("url", "")).strip()
|
||||
if url:
|
||||
if not url.startswith(("http://", "https://")):
|
||||
url = "https://" + url
|
||||
cmd = ["open", url] if not nome else ["open", "-a", nome, url]
|
||||
subprocess.run(cmd, check=False, timeout=15)
|
||||
return f"Apro {url}" + (f" con {nome}." if nome else ".")
|
||||
if not nome:
|
||||
return "Errore: serve il nome dell'app o un indirizzo."
|
||||
p = Path(nome).expanduser()
|
||||
if p.exists():
|
||||
subprocess.run(["open", str(p)], check=False, timeout=15)
|
||||
return f"Apro {p.name}."
|
||||
res = subprocess.run(["open", "-a", nome], capture_output=True, text=True, timeout=15)
|
||||
if res.returncode != 0:
|
||||
return f"Non trovo un'app chiamata '{nome}'."
|
||||
return f"Apro {nome}."
|
||||
|
||||
|
||||
def chiudi(params: dict, ctx: dict) -> str:
|
||||
nome = str(params.get("nome", "")).strip()
|
||||
if not nome:
|
||||
return "Errore: serve il nome dell'app."
|
||||
try:
|
||||
osa('on run argv\ntell application (item 1 of argv) to quit\nend run', nome, timeout=20)
|
||||
return f"Chiudo {nome}."
|
||||
except Exception as err:
|
||||
return f"Non riesco a chiudere {nome}: {err}"
|
||||
|
||||
|
||||
def in_esecuzione(params: dict, ctx: dict) -> str:
|
||||
out = osa('tell application "System Events" to get name of every process whose background only is false')
|
||||
return "App aperte: " + out
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "app_apri", "description": "Apre un'applicazione (per nome, es. Safari, Xcode), un sito (url) o un file. Se dai sia app che url, apre l'url con quell'app.",
|
||||
"parameters": {"type": "object", "properties": {"nome": {"type": "string"}, "url": {"type": "string"}}}, "run": apri},
|
||||
{"name": "app_chiudi", "description": "Chiude un'applicazione per nome.",
|
||||
"parameters": {"type": "object", "properties": {"nome": {"type": "string"}}, "required": ["nome"]}, "run": chiudi},
|
||||
{"name": "app_aperte", "description": "Elenca le applicazioni aperte.", "parameters": {"type": "object", "properties": {}}, "run": in_esecuzione},
|
||||
]
|
||||
@@ -0,0 +1,41 @@
|
||||
"""Avvisi del monitor di Mail, WhatsApp e Telegram: elenco, chiusura, controllo immediato."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
|
||||
def _mon():
|
||||
from avatar import monitor as m
|
||||
if m.monitor is None:
|
||||
raise RuntimeError("Il monitor non è attivo in questo processo.")
|
||||
return m.monitor
|
||||
|
||||
|
||||
def elenca(params: dict, ctx: dict) -> str:
|
||||
al = _mon().pending()
|
||||
if not al:
|
||||
return "Nessun avviso aperto."
|
||||
return "Avvisi aperti:\n" + "\n".join(f"- [{a['id']}] {a['quando']} {a['fonte']} · {a['chi']} — {a['motivo']} ({a['priorita']})\n «{a['testo'][:160]}»" for a in al)
|
||||
|
||||
|
||||
def chiudi(params: dict, ctx: dict) -> str:
|
||||
n = _mon().close(str(params.get("id", "")).strip())
|
||||
return f"Chiusi {n} avvisi." if n else "Nessun avviso corrispondente."
|
||||
|
||||
|
||||
def controlla(params: dict, ctx: dict) -> str:
|
||||
al = _mon().scan()
|
||||
return f"Controllo fatto: {len(al)} nuovi avvisi." + ("" if not al else "\n" + "\n".join(f"- {a['fonte']} · {a['chi']} — {a['motivo']}" for a in al))
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "attenzione_elenca", "description": "Elenca gli avvisi aperti del monitor (messaggi Mail, WhatsApp e Telegram che richiedono attenzione).",
|
||||
"parameters": {"type": "object", "properties": {}}, "run": elenca},
|
||||
{"name": "attenzione_chiudi", "description": "Chiude un avviso per id o per nome del mittente; senza parametri chiude tutti.",
|
||||
"parameters": {"type": "object", "properties": {"id": {"type": "string"}}}, "run": chiudi},
|
||||
{"name": "attenzione_controlla", "description": "Esegue subito un controllo di Mail, WhatsApp e Telegram alla ricerca di messaggi che richiedono attenzione.",
|
||||
"parameters": {"type": "object", "properties": {}}, "run": controlla},
|
||||
]
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Browser e web: legge una pagina, la pagina aperta in Safari, cerca su YouTube o Google."""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
import requests
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import osa # noqa: E402
|
||||
|
||||
MAX_CHARS = 6000
|
||||
UA = {"User-Agent": "Mozilla/5.0 (Macintosh) AvatarPy/1.0"}
|
||||
|
||||
|
||||
def _html_to_text(page: str) -> str:
|
||||
page = re.sub(r"<(script|style|noscript|svg|header|footer|nav)[^>]*>.*?</\1>", " ", page, flags=re.S | re.I)
|
||||
page = re.sub(r"<br\s*/?>|</p>|</div>|</h\d>|</li>|</tr>", "\n", page, flags=re.I)
|
||||
text = html.unescape(re.sub(r"<[^>]+>", " ", page))
|
||||
lines = [" ".join(l.split()) for l in text.splitlines()]
|
||||
return "\n".join(l for l in lines if len(l) > 2)
|
||||
|
||||
|
||||
def leggi_pagina(params: dict, ctx: dict) -> str:
|
||||
url = str(params.get("url", "")).strip()
|
||||
if not url:
|
||||
try:
|
||||
url = osa('tell application "Safari" to return URL of front document')
|
||||
except Exception:
|
||||
return "Nessun indirizzo dato e Safari non ha pagine aperte."
|
||||
if not url.startswith("http"):
|
||||
url = "https://" + url
|
||||
r = requests.get(url, headers=UA, timeout=15)
|
||||
r.raise_for_status()
|
||||
title = re.search(r"<title[^>]*>(.*?)</title>", r.text, flags=re.S | re.I)
|
||||
text = _html_to_text(r.text)
|
||||
return f"{html.unescape(title.group(1).strip()) if title else url}\n{text[:MAX_CHARS]}{' […]' if len(text) > MAX_CHARS else ''}"
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
dove = str(params.get("dove", "google")).lower()
|
||||
if not q:
|
||||
return "Errore: serve la query."
|
||||
url = {"youtube": f"https://www.youtube.com/results?search_query={quote_plus(q)}",
|
||||
"maps": f"https://www.google.com/maps/search/{quote_plus(q)}",
|
||||
"amazon": f"https://www.amazon.it/s?k={quote_plus(q)}",
|
||||
"wikipedia": f"https://it.wikipedia.org/w/index.php?search={quote_plus(q)}"}.get(dove, f"https://www.google.com/search?q={quote_plus(q)}")
|
||||
subprocess.run(["open", url], check=False, timeout=10)
|
||||
return f"Apro la ricerca di «{q}» su {dove}."
|
||||
|
||||
|
||||
def youtube(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
if not q:
|
||||
return "Errore: serve cosa cercare."
|
||||
r = requests.get(f"https://www.youtube.com/results?search_query={quote_plus(q)}", headers=UA, timeout=15)
|
||||
m = re.search(r'"videoId":"([\w-]{11})"', r.text)
|
||||
if not m:
|
||||
subprocess.run(["open", f"https://www.youtube.com/results?search_query={quote_plus(q)}"], check=False)
|
||||
return f"Apro i risultati YouTube per «{q}»."
|
||||
subprocess.run(["open", f"https://www.youtube.com/watch?v={m.group(1)}"], check=False)
|
||||
t = re.search(r'"videoId":"' + m.group(1) + r'".{0,400}?"title":\{"runs":\[\{"text":"([^"]+)"', r.text)
|
||||
return f"Riproduco su YouTube: {html.unescape(t.group(1)) if t else q}."
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "web_leggi_pagina", "description": "Legge il testo di una pagina web (dato l'url) o della pagina aperta in Safari, per riassumerla o rispondere.",
|
||||
"parameters": {"type": "object", "properties": {"url": {"type": "string"}}}, "run": leggi_pagina},
|
||||
{"name": "web_cerca_apri", "description": "Apre nel browser una ricerca su Google, YouTube, Maps, Amazon o Wikipedia.",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}, "dove": {"type": "string", "enum": ["google", "youtube", "maps", "amazon", "wikipedia"]}}, "required": ["query"]}, "run": cerca},
|
||||
{"name": "youtube_riproduci", "description": "Cerca un video su YouTube e apre direttamente il primo risultato.",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}, "run": youtube},
|
||||
]
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Riepilogo della giornata: calendario, promemoria, email non lette e meteo in un colpo solo."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
|
||||
def riepilogo(params: dict, ctx: dict) -> str:
|
||||
from avatar.plugins import registry
|
||||
citta = str(params.get("citta", "")).strip() or "Roma"
|
||||
parts = []
|
||||
for label, name, args in (
|
||||
("Calendario", "calendario_eventi", {"da": "oggi", "a": "domani"}),
|
||||
("Promemoria", "promemoria_elenca", {"quando": "oggi"}),
|
||||
("Email", "mail_non_lette", {"limite": 5}),
|
||||
("Meteo", "meteo", {"citta": citta, "giorni": 1}),
|
||||
):
|
||||
if not registry.has(name):
|
||||
continue
|
||||
try:
|
||||
parts.append(f"## {label}\n{registry.run(name, args)}")
|
||||
except Exception as err:
|
||||
parts.append(f"## {label}\nnon disponibile ({err})")
|
||||
return ("\n\n".join(parts) + "\n\nFai un riepilogo parlato breve e naturale: prima gli impegni, poi i promemoria, "
|
||||
"poi le email importanti (solo mittente e oggetto), infine il meteo in una frase.")
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "buongiorno", "description": "Riepilogo della giornata: impegni di oggi e domani, promemoria, email non lette e meteo. Usalo per 'buongiorno', 'come si presenta la giornata', 'riepilogo'.",
|
||||
"parameters": {"type": "object", "properties": {"citta": {"type": "string", "description": "Città per il meteo (default Roma)."}}}, "run": riepilogo},
|
||||
]
|
||||
@@ -0,0 +1,227 @@
|
||||
"""Calendario di macOS: legge, crea, sposta e cancella eventi (EventKit)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from datetime import date, datetime, timedelta
|
||||
|
||||
_store = None
|
||||
_lock = threading.Lock()
|
||||
GIORNI = ["lun", "mar", "mer", "gio", "ven", "sab", "dom"]
|
||||
MESI = ["gen", "feb", "mar", "apr", "mag", "giu", "lug", "ago", "set", "ott", "nov", "dic"]
|
||||
|
||||
|
||||
def _get_store():
|
||||
"""EKEventStore con accesso richiesto una sola volta."""
|
||||
global _store
|
||||
with _lock:
|
||||
if _store is not None:
|
||||
return _store
|
||||
import EventKit
|
||||
store = EventKit.EKEventStore.alloc().init()
|
||||
done, res = threading.Event(), {}
|
||||
|
||||
def cb(granted, err):
|
||||
res["granted"] = bool(granted)
|
||||
done.set()
|
||||
|
||||
if hasattr(store, "requestFullAccessToEventsWithCompletion_"):
|
||||
store.requestFullAccessToEventsWithCompletion_(cb)
|
||||
else:
|
||||
store.requestAccessToEntityType_completion_(EventKit.EKEntityTypeEvent, cb)
|
||||
done.wait(60)
|
||||
if not res.get("granted"):
|
||||
raise RuntimeError("Accesso al Calendario negato: concedilo in Impostazioni di Sistema > Privacy e sicurezza > Calendari.")
|
||||
_store = store
|
||||
return store
|
||||
|
||||
|
||||
def _nsdate(dt: datetime):
|
||||
from Foundation import NSDate
|
||||
return NSDate.dateWithTimeIntervalSince1970_(dt.timestamp())
|
||||
|
||||
|
||||
def _pydate(nsdate) -> datetime:
|
||||
return datetime.fromtimestamp(nsdate.timeIntervalSince1970())
|
||||
|
||||
|
||||
def _parse_day(s: str) -> date:
|
||||
s = (s or "").strip().lower()
|
||||
today = date.today()
|
||||
if s in ("", "oggi", "today"):
|
||||
return today
|
||||
if s in ("domani", "tomorrow"):
|
||||
return today + timedelta(days=1)
|
||||
if s in ("dopodomani",):
|
||||
return today + timedelta(days=2)
|
||||
if s in ("ieri",):
|
||||
return today - timedelta(days=1)
|
||||
return datetime.strptime(s[:10], "%Y-%m-%d").date()
|
||||
|
||||
|
||||
def _parse_dt(s: str) -> datetime:
|
||||
s = (s or "").strip().replace("T", " ")
|
||||
for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
|
||||
try:
|
||||
return datetime.strptime(s, fmt)
|
||||
except ValueError:
|
||||
continue
|
||||
raise ValueError(f"Data e ora non valide: '{s}'. Usa il formato AAAA-MM-GG HH:MM.")
|
||||
|
||||
|
||||
def _fmt(e) -> str:
|
||||
start, end = _pydate(e.startDate()), _pydate(e.endDate())
|
||||
day = f"{GIORNI[start.weekday()]} {start.day} {MESI[start.month - 1]}"
|
||||
when = "tutto il giorno" if e.isAllDay() else f"{start:%H:%M}–{end:%H:%M}"
|
||||
extra = f" @ {e.location()}" if e.location() else ""
|
||||
return f"{day} {when}: {e.title()} ({e.calendar().title()}){extra}"
|
||||
|
||||
|
||||
def _events_between(start: datetime, end: datetime):
|
||||
store = _get_store()
|
||||
pred = store.predicateForEventsWithStartDate_endDate_calendars_(_nsdate(start), _nsdate(end), None)
|
||||
evs = list(store.eventsMatchingPredicate_(pred) or [])
|
||||
evs.sort(key=lambda e: e.startDate().timeIntervalSince1970())
|
||||
return evs
|
||||
|
||||
|
||||
def eventi(params: dict, ctx: dict) -> str:
|
||||
da = _parse_day(params.get("da", ""))
|
||||
a = _parse_day(params.get("a", "")) if params.get("a") else da
|
||||
if a < da:
|
||||
a = da
|
||||
evs = _events_between(datetime.combine(da, datetime.min.time()), datetime.combine(a + timedelta(days=1), datetime.min.time()))
|
||||
label = f"{da:%d/%m}" if da == a else f"dal {da:%d/%m} al {a:%d/%m}"
|
||||
if not evs:
|
||||
return f"Nessun evento {label}."
|
||||
return f"Eventi {label}:\n" + "\n".join(f"- {_fmt(e)}" for e in evs[:40])
|
||||
|
||||
|
||||
def _find_calendar(store, name: str):
|
||||
import EventKit
|
||||
cals = [c for c in store.calendarsForEntityType_(EventKit.EKEntityTypeEvent) if c.allowsContentModifications()]
|
||||
if name:
|
||||
for c in cals:
|
||||
if c.title().lower() == name.lower():
|
||||
return c
|
||||
for c in cals:
|
||||
if name.lower() in c.title().lower():
|
||||
return c
|
||||
return store.defaultCalendarForNewEvents() or (cals[0] if cals else None)
|
||||
|
||||
|
||||
def crea(params: dict, ctx: dict) -> str:
|
||||
import EventKit
|
||||
store = _get_store()
|
||||
titolo = str(params.get("titolo", "")).strip()
|
||||
if not titolo:
|
||||
return "Errore: serve il titolo dell'evento."
|
||||
start = _parse_dt(str(params.get("inizio", "")))
|
||||
all_day = bool(params.get("tutto_il_giorno", False))
|
||||
if all_day:
|
||||
start = start.replace(hour=0, minute=0)
|
||||
end = start + timedelta(days=1)
|
||||
elif params.get("fine"):
|
||||
end = _parse_dt(str(params["fine"]))
|
||||
else:
|
||||
end = start + timedelta(minutes=int(params.get("durata_minuti") or 60))
|
||||
ev = EventKit.EKEvent.eventWithEventStore_(store)
|
||||
ev.setTitle_(titolo)
|
||||
ev.setStartDate_(_nsdate(start))
|
||||
ev.setEndDate_(_nsdate(end))
|
||||
ev.setAllDay_(all_day)
|
||||
cal = _find_calendar(store, str(params.get("calendario", "")))
|
||||
if cal is None:
|
||||
return "Errore: nessun calendario modificabile disponibile."
|
||||
ev.setCalendar_(cal)
|
||||
if params.get("luogo"):
|
||||
ev.setLocation_(str(params["luogo"]))
|
||||
if params.get("note"):
|
||||
ev.setNotes_(str(params["note"]))
|
||||
ok, err = store.saveEvent_span_error_(ev, EventKit.EKSpanThisEvent, None)
|
||||
if not ok:
|
||||
return f"Errore nel salvataggio: {err}"
|
||||
return f"Evento creato: {_fmt(ev)}"
|
||||
|
||||
|
||||
def _match(params: dict) -> tuple[list, str]:
|
||||
titolo = str(params.get("titolo", "")).strip().lower()
|
||||
giorno = _parse_day(params.get("giorno", ""))
|
||||
evs = _events_between(datetime.combine(giorno, datetime.min.time()), datetime.combine(giorno + timedelta(days=1), datetime.min.time()))
|
||||
found = [e for e in evs if titolo and titolo in (e.title() or "").lower()] if titolo else evs
|
||||
return found, f"{giorno:%d/%m}"
|
||||
|
||||
|
||||
def elimina(params: dict, ctx: dict) -> str:
|
||||
import EventKit
|
||||
found, label = _match(params)
|
||||
if not found:
|
||||
return f"Nessun evento corrispondente il {label}."
|
||||
if len(found) > 1:
|
||||
return "Ho trovato più eventi, dimmi quale:\n" + "\n".join(f"- {_fmt(e)}" for e in found)
|
||||
ev = found[0]
|
||||
desc = _fmt(ev)
|
||||
|
||||
def do() -> str:
|
||||
store = _get_store()
|
||||
ok, err = store.removeEvent_span_error_(ev, EventKit.EKSpanThisEvent, None)
|
||||
msg = f"Evento cancellato: {desc}" if ok else f"Errore nella cancellazione: {err}"
|
||||
if ctx.get("say"):
|
||||
ctx["say"](msg)
|
||||
return msg
|
||||
|
||||
confirm = ctx.get("confirm")
|
||||
if confirm:
|
||||
return confirm("calendario_elimina", "Cancellare l'evento?", desc, do)
|
||||
# Senza interfaccia (server MCP): serve la conferma esplicita nei parametri.
|
||||
if str(params.get("confermato", "")).lower() in ("true", "1", "sì", "si", "yes"):
|
||||
return do()
|
||||
return f"Prima di cancellare chiedi conferma all'utente per: {desc}. Poi richiama con confermato=true."
|
||||
|
||||
|
||||
def sposta(params: dict, ctx: dict) -> str:
|
||||
import EventKit
|
||||
found, label = _match(params)
|
||||
if not found:
|
||||
return f"Nessun evento corrispondente il {label}."
|
||||
if len(found) > 1:
|
||||
return "Ho trovato più eventi, dimmi quale:\n" + "\n".join(f"- {_fmt(e)}" for e in found)
|
||||
ev = found[0]
|
||||
new_start = _parse_dt(str(params.get("nuovo_inizio", "")))
|
||||
duration = _pydate(ev.endDate()) - _pydate(ev.startDate())
|
||||
ev.setStartDate_(_nsdate(new_start))
|
||||
ev.setEndDate_(_nsdate(new_start + duration))
|
||||
ok, err = _get_store().saveEvent_span_error_(ev, EventKit.EKSpanThisEvent, None)
|
||||
return f"Evento spostato: {_fmt(ev)}" if ok else f"Errore: {err}"
|
||||
|
||||
|
||||
DAY = {"type": "string", "description": "Giorno: 'oggi', 'domani', 'dopodomani' oppure AAAA-MM-GG."}
|
||||
TOOLS = [
|
||||
{"name": "calendario_eventi",
|
||||
"description": "Elenca gli eventi del Calendario di macOS in un giorno o in un intervallo. Usalo per domande come 'che impegni ho', 'cosa ho domani', 'sono libero venerdì'.",
|
||||
"parameters": {"type": "object", "properties": {"da": DAY, "a": {**DAY, "description": "Ultimo giorno dell'intervallo (facoltativo)."}}, "required": ["da"]},
|
||||
"run": eventi},
|
||||
{"name": "calendario_crea",
|
||||
"description": "Crea un evento nel Calendario. Converti tu le espressioni come 'domani alle 15' in data e ora assolute usando la data odierna del prompt.",
|
||||
"parameters": {"type": "object", "properties": {
|
||||
"titolo": {"type": "string"},
|
||||
"inizio": {"type": "string", "description": "Inizio, formato AAAA-MM-GG HH:MM."},
|
||||
"fine": {"type": "string", "description": "Fine, formato AAAA-MM-GG HH:MM (facoltativo)."},
|
||||
"durata_minuti": {"type": "integer", "description": "Durata in minuti se manca la fine (default 60)."},
|
||||
"tutto_il_giorno": {"type": "boolean"},
|
||||
"calendario": {"type": "string", "description": "Nome del calendario (facoltativo, altrimenti quello predefinito)."},
|
||||
"luogo": {"type": "string"}, "note": {"type": "string"}},
|
||||
"required": ["titolo", "inizio"]},
|
||||
"run": crea},
|
||||
{"name": "calendario_sposta",
|
||||
"description": "Sposta un evento esistente a un nuovo orario, mantenendone la durata.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string", "description": "Parte del titolo."}, "giorno": DAY,
|
||||
"nuovo_inizio": {"type": "string", "description": "Nuovo inizio, AAAA-MM-GG HH:MM."}},
|
||||
"required": ["titolo", "giorno", "nuovo_inizio"]},
|
||||
"run": sposta},
|
||||
{"name": "calendario_elimina",
|
||||
"description": "Cancella un evento. Azione irreversibile: viene chiesta conferma sullo schermo; se lo strumento risponde con [CONFIRMATION_PENDING], di' all'utente di confermare sul pannello e non dire che è fatto.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string", "description": "Parte del titolo."}, "giorno": DAY,
|
||||
"confermato": {"type": "boolean", "description": "Solo senza interfaccia: true dopo la conferma esplicita dell'utente."}},
|
||||
"required": ["titolo", "giorno"]},
|
||||
"run": elimina},
|
||||
]
|
||||
@@ -0,0 +1,40 @@
|
||||
"""Comandi Rapidi (Shortcuts) di macOS: elenca ed esegue i tuoi comandi, anche per la domotica HomeKit."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import sh # noqa: E402
|
||||
|
||||
|
||||
def elenca(params: dict, ctx: dict) -> str:
|
||||
names = [n for n in sh("shortcuts", "list").splitlines() if n.strip()]
|
||||
q = str(params.get("filtro", "")).strip().lower()
|
||||
if q:
|
||||
names = [n for n in names if q in n.lower()]
|
||||
if not names:
|
||||
return "Nessun comando rapido" + (f" contenente '{q}'." if q else ".")
|
||||
return f"Comandi rapidi ({len(names)}):\n" + "\n".join(f"- {n}" for n in names[:40])
|
||||
|
||||
|
||||
def esegui(params: dict, ctx: dict) -> str:
|
||||
nome = str(params.get("nome", "")).strip()
|
||||
if not nome:
|
||||
return "Errore: serve il nome del comando rapido."
|
||||
names = [n for n in sh("shortcuts", "list").splitlines() if n.strip()]
|
||||
match = next((n for n in names if n.lower() == nome.lower()), None) or next((n for n in names if nome.lower() in n.lower()), None)
|
||||
if not match:
|
||||
return f"Nessun comando rapido chiamato '{nome}'."
|
||||
inp = str(params.get("input", "")).strip()
|
||||
out = sh("shortcuts", "run", match, *(["-i", "-"] if inp else []), timeout=120) if not inp else \
|
||||
__import__("subprocess").run(["shortcuts", "run", match, "-i", "-"], input=inp, capture_output=True, text=True, timeout=120).stdout
|
||||
return f"Eseguito «{match}»." + (f" Risultato: {out.strip()[:500]}" if out and out.strip() else "")
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "comandi_rapidi_elenca", "description": "Elenca i Comandi Rapidi disponibili sul Mac (anche quelli per luci, prese e casa).",
|
||||
"parameters": {"type": "object", "properties": {"filtro": {"type": "string"}}}, "run": elenca},
|
||||
{"name": "comandi_rapidi_esegui", "description": "Esegue un Comando Rapido per nome (es. 'Accendi luci soggiorno'). Usalo per domotica e automazioni create dall'utente.",
|
||||
"parameters": {"type": "object", "properties": {"nome": {"type": "string"}, "input": {"type": "string", "description": "Testo da passare in ingresso (facoltativo)."}}, "required": ["nome"]}, "run": esegui},
|
||||
]
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Contatti di macOS: cerca numero, email e dati di una persona (framework Contacts)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
_store = None
|
||||
_lock = threading.Lock()
|
||||
|
||||
|
||||
def _get_store():
|
||||
global _store
|
||||
with _lock:
|
||||
if _store is not None:
|
||||
return _store
|
||||
import Contacts
|
||||
store = Contacts.CNContactStore.alloc().init()
|
||||
done, res = threading.Event(), {}
|
||||
|
||||
def cb(granted, err):
|
||||
res["granted"] = bool(granted)
|
||||
done.set()
|
||||
|
||||
store.requestAccessForEntityType_completionHandler_(Contacts.CNEntityTypeContacts, cb)
|
||||
done.wait(60)
|
||||
if not res.get("granted"):
|
||||
raise RuntimeError("Accesso ai Contatti negato: concedilo in Impostazioni di Sistema > Privacy e sicurezza > Contatti.")
|
||||
_store = store
|
||||
return store
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
import Contacts
|
||||
q = str(params.get("nome", "")).strip()
|
||||
if not q:
|
||||
return "Errore: serve il nome."
|
||||
store = _get_store()
|
||||
keys = [Contacts.CNContactGivenNameKey, Contacts.CNContactFamilyNameKey, Contacts.CNContactOrganizationNameKey,
|
||||
Contacts.CNContactPhoneNumbersKey, Contacts.CNContactEmailAddressesKey]
|
||||
pred = Contacts.CNContact.predicateForContactsMatchingName_(q)
|
||||
people, err = store.unifiedContactsMatchingPredicate_keysToFetch_error_(pred, keys, None)
|
||||
if err:
|
||||
return f"Errore: {err}"
|
||||
people = list(people or [])[:6]
|
||||
if not people:
|
||||
return f"Nessun contatto trovato per '{q}'."
|
||||
lines = []
|
||||
for p in people:
|
||||
name = f"{p.givenName()} {p.familyName()}".strip() or p.organizationName()
|
||||
bits = [name]
|
||||
tels = [ph.value().stringValue() for ph in p.phoneNumbers()]
|
||||
mails = [str(e.value()) for e in p.emailAddresses()]
|
||||
if tels:
|
||||
bits.append("tel " + ", ".join(tels))
|
||||
if mails:
|
||||
bits.append("email " + ", ".join(mails))
|
||||
if p.organizationName() and p.organizationName() != name:
|
||||
bits.append(p.organizationName())
|
||||
lines.append("- " + " · ".join(bits))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "contatti_cerca", "description": "Cerca una persona in Contatti e restituisce telefono, email e azienda. Usalo per trovare l'indirizzo email o il numero prima di scrivere a qualcuno per nome.",
|
||||
"parameters": {"type": "object", "properties": {"nome": {"type": "string"}}, "required": ["nome"]}, "run": cerca},
|
||||
]
|
||||
+100
@@ -0,0 +1,100 @@
|
||||
"""File e documenti: cerca con Spotlight, elenca cartelle, legge testo, PDF e Word."""
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
HOME = Path.home()
|
||||
MAX_CHARS = 6000
|
||||
FOLDERS = {"desktop": HOME / "Desktop", "scrivania": HOME / "Desktop", "documenti": HOME / "Documents", "download": HOME / "Downloads",
|
||||
"immagini": HOME / "Pictures", "musica": HOME / "Music", "home": HOME}
|
||||
|
||||
|
||||
def _folder(name: str) -> Path:
|
||||
n = (name or "").strip().lower()
|
||||
if n in FOLDERS:
|
||||
return FOLDERS[n]
|
||||
p = Path(name).expanduser()
|
||||
return p if p.exists() else HOME
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
if not q:
|
||||
return "Errore: serve il campo 'query'."
|
||||
folder = _folder(str(params.get("cartella", "home")))
|
||||
kind = str(params.get("tipo", "")).lower()
|
||||
query = f'kMDItemFSName == "*{q}*"cd || kMDItemTextContent == "*{q}*"cd'
|
||||
if kind == "pdf":
|
||||
query = f'({query}) && kMDItemContentType == "com.adobe.pdf"'
|
||||
elif kind in ("immagine", "immagini"):
|
||||
query = f'({query}) && kMDItemContentTypeTree == "public.image"'
|
||||
elif kind in ("documento", "documenti"):
|
||||
query = f'({query}) && (kMDItemContentTypeTree == "public.text" || kMDItemContentType == "com.adobe.pdf" || kMDItemContentType == "org.openxmlformats.wordprocessingml.document")'
|
||||
res = subprocess.run(["mdfind", "-onlyin", str(folder), query], capture_output=True, text=True, timeout=30)
|
||||
paths = [Path(p) for p in res.stdout.splitlines() if p and "/Library/" not in p][:200]
|
||||
if not paths:
|
||||
return f"Nessun file trovato per '{q}' in {folder.name or 'home'}."
|
||||
paths.sort(key=lambda p: p.stat().st_mtime if p.exists() else 0, reverse=True)
|
||||
lines = [f"- {p.name} ({datetime.fromtimestamp(p.stat().st_mtime):%d/%m/%Y}) — {p.parent}" for p in paths[:12] if p.exists()]
|
||||
return f"Trovati {len(paths)} file per '{q}':\n" + "\n".join(lines)
|
||||
|
||||
|
||||
def elenca(params: dict, ctx: dict) -> str:
|
||||
folder = _folder(str(params.get("cartella", "desktop")))
|
||||
items = sorted([p for p in folder.iterdir() if not p.name.startswith(".")], key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
if not items:
|
||||
return f"La cartella {folder.name} è vuota."
|
||||
lines = [f"- {p.name}{'/' if p.is_dir() else ''} ({datetime.fromtimestamp(p.stat().st_mtime):%d/%m %H:%M})" for p in items[:25]]
|
||||
return f"{folder.name} ({len(items)} elementi, i più recenti):\n" + "\n".join(lines)
|
||||
|
||||
|
||||
def _read(path: Path) -> str:
|
||||
ext = path.suffix.lower()
|
||||
if ext == ".pdf":
|
||||
from pypdf import PdfReader
|
||||
reader = PdfReader(str(path))
|
||||
return "\n".join((page.extract_text() or "") for page in reader.pages[:30])
|
||||
if ext == ".docx":
|
||||
import docx
|
||||
d = docx.Document(str(path))
|
||||
return "\n".join(p.text for p in d.paragraphs)
|
||||
if ext in (".txt", ".md", ".csv", ".json", ".py", ".js", ".ts", ".html", ".rtf", ".log"):
|
||||
return path.read_text(encoding="utf-8", errors="replace")
|
||||
try:
|
||||
return subprocess.run(["textutil", "-convert", "txt", "-stdout", str(path)], capture_output=True, text=True, timeout=30).stdout
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def leggi(params: dict, ctx: dict) -> str:
|
||||
name = str(params.get("percorso", "")).strip()
|
||||
if not name:
|
||||
return "Errore: serve il percorso o il nome del file."
|
||||
path = Path(name).expanduser()
|
||||
if not path.exists():
|
||||
res = subprocess.run(["mdfind", "-onlyin", str(HOME), f'kMDItemFSName == "*{path.name}*"cd'], capture_output=True, text=True, timeout=20)
|
||||
cands = [Path(p) for p in res.stdout.splitlines() if p and "/Library/" not in p]
|
||||
if not cands:
|
||||
return f"File non trovato: {name}"
|
||||
cands.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
path = cands[0]
|
||||
try:
|
||||
text = _read(path)
|
||||
except Exception as err:
|
||||
return f"Non riesco a leggere {path.name}: {err}"
|
||||
text = " ".join(text.split())
|
||||
if not text:
|
||||
return f"{path.name}: nessun testo estraibile."
|
||||
return f"{path.name} ({len(text)} caratteri):\n{text[:MAX_CHARS]}{' […]' if len(text) > MAX_CHARS else ''}"
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "file_cerca", "description": "Cerca file per nome o contenuto con Spotlight, in tutta la home o in una cartella (desktop, documenti, download…). Tipo: pdf, immagine, documento.",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}, "cartella": {"type": "string"}, "tipo": {"type": "string"}}, "required": ["query"]}, "run": cerca},
|
||||
{"name": "file_elenca", "description": "Elenca i file di una cartella (desktop, documenti, download o un percorso), i più recenti prima.",
|
||||
"parameters": {"type": "object", "properties": {"cartella": {"type": "string"}}}, "run": elenca},
|
||||
{"name": "file_leggi", "description": "Legge il testo di un file (txt, md, PDF, Word, pagine, rtf) per riassumerlo o rispondere a domande. Accetta un percorso o solo il nome (lo cerca).",
|
||||
"parameters": {"type": "object", "properties": {"percorso": {"type": "string"}}, "required": ["percorso"]}, "run": leggi},
|
||||
]
|
||||
+126
@@ -0,0 +1,126 @@
|
||||
"""Controlli del Mac: volume, luminosità, non disturbare, Wi-Fi, batteria, disco, schermo, screenshot."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import osa, sh, confirm_or_param, CONFERMATO # noqa: E402
|
||||
|
||||
|
||||
def volume(params: dict, ctx: dict) -> str:
|
||||
az = str(params.get("azione", "imposta")).lower()
|
||||
cur = int(osa("output volume of (get volume settings)"))
|
||||
if az == "leggi":
|
||||
muted = osa("output muted of (get volume settings)") == "true"
|
||||
return f"Volume al {cur}%" + (" (silenziato)." if muted else ".")
|
||||
if az == "muto":
|
||||
osa("set volume with output muted"); return "Audio silenziato."
|
||||
if az == "riattiva":
|
||||
osa("set volume without output muted"); return "Audio riattivato."
|
||||
v = int(params.get("valore") or 0)
|
||||
if az == "alza":
|
||||
v = min(100, cur + (int(params.get("valore") or 10)))
|
||||
elif az == "abbassa":
|
||||
v = max(0, cur - (int(params.get("valore") or 10)))
|
||||
v = max(0, min(100, v))
|
||||
osa(f"set volume output volume {v}")
|
||||
return f"Volume al {v}%."
|
||||
|
||||
|
||||
def luminosita(params: dict, ctx: dict) -> str:
|
||||
az = str(params.get("azione", "alza")).lower()
|
||||
key = 144 if az == "alza" else 145
|
||||
n = max(1, min(16, int(params.get("passi") or 4)))
|
||||
for _ in range(n):
|
||||
osa(f'tell application "System Events" to key code {key}')
|
||||
return f"Luminosità {'aumentata' if az == 'alza' else 'diminuita'}."
|
||||
|
||||
|
||||
def non_disturbare(params: dict, ctx: dict) -> str:
|
||||
on = bool(params.get("attiva", True))
|
||||
try:
|
||||
sh("shortcuts", "run", "Non disturbare ON" if on else "Non disturbare OFF")
|
||||
return "Non disturbare " + ("attivato." if on else "disattivato.")
|
||||
except Exception:
|
||||
return ("Per comandare Non disturbare crea due Comandi Rapidi chiamati 'Non disturbare ON' e 'Non disturbare OFF' "
|
||||
"con l'azione 'Imposta full immersion'. Poi funzionerà a voce.")
|
||||
|
||||
|
||||
def wifi(params: dict, ctx: dict) -> str:
|
||||
on = bool(params.get("attiva", True))
|
||||
try:
|
||||
dev = sh("networksetup", "-listallhardwareports")
|
||||
m = re.search(r"Hardware Port: Wi-Fi\nDevice: (\w+)", dev)
|
||||
iface = m.group(1) if m else "en0"
|
||||
sh("networksetup", "-setairportpower", iface, "on" if on else "off")
|
||||
return "Wi-Fi " + ("acceso." if on else "spento.")
|
||||
except Exception as err:
|
||||
return f"Non riesco a cambiare il Wi-Fi: {err}"
|
||||
|
||||
|
||||
def stato(params: dict, ctx: dict) -> str:
|
||||
parts = []
|
||||
try:
|
||||
batt = sh("pmset", "-g", "batt")
|
||||
m = re.search(r"(\d+)%;\s*([^;]+);", batt)
|
||||
if m:
|
||||
st = {"charging": "in carica", "discharging": "in uso a batteria", "charged": "carica", "AC attached": "collegato"}.get(m.group(2).strip(), m.group(2).strip())
|
||||
parts.append(f"Batteria al {m.group(1)}%, {st}")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
df = sh("df", "-h", "/System/Volumes/Data").splitlines()[-1].split()
|
||||
parts.append(f"Disco: {df[3]} liberi su {df[1]} ({df[4]} usato)")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
up = sh("uptime")
|
||||
m = re.search(r"up\s+([^,]+(?:,\s*\d+:\d+)?)", up)
|
||||
if m:
|
||||
parts.append(f"Acceso da {m.group(1).strip()}")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
wifi_name = sh("networksetup", "-getairportnetwork", "en0")
|
||||
parts.append(wifi_name.replace("Current Wi-Fi Network:", "Wi-Fi:").strip())
|
||||
except Exception:
|
||||
pass
|
||||
return ". ".join(parts) + "." if parts else "Stato non disponibile."
|
||||
|
||||
|
||||
def schermo(params: dict, ctx: dict) -> str:
|
||||
az = str(params.get("azione", "spegni")).lower()
|
||||
if az == "spegni":
|
||||
sh("pmset", "displaysleepnow"); return "Schermo spento."
|
||||
if az == "blocca":
|
||||
osa('tell application "System Events" to keystroke "q" using {command down, control down}'); return "Mac bloccato."
|
||||
if az == "sospendi":
|
||||
return confirm_or_param(ctx, params, "mac_sospendi", "Mettere il Mac in stop?", "Il Mac andrà in stop.", lambda: (sh("pmset", "sleepnow"), "Stop.")[1])
|
||||
return "Azione sconosciuta: spegni, blocca o sospendi."
|
||||
|
||||
|
||||
def screenshot(params: dict, ctx: dict) -> str:
|
||||
dest = Path.home() / "Desktop" / f"Screenshot {datetime.now():%Y-%m-%d %H.%M.%S}.png"
|
||||
sh("screencapture", "-x", str(dest))
|
||||
return f"Screenshot salvato sul Desktop: {dest.name}"
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "mac_volume", "description": "Volume di sistema: leggi, imposta (0-100), alza, abbassa, muto, riattiva.",
|
||||
"parameters": {"type": "object", "properties": {"azione": {"type": "string", "enum": ["leggi", "imposta", "alza", "abbassa", "muto", "riattiva"]}, "valore": {"type": "integer"}}, "required": ["azione"]}, "run": volume},
|
||||
{"name": "mac_luminosita", "description": "Alza o abbassa la luminosità dello schermo (a passi).",
|
||||
"parameters": {"type": "object", "properties": {"azione": {"type": "string", "enum": ["alza", "abbassa"]}, "passi": {"type": "integer"}}, "required": ["azione"]}, "run": luminosita},
|
||||
{"name": "mac_non_disturbare", "description": "Attiva o disattiva Non disturbare (full immersion).",
|
||||
"parameters": {"type": "object", "properties": {"attiva": {"type": "boolean"}}, "required": ["attiva"]}, "run": non_disturbare},
|
||||
{"name": "mac_wifi", "description": "Accende o spegne il Wi-Fi.",
|
||||
"parameters": {"type": "object", "properties": {"attiva": {"type": "boolean"}}, "required": ["attiva"]}, "run": wifi},
|
||||
{"name": "mac_stato", "description": "Stato del Mac: batteria, spazio su disco, tempo di accensione, rete Wi-Fi.",
|
||||
"parameters": {"type": "object", "properties": {}}, "run": stato},
|
||||
{"name": "mac_schermo", "description": "Spegne lo schermo, blocca il Mac o lo mette in stop (con conferma).",
|
||||
"parameters": {"type": "object", "properties": {"azione": {"type": "string", "enum": ["spegni", "blocca", "sospendi"]}, "confermato": CONFERMATO}, "required": ["azione"]}, "run": schermo},
|
||||
{"name": "mac_screenshot", "description": "Cattura lo schermo e salva l'immagine sul Desktop.", "parameters": {"type": "object", "properties": {}}, "run": screenshot},
|
||||
]
|
||||
+223
@@ -0,0 +1,223 @@
|
||||
"""Mail di macOS: posta non letta, ricerca, lettura (dall'indice di Mail) e invio (AppleScript)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
import sqlite3
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
MAIL_DIR = Path.home() / "Library" / "Mail"
|
||||
INDEX = MAIL_DIR / "V10" / "MailData" / "Envelope Index"
|
||||
MAX_BODY = 1800
|
||||
|
||||
SELECT = """
|
||||
SELECT m.ROWID, COALESCE(a.comment, ''), COALESCE(a.address, ''), COALESCE(s.subject, ''),
|
||||
m.date_received, m.read, mb.url, COALESCE(su.summary, '')
|
||||
FROM messages m
|
||||
JOIN mailboxes mb ON m.mailbox = mb.ROWID
|
||||
LEFT JOIN addresses a ON m.sender = a.ROWID
|
||||
LEFT JOIN subjects s ON m.subject = s.ROWID
|
||||
LEFT JOIN summaries su ON m.summary = su.ROWID
|
||||
WHERE m.deleted = 0 AND (mb.url LIKE '%/INBOX' OR mb.url LIKE '%/Inbox')
|
||||
"""
|
||||
|
||||
|
||||
def _query(where: str, params: tuple, limit: int) -> list[dict]:
|
||||
if not INDEX.exists():
|
||||
raise RuntimeError("Indice di Mail non trovato: l'app Mail è configurata su questo Mac?")
|
||||
con = sqlite3.connect(f"file:{INDEX}?mode=ro", uri=True, timeout=5)
|
||||
try:
|
||||
rows = con.execute(f"{SELECT} {where} ORDER BY m.date_received DESC LIMIT ?", (*params, limit)).fetchall()
|
||||
finally:
|
||||
con.close()
|
||||
out = []
|
||||
for rid, name, addr, subj, ts, read, url, summary in rows:
|
||||
who = f"{name} <{addr}>" if name and addr and name != addr else (name or addr or "?")
|
||||
out.append({"id": rid, "da": who, "oggetto": subj or "(senza oggetto)", "data": datetime.fromtimestamp(ts or 0),
|
||||
"letta": bool(read), "url": url, "anteprima": " ".join(str(summary).split())[:120]})
|
||||
return out
|
||||
|
||||
|
||||
def _fmt(rows: list[dict], preview: bool = False) -> str:
|
||||
lines = []
|
||||
for r in rows:
|
||||
line = f"- [{r['id']}] {'' if r['letta'] else '● '}{r['data']:%d/%m %H:%M} — {r['da']}: {r['oggetto']}"
|
||||
if preview and r["anteprima"]:
|
||||
line += f"\n {r['anteprima']}"
|
||||
lines.append(line)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def non_lette(params: dict, ctx: dict) -> str:
|
||||
lim = max(1, min(int(params.get("limite") or 10), 30))
|
||||
rows = _query("AND m.read = 0", (), lim)
|
||||
if not rows:
|
||||
return "Nessuna email non letta nella posta in arrivo."
|
||||
return f"Email non lette (le più recenti, max {lim}):\n{_fmt(rows, preview=True)}\nPer leggerne una usa mail_leggi con l'id tra parentesi."
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
lim = max(1, min(int(params.get("limite") or 10), 30))
|
||||
if not q:
|
||||
return "Errore: serve il campo 'query'."
|
||||
like = f"%{q}%"
|
||||
rows = _query("AND (s.subject LIKE ? OR a.address LIKE ? OR a.comment LIKE ?)", (like, like, like), lim)
|
||||
return f"Risultati per '{q}':\n{_fmt(rows)}\nPer leggerne una usa mail_leggi con l'id." if rows else f"Nessuna email trovata per '{q}'."
|
||||
|
||||
|
||||
# ── Lettura dal file .emlx ────────────────────────────────────────────────────
|
||||
def _mailbox_dir(url: str) -> Path | None:
|
||||
m = re.match(r"^[a-z]+://([^/]+)/(.+)$", url)
|
||||
if not m:
|
||||
return None
|
||||
account, path = m.group(1), m.group(2)
|
||||
d = MAIL_DIR / "V10" / account
|
||||
for part in path.split("/"):
|
||||
d = d / f"{part}.mbox"
|
||||
return d if d.exists() else None
|
||||
|
||||
|
||||
def _find_emlx(rid: int, url: str) -> Path | None:
|
||||
base = _mailbox_dir(url)
|
||||
if base is None:
|
||||
return None
|
||||
digits = list(str(rid // 1000))[::-1] if rid >= 1000 else []
|
||||
for sub in base.iterdir():
|
||||
if not sub.is_dir():
|
||||
continue
|
||||
data = sub / "Data"
|
||||
cand = data.joinpath(*digits) / "Messages" if digits else data / "Messages"
|
||||
for name in (f"{rid}.emlx", f"{rid}.partial.emlx"):
|
||||
p = cand / name
|
||||
if p.exists():
|
||||
return p
|
||||
for p in base.rglob(f"{rid}*.emlx"):
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def _emlx_message(path: Path):
|
||||
import email
|
||||
from email import policy
|
||||
data = path.read_bytes()
|
||||
nl = data.find(b"\n")
|
||||
try:
|
||||
n = int(data[:nl].strip())
|
||||
raw = data[nl + 1:nl + 1 + n]
|
||||
except ValueError:
|
||||
raw = data
|
||||
return email.message_from_bytes(raw, policy=policy.default)
|
||||
|
||||
|
||||
def _body_text(msg) -> str:
|
||||
try:
|
||||
part = msg.get_body(preferencelist=("plain", "html"))
|
||||
except Exception:
|
||||
part = None
|
||||
if part is None:
|
||||
return ""
|
||||
try:
|
||||
text = part.get_content()
|
||||
except Exception:
|
||||
return ""
|
||||
if part.get_content_type() == "text/html":
|
||||
text = re.sub(r"<(script|style)[^>]*>.*?</\1>", " ", text, flags=re.S | re.I)
|
||||
text = re.sub(r"<br\s*/?>|</p>|</div>|</tr>|</li>", "\n", text, flags=re.I)
|
||||
text = html.unescape(re.sub(r"<[^>]+>", " ", text))
|
||||
return " ".join(text.split())
|
||||
|
||||
|
||||
def leggi(params: dict, ctx: dict) -> str:
|
||||
mid = str(params.get("id", "")).strip()
|
||||
if not mid.isdigit():
|
||||
return "Errore: serve l'id numerico del messaggio preso dall'elenco."
|
||||
rows = _query("AND m.ROWID = ?", (int(mid),), 1)
|
||||
if not rows:
|
||||
return "Messaggio non trovato."
|
||||
r = rows[0]
|
||||
path = _find_emlx(r["id"], r["url"])
|
||||
body = ""
|
||||
if path is not None:
|
||||
try:
|
||||
body = _body_text(_emlx_message(path))
|
||||
except Exception as err:
|
||||
body = f"(testo non leggibile: {err})"
|
||||
if not body:
|
||||
body = r["anteprima"] or "(testo non disponibile: il messaggio non è ancora scaricato in locale)"
|
||||
if len(body) > MAX_BODY:
|
||||
body = body[:MAX_BODY] + " […]"
|
||||
return f"Da: {r['da']}\nOggetto: {r['oggetto']}\nData: {r['data']:%d/%m/%Y %H:%M}\n\n{body}"
|
||||
|
||||
|
||||
# ── Invio via AppleScript ─────────────────────────────────────────────────────
|
||||
SEND_SCRIPT = '''
|
||||
on run argv
|
||||
set addr to item 1 of argv
|
||||
set subj to item 2 of argv
|
||||
set body to item 3 of argv
|
||||
tell application "Mail"
|
||||
set msg to make new outgoing message with properties {subject:subj, content:body, visible:false}
|
||||
tell msg to make new to recipient at end of to recipients with properties {address:addr}
|
||||
send msg
|
||||
end tell
|
||||
return "ok"
|
||||
end run
|
||||
'''
|
||||
|
||||
|
||||
def _osa(script: str, *args: str, timeout: int = 60) -> str:
|
||||
res = subprocess.run(["osascript", "-e", script, "--", *args], capture_output=True, text=True, timeout=timeout)
|
||||
if res.returncode != 0:
|
||||
err = (res.stderr or "").strip()
|
||||
if "-1743" in err:
|
||||
raise RuntimeError("Accesso a Mail negato: concedilo in Impostazioni di Sistema > Privacy e sicurezza > Automazione.")
|
||||
raise RuntimeError(err.splitlines()[-1] if err else "errore AppleScript")
|
||||
return res.stdout.rstrip("\n")
|
||||
|
||||
|
||||
def invia(params: dict, ctx: dict) -> str:
|
||||
a = str(params.get("a", "")).strip()
|
||||
oggetto = str(params.get("oggetto", "")).strip()
|
||||
testo = str(params.get("testo", "")).strip()
|
||||
if "@" not in a or not oggetto or not testo:
|
||||
return "Errore: servono destinatario (indirizzo email), oggetto e testo."
|
||||
desc = f"A: {a}\nOggetto: {oggetto}\n{testo[:160]}{'…' if len(testo) > 160 else ''}"
|
||||
|
||||
def do() -> str:
|
||||
_osa(SEND_SCRIPT, a, oggetto, testo)
|
||||
msg = f"Email inviata a {a} con oggetto «{oggetto}»."
|
||||
if ctx.get("say"):
|
||||
ctx["say"](msg)
|
||||
return msg
|
||||
|
||||
confirm = ctx.get("confirm")
|
||||
if confirm:
|
||||
return confirm("mail_invia", "Inviare questa email?", desc, do)
|
||||
if str(params.get("confermato", "")).lower() in ("true", "1", "sì", "si", "yes"):
|
||||
return do()
|
||||
return f"Prima di inviare chiedi conferma all'utente leggendo destinatario, oggetto e testo:\n{desc}\nPoi richiama con confermato=true."
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "mail_non_lette",
|
||||
"description": "Elenca le email non lette nella posta in arrivo di Mail (mittente, oggetto, data, anteprima, id). Usalo per 'ho email nuove?', 'leggimi la posta'.",
|
||||
"parameters": {"type": "object", "properties": {"limite": {"type": "integer", "description": "Quante al massimo (default 10)."}}},
|
||||
"run": non_lette},
|
||||
{"name": "mail_cerca",
|
||||
"description": "Cerca email nella posta in arrivo per parola nell'oggetto o nel mittente (nome o indirizzo).",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}, "limite": {"type": "integer"}}, "required": ["query"]},
|
||||
"run": cerca},
|
||||
{"name": "mail_leggi",
|
||||
"description": "Legge il contenuto di un'email dato il suo id (dall'elenco). Riassumi il testo a voce, non leggerlo integralmente se è lungo.",
|
||||
"parameters": {"type": "object", "properties": {"id": {"type": "string", "description": "Id numerico del messaggio."}}, "required": ["id"]},
|
||||
"run": leggi},
|
||||
{"name": "mail_invia",
|
||||
"description": "Invia un'email dall'account predefinito di Mail. Azione irreversibile: viene chiesta conferma sullo schermo; se lo strumento risponde con [CONFIRMATION_PENDING], di' all'utente di confermare sul pannello e non dire che è inviata. Scrivi tu il testo completo in italiano, con saluto e firma con il nome dell'utente.",
|
||||
"parameters": {"type": "object", "properties": {"a": {"type": "string", "description": "Indirizzo email del destinatario."}, "oggetto": {"type": "string"}, "testo": {"type": "string"},
|
||||
"confermato": {"type": "boolean", "description": "Solo senza interfaccia: true dopo la conferma esplicita dell'utente."}},
|
||||
"required": ["a", "oggetto", "testo"]},
|
||||
"run": invia},
|
||||
]
|
||||
@@ -0,0 +1,41 @@
|
||||
"""Messaggi (iMessage/SMS): invia un messaggio a un numero o a un contatto, con conferma."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
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
|
||||
tell application "Messages"
|
||||
set svc to 1st account whose service type = iMessage
|
||||
set b to participant (item 1 of argv) of svc
|
||||
send (item 2 of argv) to b
|
||||
end tell
|
||||
return "ok"
|
||||
end run'''
|
||||
|
||||
|
||||
def invia(params: dict, ctx: dict) -> str:
|
||||
a = str(params.get("a", "")).strip()
|
||||
testo = str(params.get("testo", "")).strip()
|
||||
if not a or not testo:
|
||||
return "Errore: servono destinatario (numero con prefisso, es. +39…, o email iMessage) e testo."
|
||||
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."
|
||||
|
||||
def do() -> str:
|
||||
osa(SEND, a.replace(" ", ""), testo, timeout=30)
|
||||
msg = f"Messaggio inviato a {a}."
|
||||
if ctx.get("say"):
|
||||
ctx["say"](msg)
|
||||
return msg
|
||||
|
||||
return confirm_or_param(ctx, params, "messaggi_invia", "Inviare questo messaggio?", f"A: {a}\n{testo[:200]}", do)
|
||||
|
||||
|
||||
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},
|
||||
]
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Meteo con Open-Meteo (gratuito, senza chiave): oggi, domani o i prossimi giorni per una città."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import requests
|
||||
|
||||
CODES = {0: "sereno", 1: "prevalentemente sereno", 2: "parzialmente nuvoloso", 3: "coperto", 45: "nebbia", 48: "nebbia con brina",
|
||||
51: "pioviggine leggera", 53: "pioviggine", 55: "pioviggine intensa", 61: "pioggia leggera", 63: "pioggia", 65: "pioggia forte",
|
||||
71: "neve leggera", 73: "neve", 75: "neve forte", 80: "rovesci leggeri", 81: "rovesci", 82: "rovesci violenti", 95: "temporale",
|
||||
96: "temporale con grandine", 99: "temporale con grandine forte"}
|
||||
GIORNI = ["lunedì", "martedì", "mercoledì", "giovedì", "venerdì", "sabato", "domenica"]
|
||||
|
||||
|
||||
def _geocode(city: str) -> tuple[float, float, str]:
|
||||
r = requests.get("https://geocoding-api.open-meteo.com/v1/search", params={"name": city, "count": 1, "language": "it"}, timeout=10)
|
||||
r.raise_for_status()
|
||||
res = (r.json().get("results") or [])
|
||||
if not res:
|
||||
raise ValueError(f"Città non trovata: {city}")
|
||||
g = res[0]
|
||||
return g["latitude"], g["longitude"], f"{g['name']}{', ' + g['admin1'] if g.get('admin1') else ''}"
|
||||
|
||||
|
||||
def previsioni(params: dict, ctx: dict) -> str:
|
||||
city = str(params.get("citta", "")).strip() or "Roma"
|
||||
giorni = max(1, min(7, int(params.get("giorni") or 2)))
|
||||
lat, lon, label = _geocode(city)
|
||||
r = requests.get("https://api.open-meteo.com/v1/forecast", params={
|
||||
"latitude": lat, "longitude": lon, "timezone": "auto",
|
||||
"current": "temperature_2m,apparent_temperature,weather_code,wind_speed_10m,relative_humidity_2m",
|
||||
"daily": "weather_code,temperature_2m_max,temperature_2m_min,precipitation_probability_max,precipitation_sum",
|
||||
"forecast_days": giorni}, timeout=10)
|
||||
r.raise_for_status()
|
||||
d = r.json()
|
||||
cur = d.get("current", {})
|
||||
out = [f"Meteo a {label}: ora {CODES.get(cur.get('weather_code'), '')}, {cur.get('temperature_2m')}° (percepiti {cur.get('apparent_temperature')}°), vento {cur.get('wind_speed_10m')} km/h, umidità {cur.get('relative_humidity_2m')}%."]
|
||||
daily = d.get("daily", {})
|
||||
for i, day in enumerate(daily.get("time", [])):
|
||||
dt = datetime.strptime(day, "%Y-%m-%d")
|
||||
nome = "oggi" if i == 0 else ("domani" if i == 1 else GIORNI[dt.weekday()])
|
||||
out.append(f"- {nome} {dt:%d/%m}: {CODES.get(daily['weather_code'][i], '')}, min {daily['temperature_2m_min'][i]}° max {daily['temperature_2m_max'][i]}°, pioggia {daily['precipitation_probability_max'][i]}%")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "meteo", "description": "Previsioni meteo per una città: ora, oggi e i prossimi giorni.",
|
||||
"parameters": {"type": "object", "properties": {"citta": {"type": "string"}, "giorni": {"type": "integer", "description": "Quanti giorni (1-7, default 2)."}}, "required": ["citta"]},
|
||||
"run": previsioni},
|
||||
]
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Musica: controlla Apple Music (o Spotify, se aperto): riproduci, pausa, brano, playlist, volume."""
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import osa # noqa: E402
|
||||
|
||||
|
||||
def _running(name: str) -> bool:
|
||||
return subprocess.run(["pgrep", "-x", name], capture_output=True).returncode == 0
|
||||
|
||||
|
||||
def _app(launch: bool = False) -> str | None:
|
||||
"""App musicale in uso: Spotify se è l'unica aperta, altrimenti Musica (avviata se launch=True)."""
|
||||
if _running("Spotify") and not _running("Music"):
|
||||
return "Spotify"
|
||||
if _running("Music"):
|
||||
return "Music"
|
||||
if launch:
|
||||
subprocess.run(["open", "-a", "Music"], check=False, timeout=15)
|
||||
for _ in range(20):
|
||||
if _running("Music"):
|
||||
time.sleep(2)
|
||||
return "Music"
|
||||
time.sleep(0.5)
|
||||
return None
|
||||
|
||||
|
||||
def controllo(params: dict, ctx: dict) -> str:
|
||||
azione = str(params.get("azione", "")).lower()
|
||||
app = _app(launch=azione in ("play", "riproduci"))
|
||||
if app is None:
|
||||
return "Nessuna app musicale aperta: di' «riproduci» per avviare Musica, oppure apri Spotify."
|
||||
cmds = {"play": "play", "riproduci": "play", "pausa": "pause", "stop": "stop", "prossimo": "next track", "avanti": "next track",
|
||||
"precedente": "previous track", "indietro": "previous track"}
|
||||
if azione in cmds:
|
||||
osa(f'tell application "{app}" to {cmds[azione]}', timeout=20)
|
||||
return f"Fatto ({azione})."
|
||||
if azione in ("brano", "cosa suona", "info"):
|
||||
try:
|
||||
out = osa(f'tell application "{app}" to return (name of current track) & " — " & (artist of current track) & " (" & (album of current track) & ")"')
|
||||
stato = osa(f'tell application "{app}" to return player state as string')
|
||||
return f"{'In riproduzione' if stato == 'playing' else 'In pausa'}: {out}"
|
||||
except Exception:
|
||||
return "Nessun brano in riproduzione."
|
||||
if azione == "volume":
|
||||
v = max(0, min(100, int(params.get("valore") or 50)))
|
||||
osa(f'tell application "{app}" to set sound volume to {v}')
|
||||
return f"Volume musica al {v}%."
|
||||
return "Azione sconosciuta. Usa: riproduci, pausa, prossimo, precedente, brano, volume."
|
||||
|
||||
|
||||
def riproduci(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("nome", "")).strip()
|
||||
tipo = str(params.get("tipo", "auto")).lower()
|
||||
if not q:
|
||||
return "Errore: serve il nome."
|
||||
app = _app(launch=True)
|
||||
if app == "Spotify":
|
||||
return "Con Spotify posso solo controllare la riproduzione, non cercare: avvia tu il brano e poi dimmi pausa, avanti, ecc."
|
||||
if tipo in ("playlist", "auto"):
|
||||
try:
|
||||
out = osa('on run argv\ntell application "Music"\nset p to first playlist whose name contains (item 1 of argv)\nplay p\nreturn name of p\nend tell\nend run', q)
|
||||
return f"Riproduco la playlist «{out}»."
|
||||
except Exception:
|
||||
if tipo == "playlist":
|
||||
return f"Nessuna playlist chiamata '{q}'."
|
||||
for prop in ("artist", "album", "name"):
|
||||
try:
|
||||
out = osa(f'''on run argv
|
||||
tell application "Music"
|
||||
set trk to (first track of library playlist 1 whose {prop} contains (item 1 of argv))
|
||||
play trk
|
||||
return (name of trk) & " — " & (artist of trk)
|
||||
end tell
|
||||
end run''', q, timeout=60)
|
||||
return f"Riproduco: {out}."
|
||||
except Exception:
|
||||
continue
|
||||
return f"Non ho trovato '{q}' nella libreria di Musica."
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "musica_controllo", "description": "Controlla la musica: riproduci, pausa, prossimo, precedente, brano (cosa sta suonando), volume (0-100).",
|
||||
"parameters": {"type": "object", "properties": {"azione": {"type": "string", "enum": ["riproduci", "pausa", "prossimo", "precedente", "brano", "volume"]},
|
||||
"valore": {"type": "integer", "description": "Per volume: 0-100."}}, "required": ["azione"]}, "run": controllo},
|
||||
{"name": "musica_riproduci", "description": "Cerca e riproduce una playlist, un artista, un album o un brano nella libreria di Apple Music.",
|
||||
"parameters": {"type": "object", "properties": {"nome": {"type": "string"}, "tipo": {"type": "string", "enum": ["auto", "playlist", "artista", "album", "brano"]}}, "required": ["nome"]},
|
||||
"run": riproduci},
|
||||
]
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Note di macOS: crea, cerca e legge note (AppleScript)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import osa # noqa: E402
|
||||
|
||||
SEP = "␟"
|
||||
|
||||
|
||||
def _testo(body_html: str) -> str:
|
||||
t = re.sub(r"<br\s*/?>|</div>|</p>|</li>", "\n", body_html, flags=re.I)
|
||||
t = html.unescape(re.sub(r"<[^>]+>", "", t))
|
||||
return "\n".join(line.strip() for line in t.splitlines() if line.strip())
|
||||
|
||||
|
||||
def crea(params: dict, ctx: dict) -> str:
|
||||
titolo = str(params.get("titolo", "")).strip() or "Nota"
|
||||
testo = str(params.get("testo", "")).strip()
|
||||
body = f"<h1>{html.escape(titolo)}</h1>" + "".join(f"<div>{html.escape(l) or '<br>'}</div>" for l in testo.splitlines())
|
||||
osa('on run argv\ntell application "Notes" to make new note at folder "Notes" with properties {name:item 1 of argv, body:item 2 of argv}\nend run', titolo, body)
|
||||
return f"Nota creata: «{titolo}»."
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
if not q:
|
||||
return "Errore: serve il campo 'query'."
|
||||
out = osa(f'''on run argv
|
||||
set sep to "{SEP}"
|
||||
set out to ""
|
||||
set n to 0
|
||||
tell application "Notes"
|
||||
repeat with nt in (notes whose name contains (item 1 of argv) or plaintext contains (item 1 of argv))
|
||||
set n to n + 1
|
||||
if n > 8 then exit repeat
|
||||
set out to out & (name of nt) & sep & ((modification date of nt) as string) & linefeed
|
||||
end repeat
|
||||
end tell
|
||||
return out
|
||||
end run''', q, timeout=90)
|
||||
rows = [l.split(SEP) for l in out.splitlines() if l]
|
||||
if not rows:
|
||||
return f"Nessuna nota trovata per '{q}'."
|
||||
return f"Note trovate per '{q}':\n" + "\n".join(f"- {r[0]} ({r[1]})" for r in rows) + "\nPer leggerne una usa note_leggi con il titolo."
|
||||
|
||||
|
||||
def leggi(params: dict, ctx: dict) -> str:
|
||||
titolo = str(params.get("titolo", "")).strip()
|
||||
if not titolo:
|
||||
return "Errore: serve il titolo."
|
||||
out = osa('on run argv\ntell application "Notes" to return body of (first note whose name contains (item 1 of argv))\nend run', titolo, timeout=60)
|
||||
testo = _testo(out)
|
||||
return testo[:2500] + (" […]" if len(testo) > 2500 else "") if testo else "Nota vuota."
|
||||
|
||||
|
||||
def aggiungi(params: dict, ctx: dict) -> str:
|
||||
titolo = str(params.get("titolo", "")).strip()
|
||||
testo = str(params.get("testo", "")).strip()
|
||||
if not titolo or not testo:
|
||||
return "Errore: servono titolo (della nota esistente) e testo da aggiungere."
|
||||
extra = "".join(f"<div>{html.escape(l) or '<br>'}</div>" for l in testo.splitlines())
|
||||
osa('on run argv\ntell application "Notes"\nset nt to first note whose name contains (item 1 of argv)\nset body of nt to (body of nt) & (item 2 of argv)\nend tell\nend run', titolo, extra)
|
||||
return f"Aggiunto alla nota «{titolo}»."
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "note_crea", "description": "Crea una nuova nota nell'app Note con titolo e testo. Usalo per 'prendi nota', 'appunta che…'.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string"}, "testo": {"type": "string"}}, "required": ["testo"]}, "run": crea},
|
||||
{"name": "note_aggiungi", "description": "Aggiunge testo in fondo a una nota esistente, cercata per titolo.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string"}, "testo": {"type": "string"}}, "required": ["titolo", "testo"]}, "run": aggiungi},
|
||||
{"name": "note_cerca", "description": "Cerca note per parola nel titolo o nel testo.",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}, "run": cerca},
|
||||
{"name": "note_leggi", "description": "Legge il contenuto di una nota cercata per titolo.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string"}}, "required": ["titolo"]}, "run": leggi},
|
||||
]
|
||||
@@ -0,0 +1,149 @@
|
||||
"""Promemoria di macOS: elenca, aggiunge e completa promemoria (EventKit, sincronizzati con iPhone)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
_store = None
|
||||
_lock = threading.Lock()
|
||||
|
||||
|
||||
def _get_store():
|
||||
global _store
|
||||
with _lock:
|
||||
if _store is not None:
|
||||
return _store
|
||||
import EventKit
|
||||
store = EventKit.EKEventStore.alloc().init()
|
||||
done, res = threading.Event(), {}
|
||||
|
||||
def cb(granted, err):
|
||||
res["granted"] = bool(granted)
|
||||
done.set()
|
||||
|
||||
if hasattr(store, "requestFullAccessToRemindersWithCompletion_"):
|
||||
store.requestFullAccessToRemindersWithCompletion_(cb)
|
||||
else:
|
||||
store.requestAccessToEntityType_completion_(EventKit.EKEntityTypeReminder, cb)
|
||||
done.wait(60)
|
||||
if not res.get("granted"):
|
||||
raise RuntimeError("Accesso ai Promemoria negato: concedilo in Impostazioni di Sistema > Privacy e sicurezza > Promemoria.")
|
||||
_store = store
|
||||
return store
|
||||
|
||||
|
||||
def _parse_dt(s: str) -> datetime | None:
|
||||
s = (s or "").strip().replace("T", " ")
|
||||
if not s:
|
||||
return None
|
||||
for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d"):
|
||||
try:
|
||||
return datetime.strptime(s, fmt)
|
||||
except ValueError:
|
||||
continue
|
||||
raise ValueError(f"Data non valida: '{s}'. Usa AAAA-MM-GG HH:MM.")
|
||||
|
||||
|
||||
def _due(r) -> datetime | None:
|
||||
comps = r.dueDateComponents()
|
||||
if comps is None:
|
||||
return None
|
||||
try:
|
||||
return datetime(comps.year(), comps.month(), comps.day(), comps.hour() if comps.hour() != 9223372036854775807 else 0,
|
||||
comps.minute() if comps.minute() != 9223372036854775807 else 0)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _fetch_incomplete():
|
||||
store = _get_store()
|
||||
pred = store.predicateForIncompleteRemindersWithDueDateStarting_ending_calendars_(None, None, None)
|
||||
done, out = threading.Event(), []
|
||||
|
||||
def cb(items):
|
||||
out.extend(list(items or []))
|
||||
done.set()
|
||||
|
||||
store.fetchRemindersMatchingPredicate_completion_(pred, cb)
|
||||
done.wait(30)
|
||||
return out
|
||||
|
||||
|
||||
def _fmt(r) -> str:
|
||||
d = _due(r)
|
||||
when = f" (entro {d:%d/%m %H:%M})" if d and (d.hour or d.minute) else (f" (entro {d:%d/%m})" if d else "")
|
||||
return f"- {r.title()}{when} [{r.calendar().title()}]"
|
||||
|
||||
|
||||
def elenca(params: dict, ctx: dict) -> str:
|
||||
items = _fetch_incomplete()
|
||||
quando = str(params.get("quando", "tutti")).lower()
|
||||
today = datetime.now().date()
|
||||
if quando == "oggi":
|
||||
items = [r for r in items if (d := _due(r)) and d.date() <= today]
|
||||
elif quando == "settimana":
|
||||
items = [r for r in items if (d := _due(r)) and d.date() <= today + timedelta(days=7)]
|
||||
items.sort(key=lambda r: (_due(r) or datetime.max))
|
||||
if not items:
|
||||
return "Nessun promemoria da fare" + (" per oggi." if quando == "oggi" else ".")
|
||||
return f"Promemoria da fare ({len(items)}):\n" + "\n".join(_fmt(r) for r in items[:30])
|
||||
|
||||
|
||||
def aggiungi(params: dict, ctx: dict) -> str:
|
||||
import EventKit
|
||||
from Foundation import NSDateComponents
|
||||
store = _get_store()
|
||||
titolo = str(params.get("titolo", "")).strip()
|
||||
if not titolo:
|
||||
return "Errore: serve il titolo."
|
||||
r = EventKit.EKReminder.reminderWithEventStore_(store)
|
||||
r.setTitle_(titolo)
|
||||
lista = str(params.get("lista", "")).strip()
|
||||
cal = None
|
||||
if lista:
|
||||
for c in store.calendarsForEntityType_(EventKit.EKEntityTypeReminder):
|
||||
if c.title().lower() == lista.lower():
|
||||
cal = c
|
||||
r.setCalendar_(cal or store.defaultCalendarForNewReminders())
|
||||
scadenza = _parse_dt(str(params.get("scadenza", "")))
|
||||
if scadenza:
|
||||
comps = NSDateComponents.alloc().init()
|
||||
comps.setYear_(scadenza.year); comps.setMonth_(scadenza.month); comps.setDay_(scadenza.day)
|
||||
comps.setHour_(scadenza.hour); comps.setMinute_(scadenza.minute)
|
||||
r.setDueDateComponents_(comps)
|
||||
from Foundation import NSDate
|
||||
alarm = EventKit.EKAlarm.alarmWithAbsoluteDate_(NSDate.dateWithTimeIntervalSince1970_(scadenza.timestamp()))
|
||||
r.addAlarm_(alarm)
|
||||
if params.get("note"):
|
||||
r.setNotes_(str(params["note"]))
|
||||
ok, err = store.saveReminder_commit_error_(r, True, None)
|
||||
return f"Promemoria aggiunto: {_fmt(r)[2:]}" if ok else f"Errore: {err}"
|
||||
|
||||
|
||||
def completa(params: dict, ctx: dict) -> str:
|
||||
titolo = str(params.get("titolo", "")).strip().lower()
|
||||
if not titolo:
|
||||
return "Errore: serve il titolo (anche parziale)."
|
||||
found = [r for r in _fetch_incomplete() if titolo in (r.title() or "").lower()]
|
||||
if not found:
|
||||
return "Nessun promemoria con quel titolo."
|
||||
if len(found) > 1 and not params.get("tutti"):
|
||||
return "Ho trovato più promemoria, dimmi quale:\n" + "\n".join(_fmt(r) for r in found)
|
||||
store = _get_store()
|
||||
for r in found:
|
||||
r.setCompleted_(True)
|
||||
store.saveReminder_commit_error_(r, True, None)
|
||||
return "Completato: " + ", ".join(r.title() for r in found)
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "promemoria_elenca", "description": "Elenca i promemoria da fare dell'app Promemoria (tutti, di oggi o della settimana).",
|
||||
"parameters": {"type": "object", "properties": {"quando": {"type": "string", "enum": ["tutti", "oggi", "settimana"]}}}, "run": elenca},
|
||||
{"name": "promemoria_aggiungi", "description": "Aggiunge un promemoria (o una voce a una lista, per esempio 'Spesa'). Converti 'domani alle 17' in AAAA-MM-GG HH:MM usando la data del prompt.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string"}, "scadenza": {"type": "string", "description": "AAAA-MM-GG HH:MM, facoltativa."},
|
||||
"lista": {"type": "string", "description": "Nome della lista (facoltativo)."}, "note": {"type": "string"}}, "required": ["titolo"]},
|
||||
"run": aggiungi},
|
||||
{"name": "promemoria_completa", "description": "Segna come fatto un promemoria cercandolo per titolo.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string"}, "tutti": {"type": "boolean", "description": "Completa tutti quelli che corrispondono."}}, "required": ["titolo"]},
|
||||
"run": completa},
|
||||
]
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Telegram con il tuo account: chat non lette, lettura, ricerca e invio (con conferma)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import confirm_or_param, CONFERMATO # noqa: E402
|
||||
from avatar import telegram_client as tg # noqa: E402
|
||||
|
||||
|
||||
def _name(entity) -> str:
|
||||
for attr in ("title",):
|
||||
if getattr(entity, attr, None):
|
||||
return entity.title
|
||||
return f"{getattr(entity, 'first_name', '') or ''} {getattr(entity, 'last_name', '') or ''}".strip() or getattr(entity, "username", "") or "?"
|
||||
|
||||
|
||||
def _preview(msg) -> str:
|
||||
t = (msg.message or "").strip() if msg else ""
|
||||
if not t and msg is not None:
|
||||
t = "[media]" if getattr(msg, "media", None) else ""
|
||||
return " ".join(t.split())[:140]
|
||||
|
||||
|
||||
def non_letti(params: dict, ctx: dict) -> str:
|
||||
lim = max(1, min(int(params.get("limite") or 10), 30))
|
||||
|
||||
async def go(client):
|
||||
out, total = [], 0
|
||||
async for d in client.iter_dialogs(limit=100):
|
||||
if d.unread_count > 0 and not d.is_channel or (d.is_channel and d.is_group and d.unread_count > 0):
|
||||
total += d.unread_count
|
||||
sender = ""
|
||||
if d.message and d.is_group:
|
||||
try:
|
||||
s = await d.message.get_sender()
|
||||
sender = _name(s) + ": "
|
||||
except Exception:
|
||||
pass
|
||||
out.append(f"- {d.name} ({d.unread_count} non lett{'o' if d.unread_count == 1 else 'i'}): {sender}{_preview(d.message)}")
|
||||
if len(out) >= lim:
|
||||
break
|
||||
return out, total
|
||||
|
||||
out, total = tg.run(go)
|
||||
if not out:
|
||||
return "Nessun messaggio Telegram non letto (canali esclusi)."
|
||||
return f"Telegram, {total} messaggi non letti:\n" + "\n".join(out) + "\nPer leggere una chat usa telegram_leggi con il nome."
|
||||
|
||||
|
||||
async def _find_dialog(client, name: str):
|
||||
name_l = name.lower()
|
||||
best = None
|
||||
async for d in client.iter_dialogs(limit=300):
|
||||
n = (d.name or "").lower()
|
||||
if n == name_l:
|
||||
return d
|
||||
if name_l in n and best is None:
|
||||
best = d
|
||||
if best is None:
|
||||
try:
|
||||
ent = await client.get_entity(name)
|
||||
return ent
|
||||
except Exception:
|
||||
return None
|
||||
return best
|
||||
|
||||
|
||||
def leggi(params: dict, ctx: dict) -> str:
|
||||
chat = str(params.get("chat", "")).strip()
|
||||
n = max(1, min(int(params.get("numero") or 10), 40))
|
||||
if not chat:
|
||||
return "Errore: serve il nome della chat o del contatto."
|
||||
|
||||
async def go(client):
|
||||
d = await _find_dialog(client, chat)
|
||||
if d is None:
|
||||
return None
|
||||
entity = getattr(d, "entity", d)
|
||||
lines = []
|
||||
async for m in client.iter_messages(entity, limit=n):
|
||||
who = "tu" if m.out else _name(await m.get_sender()) if m.sender_id else "?"
|
||||
lines.append(f"[{m.date.astimezone():%d/%m %H:%M}] {who}: {_preview(m)}")
|
||||
try:
|
||||
await client.send_read_acknowledge(entity)
|
||||
except Exception:
|
||||
pass
|
||||
return getattr(d, "name", None) or _name(entity), list(reversed(lines))
|
||||
|
||||
res = tg.run(go)
|
||||
if res is None:
|
||||
return f"Nessuna chat trovata per '{chat}'."
|
||||
name, lines = res
|
||||
return f"Chat «{name}», ultimi messaggi:\n" + "\n".join(lines)
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
if not q:
|
||||
return "Errore: serve la query."
|
||||
|
||||
async def go(client):
|
||||
lines = []
|
||||
async for m in client.iter_messages(None, search=q, limit=15):
|
||||
try:
|
||||
chat = _name(await m.get_chat())
|
||||
except Exception:
|
||||
chat = "?"
|
||||
lines.append(f"[{m.date.astimezone():%d/%m %H:%M}] {chat}: {_preview(m)}")
|
||||
return lines
|
||||
|
||||
lines = tg.run(go)
|
||||
return f"Messaggi Telegram con '{q}':\n" + "\n".join(lines) if lines else f"Nessun messaggio trovato per '{q}'."
|
||||
|
||||
|
||||
def invia(params: dict, ctx: dict) -> str:
|
||||
chat = str(params.get("chat", "")).strip()
|
||||
testo = str(params.get("testo", "")).strip()
|
||||
if not chat or not testo:
|
||||
return "Errore: servono destinatario (nome della chat, @username o numero) e testo."
|
||||
|
||||
async def resolve(client):
|
||||
d = await _find_dialog(client, chat)
|
||||
if d is None:
|
||||
return None
|
||||
return getattr(d, "name", None) or _name(d)
|
||||
|
||||
name = tg.run(resolve)
|
||||
if name is None:
|
||||
return f"Non trovo '{chat}' tra le tue chat Telegram."
|
||||
|
||||
def do() -> str:
|
||||
async def send(client):
|
||||
d = await _find_dialog(client, chat)
|
||||
await client.send_message(getattr(d, "entity", d), testo)
|
||||
return f"Messaggio Telegram inviato a {name}."
|
||||
msg = tg.run(send)
|
||||
if ctx.get("say"):
|
||||
ctx["say"](msg)
|
||||
return msg
|
||||
|
||||
return confirm_or_param(ctx, params, "telegram_invia", "Inviare su Telegram?", f"A: {name}\n{testo[:200]}", do)
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "telegram_non_letti", "description": "Elenca le chat Telegram con messaggi non letti e l'ultimo messaggio di ciascuna.",
|
||||
"parameters": {"type": "object", "properties": {"limite": {"type": "integer"}}}, "run": non_letti},
|
||||
{"name": "telegram_leggi", "description": "Legge gli ultimi messaggi di una chat o di un contatto Telegram (per nome) e li segna come letti.",
|
||||
"parameters": {"type": "object", "properties": {"chat": {"type": "string"}, "numero": {"type": "integer"}}, "required": ["chat"]}, "run": leggi},
|
||||
{"name": "telegram_cerca", "description": "Cerca messaggi in tutte le chat Telegram per parola.",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}, "run": cerca},
|
||||
{"name": "telegram_invia", "description": "Invia un messaggio Telegram a un contatto o a un gruppo (per nome o @username). Azione irreversibile con conferma sullo schermo; con [CONFIRMATION_PENDING] di' all'utente di confermare e non dire che è inviato.",
|
||||
"parameters": {"type": "object", "properties": {"chat": {"type": "string"}, "testo": {"type": "string"}, "confermato": CONFERMATO}, "required": ["chat", "testo"]}, "run": invia},
|
||||
]
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Timer, sveglie e attività programmate: l'app avvisa a voce o esegue un comando all'ora stabilita, anche ogni giorno."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import uuid
|
||||
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 # noqa: E402
|
||||
|
||||
FILE = DATA_DIR / "timers.json"
|
||||
|
||||
|
||||
def load() -> list[dict]:
|
||||
try:
|
||||
return json.loads(FILE.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def save(items: list[dict]) -> None:
|
||||
FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
FILE.write_text(json.dumps(items, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
|
||||
|
||||
def _when(params: dict) -> datetime:
|
||||
now = datetime.now()
|
||||
if params.get("minuti"):
|
||||
return now + timedelta(minutes=float(params["minuti"]))
|
||||
s = str(params.get("ora", "")).strip().replace("T", " ")
|
||||
if not s:
|
||||
raise ValueError("Serve 'minuti' oppure 'ora'.")
|
||||
if len(s) <= 5: # HH:MM oggi (o domani se già passata)
|
||||
t = datetime.strptime(s, "%H:%M").time()
|
||||
dt = datetime.combine(now.date(), t)
|
||||
return dt if dt > now else dt + timedelta(days=1)
|
||||
return datetime.strptime(s, "%Y-%m-%d %H:%M")
|
||||
|
||||
|
||||
def imposta(params: dict, ctx: dict) -> str:
|
||||
testo = str(params.get("testo", "")).strip() or "Timer scaduto"
|
||||
try:
|
||||
at = _when(params)
|
||||
except ValueError as err:
|
||||
return f"Errore: {err}"
|
||||
item = {"id": uuid.uuid4().hex[:6], "at": at.strftime("%Y-%m-%d %H:%M"), "testo": testo,
|
||||
"ripeti": "giorno" if params.get("ogni_giorno") else "no", "comando": bool(params.get("esegui_come_comando"))}
|
||||
items = load(); items.append(item); save(items)
|
||||
quando = at.strftime("%H:%M") if at.date() == datetime.now().date() else at.strftime("%d/%m alle %H:%M")
|
||||
tipo = "comando" if item["comando"] else "avviso"
|
||||
return f"{tipo.capitalize()} programmato per le {quando}{' ogni giorno' if item['ripeti'] == 'giorno' else ''}: «{testo}» [id {item['id']}]."
|
||||
|
||||
|
||||
def elenca(params: dict, ctx: dict) -> str:
|
||||
items = sorted(load(), key=lambda x: x["at"])
|
||||
if not items:
|
||||
return "Nessun timer o attività programmata."
|
||||
return "Programmati:\n" + "\n".join(f"- [{i['id']}] {i['at']}{' ogni giorno' if i.get('ripeti') == 'giorno' else ''}: {i['testo']}{' (comando)' if i.get('comando') else ''}" for i in items)
|
||||
|
||||
|
||||
def annulla(params: dict, ctx: dict) -> str:
|
||||
key = str(params.get("id", "")).strip().lower()
|
||||
items = load()
|
||||
keep = [i for i in items if i["id"] != key and key not in i["testo"].lower()] if key else []
|
||||
if key and len(keep) == len(items):
|
||||
return "Nessun timer corrispondente."
|
||||
if not key:
|
||||
save([]); return "Tutti i timer annullati."
|
||||
save(keep)
|
||||
return f"Annullati {len(items) - len(keep)} timer."
|
||||
|
||||
|
||||
def due(now: datetime | None = None) -> list[dict]:
|
||||
"""Restituisce gli elementi scaduti e aggiorna il file (usato dallo scheduler dell'app)."""
|
||||
now = now or datetime.now()
|
||||
items, fired, keep = load(), [], []
|
||||
for i in items:
|
||||
try:
|
||||
at = datetime.strptime(i["at"], "%Y-%m-%d %H:%M")
|
||||
except ValueError:
|
||||
continue
|
||||
if at <= now:
|
||||
fired.append(i)
|
||||
if i.get("ripeti") == "giorno":
|
||||
nxt = at + timedelta(days=1)
|
||||
while nxt <= now:
|
||||
nxt += timedelta(days=1)
|
||||
keep.append({**i, "at": nxt.strftime("%Y-%m-%d %H:%M")})
|
||||
else:
|
||||
keep.append(i)
|
||||
if fired:
|
||||
save(keep)
|
||||
return fired
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "timer_imposta",
|
||||
"description": "Imposta un timer ('tra 10 minuti'), una sveglia/avviso a un'ora, o un'attività programmata anche giornaliera. Con esegui_come_comando=true, all'ora stabilita il testo viene eseguito come richiesta all'assistente (es. 'leggimi la posta'), altrimenti viene annunciato a voce.",
|
||||
"parameters": {"type": "object", "properties": {"testo": {"type": "string", "description": "Cosa annunciare, o il comando da eseguire."},
|
||||
"minuti": {"type": "number", "description": "Fra quanti minuti."},
|
||||
"ora": {"type": "string", "description": "HH:MM oppure AAAA-MM-GG HH:MM."},
|
||||
"ogni_giorno": {"type": "boolean"}, "esegui_come_comando": {"type": "boolean"}}, "required": ["testo"]},
|
||||
"run": imposta},
|
||||
{"name": "timer_elenca", "description": "Elenca timer e attività programmate.", "parameters": {"type": "object", "properties": {}}, "run": elenca},
|
||||
{"name": "timer_annulla", "description": "Annulla un timer per id o parola del testo; senza id annulla tutti.",
|
||||
"parameters": {"type": "object", "properties": {"id": {"type": "string"}}}, "run": annulla},
|
||||
]
|
||||
@@ -0,0 +1,279 @@
|
||||
"""Archivio delle chat WhatsApp esportate (.txt o .zip): indice locale, ricerca, lettura per periodo, riepiloghi."""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import sqlite3
|
||||
import sys
|
||||
import zipfile
|
||||
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 # noqa: E402
|
||||
|
||||
ARCHIVE = DATA_DIR / "whatsapp"
|
||||
DB = ARCHIVE / "index.sqlite"
|
||||
MAX_CHARS = 12000
|
||||
# iOS: [21/11/20, 13:56:41] Nome: testo — Android: 21/11/20, 13:56 - Nome: testo
|
||||
RE_IOS = re.compile(r"^?\[(\d{1,2}/\d{1,2}/\d{2,4}),? (\d{1,2}:\d{2}(?::\d{2})?)\] ([^:]{1,80}?): (.*)$")
|
||||
RE_AND = re.compile(r"^?(\d{1,2}/\d{1,2}/\d{2,4}),? (\d{1,2}:\d{2}(?::\d{2})?) - ([^:]{1,80}?): (.*)$")
|
||||
SKIP = ("crittografati end-to-end", "end-to-end encrypted")
|
||||
MEDIA = {"immagine omessa": "[immagine]", "video omesso": "[video]", "audio omesso": "[audio]", "sticker omesso": "[sticker]",
|
||||
"GIF omessa": "[gif]", "documento omesso": "[documento]", "image omitted": "[immagine]", "video omitted": "[video]"}
|
||||
|
||||
|
||||
def _parse_ts(d: str, t: str) -> datetime | None:
|
||||
for fmt in ("%d/%m/%y %H:%M:%S", "%d/%m/%Y %H:%M:%S", "%d/%m/%y %H:%M", "%d/%m/%Y %H:%M"):
|
||||
try:
|
||||
return datetime.strptime(f"{d} {t}", fmt)
|
||||
except ValueError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def parse(text: str):
|
||||
msgs = []
|
||||
for raw in text.splitlines():
|
||||
line = raw.rstrip("\r").replace("", "")
|
||||
m = RE_IOS.match(line) or RE_AND.match(line)
|
||||
if m:
|
||||
ts = _parse_ts(m.group(1), m.group(2))
|
||||
body = m.group(4).strip()
|
||||
if ts is None or any(s in body for s in SKIP):
|
||||
continue
|
||||
for k, v in MEDIA.items():
|
||||
if body == k:
|
||||
body = v
|
||||
msgs.append([ts, m.group(3).strip(), body])
|
||||
elif msgs and line.strip():
|
||||
msgs[-1][2] += "\n" + line.strip()
|
||||
return msgs
|
||||
|
||||
|
||||
def _db() -> sqlite3.Connection:
|
||||
ARCHIVE.mkdir(parents=True, exist_ok=True)
|
||||
con = sqlite3.connect(DB)
|
||||
con.execute("CREATE TABLE IF NOT EXISTS chats (name TEXT PRIMARY KEY, file TEXT, mtime REAL, count INTEGER, first TEXT, last TEXT, participants TEXT)")
|
||||
con.execute("CREATE TABLE IF NOT EXISTS messages (id INTEGER PRIMARY KEY, chat TEXT, ts TEXT, sender TEXT, text TEXT)")
|
||||
con.execute("CREATE INDEX IF NOT EXISTS idx_chat_ts ON messages(chat, ts)")
|
||||
con.execute("CREATE VIRTUAL TABLE IF NOT EXISTS fts USING fts5(text, content='messages', content_rowid='id', tokenize='unicode61 remove_diacritics 2')")
|
||||
con.execute("CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN INSERT INTO fts(rowid, text) VALUES (new.id, new.text); END")
|
||||
con.execute("CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN INSERT INTO fts(fts, rowid, text) VALUES ('delete', old.id, old.text); END")
|
||||
for col in ("source TEXT", "jid TEXT", "from_me INTEGER DEFAULT 0"):
|
||||
try:
|
||||
con.execute(f"ALTER TABLE messages ADD COLUMN {col}")
|
||||
except sqlite3.OperationalError:
|
||||
pass
|
||||
con.execute("PRAGMA journal_mode=WAL")
|
||||
return con
|
||||
|
||||
|
||||
def _index_file(con: sqlite3.Connection, name: str, path: Path) -> int:
|
||||
msgs = parse(path.read_text(encoding="utf-8", errors="replace"))
|
||||
con.execute("DELETE FROM messages WHERE chat = ? AND (source IS NULL OR source = 'export')", (name,))
|
||||
con.executemany("INSERT INTO messages (chat, ts, sender, text, source) VALUES (?, ?, ?, ?, 'export')",
|
||||
[(name, ts.strftime("%Y-%m-%d %H:%M:%S"), s, t) for ts, s, t in msgs])
|
||||
con.execute("INSERT INTO fts(fts) VALUES ('rebuild')")
|
||||
senders = sorted({s for _, s, _ in msgs}, key=lambda s: -sum(1 for _, x, _ in msgs if x == s))
|
||||
con.execute("INSERT OR REPLACE INTO chats VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(name, str(path), path.stat().st_mtime, len(msgs), msgs[0][0].strftime("%Y-%m-%d") if msgs else "",
|
||||
msgs[-1][0].strftime("%Y-%m-%d") if msgs else "", ", ".join(senders[:6])))
|
||||
con.commit()
|
||||
return len(msgs)
|
||||
|
||||
|
||||
def _sync(con: sqlite3.Connection) -> list[str]:
|
||||
"""Indicizza i .txt nuovi o modificati presenti nella cartella archivio."""
|
||||
ARCHIVE.mkdir(parents=True, exist_ok=True)
|
||||
known = {r[0]: r for r in con.execute("SELECT name, file, mtime FROM chats")}
|
||||
done = []
|
||||
for path in sorted(ARCHIVE.glob("*.txt")):
|
||||
name = path.stem
|
||||
if name in known and abs(known[name][2] - path.stat().st_mtime) < 1:
|
||||
continue
|
||||
n = _index_file(con, name, path)
|
||||
done.append(f"{name} ({n} messaggi)")
|
||||
return done
|
||||
|
||||
|
||||
def _chat_name(con: sqlite3.Connection, chat: str) -> str | None:
|
||||
names = [r[0] for r in con.execute("SELECT name FROM chats")]
|
||||
if not chat:
|
||||
return names[0] if len(names) == 1 else None
|
||||
q = chat.lower()
|
||||
return next((n for n in names if n.lower() == q), None) or next((n for n in names if q in n.lower()), None)
|
||||
|
||||
|
||||
def _range(params: dict) -> tuple[str | None, str | None]:
|
||||
da, a = str(params.get("da", "")).strip(), str(params.get("a", "")).strip()
|
||||
today = datetime.now().date()
|
||||
alias = {"oggi": (today, today), "ieri": (today - timedelta(days=1), today - timedelta(days=1)),
|
||||
"settimana": (today - timedelta(days=7), today), "mese": (today - timedelta(days=30), today)}
|
||||
if da.lower() in alias:
|
||||
d1, d2 = alias[da.lower()]
|
||||
return d1.isoformat(), d2.isoformat() + " 23:59:59"
|
||||
return (da or None), ((a[:10] + " 23:59:59") if a else None)
|
||||
|
||||
|
||||
def _fmt(rows) -> str:
|
||||
return "\n".join(f"[{ts[:16]}] {s}: {t}" for ts, s, t in rows)
|
||||
|
||||
|
||||
def importa(params: dict, ctx: dict) -> str:
|
||||
src = Path(str(params.get("percorso", ""))).expanduser()
|
||||
nome = str(params.get("nome", "")).strip()
|
||||
if not src.exists():
|
||||
return f"Percorso non trovato: {src}"
|
||||
ARCHIVE.mkdir(parents=True, exist_ok=True)
|
||||
txt: Path | None = None
|
||||
if src.is_dir():
|
||||
cands = list(src.glob("*.txt"))
|
||||
txt = next((c for c in cands if c.name == "_chat.txt"), cands[0] if cands else None)
|
||||
elif src.suffix.lower() == ".zip":
|
||||
with zipfile.ZipFile(src) as z:
|
||||
members = [m for m in z.namelist() if m.endswith(".txt") and not m.startswith("__MACOSX") and not Path(m).name.startswith(".")]
|
||||
member = next((m for m in members if Path(m).name == "_chat.txt"), members[0] if members else None)
|
||||
if not member:
|
||||
return "Nello zip non c'è un file .txt."
|
||||
tmp = ARCHIVE / "_import.txt"
|
||||
tmp.write_bytes(z.read(member))
|
||||
txt = tmp
|
||||
else:
|
||||
txt = src
|
||||
if txt is None:
|
||||
return "Nessun file .txt trovato."
|
||||
con = _db()
|
||||
if not nome:
|
||||
# Gli zip di WhatsApp si chiamano "WhatsApp Chat - Nome.zip": usa quel nome.
|
||||
m = re.match(r"^WhatsApp Chat - (.+)$", src.stem, flags=re.I)
|
||||
if m:
|
||||
nome = m.group(1).strip()
|
||||
if not nome:
|
||||
msgs = parse(txt.read_text(encoding="utf-8", errors="replace"))
|
||||
senders = {}
|
||||
for _, s, _ in msgs:
|
||||
senders[s] = senders.get(s, 0) + 1
|
||||
top = sorted(senders, key=senders.get, reverse=True)
|
||||
nome = " & ".join(top[:2]) if len(top) > 1 else (top[0] if top else src.stem)
|
||||
safe = re.sub(r'[\\/:*?"<>|]+', "_", nome).strip() or "chat"
|
||||
dest = ARCHIVE / f"{safe}.txt"
|
||||
if txt.resolve() != dest.resolve():
|
||||
shutil.copyfile(txt, dest)
|
||||
if (ARCHIVE / "_import.txt").exists():
|
||||
(ARCHIVE / "_import.txt").unlink()
|
||||
n = _index_file(con, safe, dest)
|
||||
con.close()
|
||||
return f"Chat importata come «{safe}»: {n} messaggi indicizzati."
|
||||
|
||||
|
||||
def importa_cartella(params: dict, ctx: dict) -> str:
|
||||
folder = Path(str(params.get("percorso", ""))).expanduser()
|
||||
if not folder.is_dir():
|
||||
return f"Cartella non trovata: {folder}"
|
||||
files = sorted([p for p in folder.iterdir() if p.suffix.lower() in (".zip", ".txt") and not p.name.startswith(".")])
|
||||
if not files:
|
||||
return "Nessun file .zip o .txt nella cartella."
|
||||
results, errors = [], []
|
||||
for f in files:
|
||||
try:
|
||||
results.append(importa({"percorso": str(f)}, ctx))
|
||||
except Exception as err:
|
||||
errors.append(f"{f.name}: {err}")
|
||||
out = f"Importate {len(results)} chat:\n" + "\n".join("- " + r.replace("Chat importata come ", "") for r in results)
|
||||
if errors:
|
||||
out += "\nErrori:\n" + "\n".join("- " + e for e in errors)
|
||||
return out
|
||||
|
||||
|
||||
def elenca(params: dict, ctx: dict) -> str:
|
||||
con = _db()
|
||||
_sync(con)
|
||||
rows = con.execute("SELECT name, count, first, last, participants FROM chats ORDER BY name").fetchall()
|
||||
con.close()
|
||||
if not rows:
|
||||
return f"Archivio vuoto. Esporta una chat da WhatsApp e importala con whatsapp_importa, oppure copia il .txt in {ARCHIVE}."
|
||||
return "Chat in archivio:\n" + "\n".join(f"- {n}: {c} messaggi dal {f} al {l} (partecipanti: {p})" for n, c, f, l, p in rows)
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query", "")).strip()
|
||||
if not q:
|
||||
return "Errore: serve la query."
|
||||
lim = max(1, min(int(params.get("limite") or 15), 60))
|
||||
con = _db()
|
||||
_sync(con)
|
||||
chat = _chat_name(con, str(params.get("chat", "")))
|
||||
da, a = _range(params)
|
||||
sql = "SELECT m.ts, m.sender, m.text FROM fts JOIN messages m ON m.id = fts.rowid WHERE fts MATCH ?"
|
||||
args: list = ['"' + q.replace('"', '') + '"*' if " " not in q else " ".join('"' + w.replace('"', '') + '"' for w in q.split())]
|
||||
if chat:
|
||||
sql += " AND m.chat = ?"; args.append(chat)
|
||||
if da:
|
||||
sql += " AND m.ts >= ?"; args.append(da)
|
||||
if a:
|
||||
sql += " AND m.ts <= ?"; args.append(a)
|
||||
sql += " ORDER BY m.ts DESC LIMIT ?"; args.append(lim)
|
||||
rows = con.execute(sql, args).fetchall()
|
||||
con.close()
|
||||
if not rows:
|
||||
return f"Nessun messaggio contiene '{q}'."
|
||||
return f"Messaggi con '{q}' ({len(rows)}, i più recenti):\n" + _fmt(reversed(rows))
|
||||
|
||||
|
||||
def leggi(params: dict, ctx: dict) -> str:
|
||||
con = _db()
|
||||
_sync(con)
|
||||
chat = _chat_name(con, str(params.get("chat", "")))
|
||||
if chat is None:
|
||||
con.close()
|
||||
return "Chat non trovata: usa whatsapp_chat_elenca per vedere i nomi."
|
||||
da, a = _range(params)
|
||||
n = max(1, min(int(params.get("ultimi") or 0) or 0, 400))
|
||||
if da or a:
|
||||
sql, args = "SELECT ts, sender, text FROM messages WHERE chat = ?", [chat]
|
||||
if da:
|
||||
sql += " AND ts >= ?"; args.append(da)
|
||||
if a:
|
||||
sql += " AND ts <= ?"; args.append(a)
|
||||
rows = con.execute(sql + " ORDER BY ts LIMIT 400", args).fetchall()
|
||||
else:
|
||||
rows = list(reversed(con.execute("SELECT ts, sender, text FROM messages WHERE chat = ? ORDER BY ts DESC LIMIT ?", (chat, n or 30)).fetchall()))
|
||||
con.close()
|
||||
if not rows:
|
||||
return "Nessun messaggio nel periodo."
|
||||
text = _fmt(rows)
|
||||
if len(text) > MAX_CHARS:
|
||||
text = text[-MAX_CHARS:]
|
||||
text = "[…]\n" + text[text.find("\n") + 1:]
|
||||
return f"Chat «{chat}», {len(rows)} messaggi:\n{text}"
|
||||
|
||||
|
||||
def statistiche(params: dict, ctx: dict) -> str:
|
||||
con = _db()
|
||||
_sync(con)
|
||||
chat = _chat_name(con, str(params.get("chat", "")))
|
||||
if chat is None:
|
||||
con.close()
|
||||
return "Chat non trovata."
|
||||
rows = con.execute("SELECT sender, COUNT(*) FROM messages WHERE chat = ? GROUP BY sender ORDER BY 2 DESC", (chat,)).fetchall()
|
||||
months = con.execute("SELECT substr(ts,1,7), COUNT(*) FROM messages WHERE chat = ? GROUP BY 1 ORDER BY 1 DESC LIMIT 12", (chat,)).fetchall()
|
||||
con.close()
|
||||
return (f"Chat «{chat}»: " + ", ".join(f"{s} {c} messaggi" for s, c in rows) + ".\nUltimi mesi: " +
|
||||
", ".join(f"{m} {c}" for m, c in months))
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "whatsapp_importa", "description": "Importa una chat WhatsApp esportata (.txt, .zip o cartella con _chat.txt) nell'archivio locale e la indicizza. Il nome è facoltativo: altrimenti usa i partecipanti.",
|
||||
"parameters": {"type": "object", "properties": {"percorso": {"type": "string"}, "nome": {"type": "string"}}, "required": ["percorso"]}, "run": importa},
|
||||
{"name": "whatsapp_importa_cartella", "description": "Importa tutte le chat WhatsApp esportate (.zip o .txt) presenti in una cartella.",
|
||||
"parameters": {"type": "object", "properties": {"percorso": {"type": "string"}}, "required": ["percorso"]}, "run": importa_cartella},
|
||||
{"name": "whatsapp_chat_elenca", "description": "Elenca le chat WhatsApp in archivio con numero di messaggi, periodo e partecipanti.",
|
||||
"parameters": {"type": "object", "properties": {}}, "run": elenca},
|
||||
{"name": "whatsapp_cerca", "description": "Cerca parole nei messaggi WhatsApp archiviati (tutte le chat o una sola), con periodo facoltativo (da/a AAAA-MM-GG oppure da='settimana'|'mese').",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}, "chat": {"type": "string"}, "da": {"type": "string"}, "a": {"type": "string"}, "limite": {"type": "integer"}}, "required": ["query"]}, "run": cerca},
|
||||
{"name": "whatsapp_leggi", "description": "Legge i messaggi di una chat: gli ultimi N oppure un periodo (da/a). Usalo per riassumere o rispondere a domande su cosa è stato detto.",
|
||||
"parameters": {"type": "object", "properties": {"chat": {"type": "string"}, "ultimi": {"type": "integer"}, "da": {"type": "string"}, "a": {"type": "string"}}}, "run": leggi},
|
||||
{"name": "whatsapp_statistiche", "description": "Quanti messaggi per persona e per mese in una chat.",
|
||||
"parameters": {"type": "object", "properties": {"chat": {"type": "string"}}}, "run": statistiche},
|
||||
]
|
||||
@@ -0,0 +1,76 @@
|
||||
"""WhatsApp in tempo reale (dispositivo collegato): novità, stato e invio con conferma."""
|
||||
from __future__ import annotations
|
||||
|
||||
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.macos import confirm_or_param, CONFERMATO # noqa: E402
|
||||
from avatar import whatsapp_bridge as wb # noqa: E402
|
||||
from avatar.settings import DATA_DIR # noqa: E402
|
||||
|
||||
DB = DATA_DIR / "whatsapp" / "index.sqlite"
|
||||
|
||||
|
||||
def stato(params: dict, ctx: dict) -> str:
|
||||
return wb.status_text()
|
||||
|
||||
|
||||
def novita(params: dict, ctx: dict) -> str:
|
||||
minuti = int(params.get("minuti") or 0)
|
||||
lim = max(1, min(int(params.get("limite") or 20), 60))
|
||||
if not DB.exists():
|
||||
return "Nessun messaggio ricevuto finora."
|
||||
since = (datetime.now() - timedelta(minutes=minuti)).strftime("%Y-%m-%d %H:%M:%S") if minuti else (datetime.now().strftime("%Y-%m-%d") + " 00:00:00")
|
||||
con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
|
||||
try:
|
||||
rows = con.execute("SELECT ts, chat, sender, text FROM messages WHERE source = 'live' AND from_me = 0 AND ts >= ? ORDER BY ts DESC LIMIT ?", (since, lim)).fetchall()
|
||||
finally:
|
||||
con.close()
|
||||
if not rows:
|
||||
return "Nessun messaggio WhatsApp nuovo" + (f" negli ultimi {minuti} minuti." if minuti else " oggi.") + (" " + wb.status_text() if not wb.running() else "")
|
||||
lines = [f"[{ts[11:16]}] {chat}{' — ' + sender if sender != chat else ''}: {text[:200]}" for ts, chat, sender, text in reversed(rows)]
|
||||
return f"Messaggi WhatsApp ricevuti ({len(rows)}):\n" + "\n".join(lines)
|
||||
|
||||
|
||||
def invia(params: dict, ctx: dict) -> str:
|
||||
a = str(params.get("a", "")).strip()
|
||||
testo = str(params.get("testo", "")).strip()
|
||||
if not a or not testo:
|
||||
return "Errore: servono destinatario (nome del contatto o del gruppo, oppure numero) e testo."
|
||||
if not wb.running():
|
||||
wb.start()
|
||||
try:
|
||||
res = wb.resolve(a)
|
||||
except Exception as err:
|
||||
return f"WhatsApp non disponibile: {err}"
|
||||
if not res.get("jid"):
|
||||
return f"Non trovo '{a}' tra i contatti WhatsApp: prova con il numero di telefono."
|
||||
name = res.get("name") or a
|
||||
|
||||
def do() -> str:
|
||||
out = wb.send(a, testo)
|
||||
msg = f"Messaggio WhatsApp inviato a {out.get('name') or name}."
|
||||
if ctx.get("say"):
|
||||
ctx["say"](msg)
|
||||
return msg
|
||||
|
||||
return confirm_or_param(ctx, params, "whatsapp_invia", "Inviare su WhatsApp?", f"A: {name}\n{testo[:200]}", do)
|
||||
|
||||
|
||||
def aggiorna_contatti(params: dict, ctx: dict) -> str:
|
||||
n = wb.sync_contacts()
|
||||
return f"Rubrica sincronizzata: {n} numeri associati ai nomi."
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "whatsapp_aggiorna_contatti", "description": "Associa i numeri WhatsApp ai nomi della rubrica del Mac (da usare se le chat compaiono come numeri).",
|
||||
"parameters": {"type": "object", "properties": {}}, "run": aggiorna_contatti},
|
||||
{"name": "whatsapp_novita", "description": "Messaggi WhatsApp ricevuti in tempo reale oggi o negli ultimi N minuti (tutte le chat). Usalo per 'ci sono novità su WhatsApp?', 'chi mi ha scritto?'.",
|
||||
"parameters": {"type": "object", "properties": {"minuti": {"type": "integer"}, "limite": {"type": "integer"}}}, "run": novita},
|
||||
{"name": "whatsapp_invia", "description": "Invia un messaggio WhatsApp a un contatto, a un gruppo o a un numero. Azione irreversibile con conferma sullo schermo; con [CONFIRMATION_PENDING] di' all'utente di confermare e non dire che è inviato.",
|
||||
"parameters": {"type": "object", "properties": {"a": {"type": "string"}, "testo": {"type": "string"}, "confermato": CONFERMATO}, "required": ["a", "testo"]}, "run": invia},
|
||||
{"name": "whatsapp_stato", "description": "Stato del collegamento WhatsApp in tempo reale.", "parameters": {"type": "object", "properties": {}}, "run": stato},
|
||||
]
|
||||
Reference in new issue
Block a user