211 lines
11 KiB
Python
211 lines
11 KiB
Python
"""Newsletter e posta indesiderata: trova i mittenti che offrono la disiscrizione standard (List-Unsubscribe),
|
|
si disiscrive con conferma (one-click, email o link) e blocca i mittenti senza via d'uscita con una regola di Mail.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
import mail # noqa: E402 (plugin mail: indice, emlx, invio)
|
|
from avatar.macos import confirm_or_param, CONFERMATO # noqa: E402
|
|
from avatar.settings import DATA_DIR # noqa: E402
|
|
|
|
STATO = DATA_DIR / "disiscrizioni.json"
|
|
_last: dict = {"senders": []}
|
|
CARTELLE = {"arrivo": "(mb.url LIKE '%/INBOX' OR mb.url LIKE '%/Inbox')",
|
|
"indesiderata": "(mb.url LIKE '%Junk%' OR mb.url LIKE '%ndesiderata%' OR mb.url LIKE '%Spam%')"}
|
|
|
|
|
|
def _load() -> dict:
|
|
try:
|
|
return json.loads(STATO.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save(d: dict) -> None:
|
|
STATO.parent.mkdir(parents=True, exist_ok=True)
|
|
STATO.write_text(json.dumps(d, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def _parse_lu(value: str) -> tuple[str, str]:
|
|
"""(url https, mailto) dall'intestazione List-Unsubscribe."""
|
|
urls = re.findall(r"<([^>]+)>", value or "") or [value or ""]
|
|
http = next((u.strip() for u in urls if u.strip().lower().startswith("http")), "")
|
|
mailto = next((u.strip() for u in urls if u.strip().lower().startswith("mailto:")), "")
|
|
return http, mailto
|
|
|
|
|
|
def scansiona(giorni: int = 30, cartella: str = "arrivo", max_msg: int = 400) -> list[dict]:
|
|
where = " OR ".join(CARTELLE[c] for c in (("arrivo", "indesiderata") if cartella == "tutte" else (cartella,)))
|
|
con = sqlite3.connect(f"file:{mail.INDEX}?mode=ro", uri=True, timeout=5)
|
|
try:
|
|
rows = con.execute(f"""SELECT m.ROWID, mb.url, COALESCE(a.address,''), COALESCE(a.comment,''), COALESCE(s.subject,''), m.date_received
|
|
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
|
|
WHERE m.deleted=0 AND ({where}) AND m.date_received > ? ORDER BY m.date_received DESC LIMIT ?""", (time.time() - giorni * 86400, max_msg)).fetchall()
|
|
finally:
|
|
con.close()
|
|
senders: dict[str, dict] = {}
|
|
for rid, url, addr, name, subj, ts in rows:
|
|
addr = addr.lower()
|
|
if not addr:
|
|
continue
|
|
s = senders.setdefault(addr, {"addr": addr, "nome": name or addr, "n": 0, "oggetto": subj, "tipo": "nessuno", "http": "", "mailto": "", "oneclick": False})
|
|
s["n"] += 1
|
|
if s["tipo"] != "nessuno" and s["oneclick"]:
|
|
continue
|
|
p = mail._find_emlx(rid, url)
|
|
if not p:
|
|
continue
|
|
try:
|
|
msg = mail._emlx_message(p)
|
|
except Exception:
|
|
continue
|
|
lu = msg.get("List-Unsubscribe")
|
|
if not lu:
|
|
continue
|
|
http, mailto = _parse_lu(str(lu))
|
|
oneclick = bool(http) and "one-click" in str(msg.get("List-Unsubscribe-Post", "")).lower()
|
|
s.update(http=http or s["http"], mailto=mailto or s["mailto"], oneclick=s["oneclick"] or oneclick)
|
|
s["tipo"] = "one-click" if s["oneclick"] else ("email" if s["mailto"] else "link")
|
|
fatte = _load()
|
|
out = sorted(senders.values(), key=lambda s: (-(s["tipo"] != "nessuno"), -s["n"]))
|
|
for i, s in enumerate(out, 1):
|
|
s["k"] = i
|
|
s["fatta"] = fatte.get(s["addr"], {}).get("esito", "")
|
|
_last["senders"] = out
|
|
return out
|
|
|
|
|
|
def elenco(params: dict, ctx: dict) -> str:
|
|
giorni = max(3, min(int(params.get("giorni") or 30), 365))
|
|
cartella = str(params.get("cartella") or "arrivo").lower()
|
|
if cartella not in ("arrivo", "indesiderata", "tutte"):
|
|
cartella = "arrivo"
|
|
if ctx.get("player") is not None and hasattr(ctx["player"], "set_state"):
|
|
ctx["player"].set_state("PROCESSING · analizzo le intestazioni delle email")
|
|
senders = scansiona(giorni, cartella)
|
|
if not senders:
|
|
return f"Nessun messaggio negli ultimi {giorni} giorni nella cartella scelta."
|
|
con = [s for s in senders if s["tipo"] != "nessuno"]
|
|
senza = [s for s in senders if s["tipo"] == "nessuno"]
|
|
lim = max(5, min(int(params.get("limite") or 40), 80))
|
|
lines = [f"{s['k']}. {s['nome'][:32]} <{s['addr']}> — {s['n']} msg, disiscrizione {s['tipo']}{' (già fatta: ' + s['fatta'] + ')' if s['fatta'] else ''} — «{s['oggetto'][:50]}»" for s in con[:lim]]
|
|
out = [f"Mittenti con disiscrizione disponibile ({len(con)}) negli ultimi {giorni} giorni, cartella {cartella}:"] + lines
|
|
if senza:
|
|
out.append(f"Altri {len(senza)} mittenti senza disiscrizione automatica: in genere notifiche di servizio (banca, ordini, assistenza) da lasciare; se qualcuno è spam, usa mail_blocca invece di cliccare link. I più frequenti: " + ", ".join(f"{s['nome'][:20]} ({s['n']})" for s in senza[:8]) + ("…" if len(senza) > 8 else ""))
|
|
out.append("Per procedere: mail_disiscrivi con i numeri (es. '1,2,5') o 'tutti'.")
|
|
return "\n".join(out)
|
|
|
|
|
|
def _oneclick(url: str) -> str:
|
|
req = urllib.request.Request(url, data=b"List-Unsubscribe=One-Click", method="POST",
|
|
headers={"Content-Type": "application/x-www-form-urlencoded", "User-Agent": "Mozilla/5.0 (Macintosh) LuZa"})
|
|
with urllib.request.urlopen(req, timeout=20) as r:
|
|
return f"one-click ok (HTTP {r.status})"
|
|
|
|
|
|
def _link(url: str) -> str:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (Macintosh) LuZa"})
|
|
with urllib.request.urlopen(req, timeout=20) as r:
|
|
body = r.read(200000).decode("utf-8", errors="replace").lower()
|
|
ok = any(w in body for w in ("unsubscribed", "disiscritt", "cancellat", "removed", "rimoss", "success", "non riceverai", "you have been", "you will no longer"))
|
|
return f"link aperto (HTTP {r.status}){', conferma nella pagina' if ok else ', esito da verificare: la pagina potrebbe chiedere un clic'}"
|
|
|
|
|
|
def _mailto(mailto: str) -> str:
|
|
u = urllib.parse.urlparse(mailto)
|
|
q = urllib.parse.parse_qs(u.query)
|
|
to = u.path
|
|
subject = q.get("subject", ["unsubscribe"])[0]
|
|
body = q.get("body", ["unsubscribe"])[0]
|
|
mail._osa(mail.SEND_SCRIPT, to, subject, body)
|
|
return f"email di disiscrizione inviata a {to}"
|
|
|
|
|
|
def disiscrivi(params: dict, ctx: dict) -> str:
|
|
if not _last["senders"]:
|
|
scansiona()
|
|
sel = str(params.get("mittenti") or "").strip().lower()
|
|
if not sel:
|
|
return "Indica i mittenti: numeri dell'elenco (es. '1,3') o 'tutti'."
|
|
cand = [s for s in _last["senders"] if s["tipo"] != "nessuno"]
|
|
if sel in ("tutti", "tutte", "all"):
|
|
chosen = cand
|
|
else:
|
|
keys = {k.strip() for k in re.split(r"[,\s]+", sel) if k.strip()}
|
|
chosen = [s for s in cand if str(s["k"]) in keys or s["addr"] in keys or any(k in s["addr"] or k in s["nome"].lower() for k in keys if len(k) > 3)]
|
|
if not chosen:
|
|
return "Nessun mittente corrispondente con disiscrizione disponibile."
|
|
desc = "\n".join(f"• {s['nome'][:28]} <{s['addr']}> ({s['tipo']})" for s in chosen[:12]) + (f"\n… e altri {len(chosen) - 12}" if len(chosen) > 12 else "")
|
|
|
|
def do() -> str:
|
|
stato = _load(); res = []
|
|
for s in chosen:
|
|
if ctx.get("player") is not None and hasattr(ctx["player"], "set_state"):
|
|
ctx["player"].set_state(f"PROCESSING · disiscrizione da {s['nome'][:20]}")
|
|
try:
|
|
if s["oneclick"]:
|
|
esito = _oneclick(s["http"])
|
|
elif s["mailto"]:
|
|
esito = _mailto(s["mailto"])
|
|
elif s["http"]:
|
|
esito = _link(s["http"])
|
|
else:
|
|
esito = "nessun metodo"
|
|
except Exception as err:
|
|
esito = f"fallita: {str(err)[:80]}"
|
|
stato[s["addr"]] = {"esito": esito, "quando": time.strftime("%Y-%m-%d"), "nome": s["nome"]}
|
|
res.append(f"- {s['nome'][:30]}: {esito}")
|
|
_save(stato)
|
|
ok = sum(1 for r in res if "fallita" not in r and "nessun" not in r)
|
|
return f"Disiscrizioni eseguite: {ok} su {len(chosen)}.\n" + "\n".join(res) + "\nLe email già ricevute restano; i mittenti seri smettono entro pochi giorni."
|
|
|
|
return confirm_or_param(ctx, params, "mail_disiscrivi", f"Disiscriversi da {len(chosen)} mittenti?", desc, do)
|
|
|
|
|
|
BLOCK_SCRIPT = '''on run argv
|
|
set addr to item 1 of argv
|
|
tell application "Mail"
|
|
set r to make new rule at end of rules with properties {name:"LuZa: blocca " & addr, enabled:true, delete message:true, stop evaluating rules:true}
|
|
tell r
|
|
make new rule condition at end of rule conditions with properties {rule type:from header, qualifier:does contain value, expression:addr}
|
|
end tell
|
|
end tell
|
|
return "ok"
|
|
end run'''
|
|
|
|
|
|
def blocca(params: dict, ctx: dict) -> str:
|
|
addr = str(params.get("mittente") or "").strip().lower()
|
|
if "@" not in addr:
|
|
m = next((s for s in _last["senders"] if str(s["k"]) == addr or addr in s["nome"].lower()), None)
|
|
if not m:
|
|
return "Indica l'indirizzo email del mittente da bloccare."
|
|
addr = m["addr"]
|
|
|
|
def do() -> str:
|
|
mail._osa(BLOCK_SCRIPT, addr)
|
|
return f"Regola creata in Mail: le email da {addr} vengono cestinate all'arrivo."
|
|
|
|
return confirm_or_param(ctx, params, "mail_blocca", "Bloccare questo mittente in Mail?", f"{addr}\n(regola: cestina all'arrivo)", do)
|
|
|
|
|
|
TOOLS = [
|
|
{"name": "mail_newsletter", "description": "Trova i mittenti di newsletter e promozioni negli ultimi giorni (cartella: arrivo, indesiderata o tutte) e per ciascuno dice se offre la disiscrizione standard (one-click, email o link) e quante email ha mandato. Usalo per 'disiscrivimi dalle newsletter', 'chi mi riempie la posta?'. Nota: la posta indesiderata su questo Mac è quasi vuota, la maggior parte delle newsletter è nella posta in arrivo.",
|
|
"parameters": {"type": "object", "properties": {"giorni": {"type": "integer"}, "cartella": {"type": "string", "enum": ["arrivo", "indesiderata", "tutte"]}, "limite": {"type": "integer"}}}, "run": elenco},
|
|
{"name": "mail_disiscrivi", "description": "Esegue la disiscrizione dai mittenti scelti nell'ultimo elenco di mail_newsletter: mittenti = numeri separati da virgola (es. '1,2,5'), indirizzi, oppure 'tutti'. Usa il metodo standard di ciascuno (one-click, email automatica o link). Chiede conferma sullo schermo; con [CONFIRMATION_PENDING] di' all'utente di confermare.",
|
|
"parameters": {"type": "object", "properties": {"mittenti": {"type": "string"}, "confermato": CONFERMATO}, "required": ["mittenti"]}, "run": disiscrivi},
|
|
{"name": "mail_blocca", "description": "Blocca un mittente creando una regola in Mail che cestina le sue email all'arrivo (per lo spam senza disiscrizione). Chiede conferma.",
|
|
"parameters": {"type": "object", "properties": {"mittente": {"type": "string", "description": "indirizzo email, oppure numero o nome dall'ultimo elenco"}, "confermato": CONFERMATO}, "required": ["mittente"]}, "run": blocca},
|
|
]
|