152 lines
5.4 KiB
Python
152 lines
5.4 KiB
Python
"""Subscriptions: the bulk senders mailing a user, and unsubscribing from them.
|
|||
|
|
|
||
|
|
No service records what a user signed up for, so the list is worked out from
|
||
|
|
received mail, as Gmail does: a message is a subscription's when it carries a
|
||
|
|
``List-Unsubscribe`` header (required of bulk senders by Gmail and Yahoo since
|
||
|
|
2024). Messages are grouped by ``List-Id`` when there is one (a company can run
|
||
|
|
several lists), otherwise by sender address.
|
||
|
|
|
||
|
|
The headers are read when a message is stored (indexer.persist_message); older
|
||
|
|
mail is read once by :func:`backfill_list_headers_job`. Unsubscribing uses the
|
||
|
|
sender's one-click POST (RFC 8058) when offered, else mails the ``mailto:``
|
||
|
|
address, else hands the web page back for the user to open.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import re
|
||
|
|
import uuid
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from email.parser import BytesHeaderParser
|
||
|
|
from email.policy import default as default_policy
|
||
|
|
from typing import Any
|
||
|
|
from urllib.parse import parse_qs, unquote, urlsplit
|
||
|
|
|
||
|
|
from sqlalchemy import select, update
|
||
|
|
|
||
|
|
from app.db import session_scope
|
||
|
|
from app.indexer import rawstore
|
||
|
|
from app.models import Message
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
_MAX_HEADER = 2000
|
||
|
|
_ANGLE = re.compile(r"<([^<>]+)>")
|
||
|
|
_BATCH = 200
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class ListHeaders:
|
||
|
|
list_id: str | None
|
||
|
|
unsubscribe: str # '' when the message has none
|
||
|
|
one_click: bool
|
||
|
|
|
||
|
|
|
||
|
|
def read_list_headers(raw: bytes) -> ListHeaders:
|
||
|
|
"""The mailing-list headers of a raw message (headers only are parsed)."""
|
||
|
|
end = raw.find(b"\r\n\r\n")
|
||
|
|
if end < 0:
|
||
|
|
end = raw.find(b"\n\n")
|
||
|
|
head = raw if end < 0 else raw[: end + 2]
|
||
|
|
try:
|
||
|
|
msg = BytesHeaderParser(policy=default_policy).parsebytes(head)
|
||
|
|
list_id_raw = str(msg.get("List-Id") or "")
|
||
|
|
unsubscribe = str(msg.get("List-Unsubscribe") or "")
|
||
|
|
post = str(msg.get("List-Unsubscribe-Post") or "")
|
||
|
|
except Exception: # a malformed header must not stop the message being stored
|
||
|
|
return ListHeaders(None, "", False)
|
||
|
|
list_id = None
|
||
|
|
if list_id_raw:
|
||
|
|
match = _ANGLE.search(list_id_raw)
|
||
|
|
list_id = (match.group(1) if match else list_id_raw).strip().lower()[:300] or None
|
||
|
|
unsubscribe = " ".join(unsubscribe.split())[:_MAX_HEADER]
|
||
|
|
one_click = "list-unsubscribe=one-click" in post.replace(" ", "").lower()
|
||
|
|
return ListHeaders(list_id, unsubscribe, one_click and bool(unsubscribe))
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class UnsubscribeTargets:
|
||
|
|
https: str | None
|
||
|
|
mailto: str | None # the address
|
||
|
|
mailto_subject: str
|
||
|
|
mailto_body: str
|
||
|
|
|
||
|
|
|
||
|
|
def unsubscribe_targets(header: str) -> UnsubscribeTargets:
|
||
|
|
"""The https and mailto targets listed in a ``List-Unsubscribe`` header."""
|
||
|
|
https = mailto = None
|
||
|
|
subject, body = "unsubscribe", "unsubscribe"
|
||
|
|
for target in _ANGLE.findall(header):
|
||
|
|
target = target.strip()
|
||
|
|
lowered = target.lower()
|
||
|
|
if https is None and lowered.startswith("https://"):
|
||
|
|
https = target
|
||
|
|
elif mailto is None and lowered.startswith("mailto:"):
|
||
|
|
parts = urlsplit(target)
|
||
|
|
address = unquote(parts.path).strip()
|
||
|
|
if "@" in address and "\n" not in address and "\r" not in address:
|
||
|
|
mailto = address
|
||
|
|
query = parse_qs(parts.query)
|
||
|
|
subject = (query.get("subject") or [subject])[0][:200]
|
||
|
|
body = (query.get("body") or [body])[0][:2000]
|
||
|
|
return UnsubscribeTargets(https, mailto, subject, body)
|
||
|
|
|
||
|
|
|
||
|
|
def group_key(list_id: str | None, sender: str) -> str:
|
||
|
|
return f"list:{list_id}" if list_id else f"from:{sender.lower()}"
|
||
|
|
|
||
|
|
|
||
|
|
async def pending_count(session: Any, user_id: uuid.UUID) -> int:
|
||
|
|
from sqlalchemy import func
|
||
|
|
|
||
|
|
return int(
|
||
|
|
await session.scalar(
|
||
|
|
select(func.count())
|
||
|
|
.select_from(Message)
|
||
|
|
.where(Message.user_id == user_id, Message.list_unsubscribe.is_(None))
|
||
|
|
)
|
||
|
|
or 0
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
async def backfill_list_headers_job(ctx: dict[str, Any], user_id: str) -> int:
|
||
|
|
"""Read the list headers of a user's older mail (stored before this feature).
|
||
|
|
|
||
|
|
Works in batches; a message whose raw copy can't be had is marked as having
|
||
|
|
none, so it is not tried again. Returns how many messages were read.
|
||
|
|
"""
|
||
|
|
uid = uuid.UUID(user_id)
|
||
|
|
done = 0
|
||
|
|
while True:
|
||
|
|
async with session_scope() as session:
|
||
|
|
batch = list(
|
||
|
|
await session.scalars(
|
||
|
|
select(Message)
|
||
|
|
.where(Message.user_id == uid, Message.list_unsubscribe.is_(None))
|
||
|
|
.limit(_BATCH)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
if not batch:
|
||
|
|
break
|
||
|
|
for message in batch:
|
||
|
|
try:
|
||
|
|
raw = await rawstore.ensure_bytes(message)
|
||
|
|
headers = read_list_headers(raw)
|
||
|
|
except Exception: # missing blob, server unreachable: give up on it
|
||
|
|
headers = ListHeaders(None, "", False)
|
||
|
|
await session.execute(
|
||
|
|
update(Message)
|
||
|
|
.where(Message.id == message.id)
|
||
|
|
.values(
|
||
|
|
list_id=headers.list_id,
|
||
|
|
list_unsubscribe=headers.unsubscribe,
|
||
|
|
list_unsubscribe_post=headers.one_click,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
await session.commit()
|
||
|
|
done += len(batch)
|
||
|
|
if done:
|
||
|
|
logger.info("read list headers of %d message(s) for user %s", done, user_id)
|
||
|
|
return done
|