email-unsubscribe-links-scraper/app.py

"""
E-Mail Unsubscribe Link Finder.

Reads a collection mailbox (forwarded newsletters), lists the unsubscribe links per mail and moves
handled mails to the trash. Runs behind the reverse proxy (Caddy forward_auth / Authelia); the
mailbox credentials come from the environment.
"""

import email
import imaplib
import logging
import os
import re
from email.header import decode_header, make_header

from bs4 import BeautifulSoup
from dotenv import load_dotenv
from flask import Flask, redirect, render_template, request, url_for

load_dotenv()  # optional .env for local runs

EMAIL_USERNAME = os.environ.get("EMAIL_USERNAME", "")
EMAIL_PASSWORD = os.environ.get("EMAIL_PASSWORD", "")
EMAIL_SERVER = os.environ.get("EMAIL_SERVER", "")
EMAIL_PORT = int(os.environ.get("EMAIL_PORT", "993"))
EMAIL_FOLDER = os.environ.get("EMAIL_FOLDER", "INBOX")
TRASH_FOLDER = os.environ.get("TRASH_FOLDER", "Trash")

logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
logger = logging.getLogger(__name__)

template_dir = os.path.abspath(os.path.dirname(__file__))
app = Flask(__name__, template_folder=template_dir)

UNSUBSCRIBE_PATTERN = re.compile(
    r"\babbestellen\b|\bunsubscribe\b|\bdeabonnieren\b|\babbestellung\b|\babmelden\b", re.IGNORECASE
)
URL_PATTERN = re.compile(r"https?://[^\s<>\"']+")
UID_PATTERN = re.compile(r"^\d+$")


def is_safe_link(url: str) -> bool:
    """Only plain web links: mails are untrusted, javascript:/data: links must never become clickable."""
    return url.lower().startswith(("https://", "http://"))


def get_text_from_email(message) -> str:
    text = ""
    for part in message.walk():
        disposition = str(part.get("Content-Disposition"))
        if "text" in part.get_content_type() and "attachment" not in disposition:
            body = part.get_payload(decode=True)
            if body:
                text += body.decode(errors="ignore")
    return text


def get_subject(message) -> str:
    raw = message.get("Subject", "")
    try:
        return str(make_header(decode_header(raw)))
    except Exception:
        return raw


def find_unsubscribe_links(text: str) -> list[str]:
    """Unsubscribe links from HTML anchors (link text or URL matches) and from plain-text URLs."""
    links: list[str] = []
    soup = BeautifulSoup(text, "html.parser")
    for anchor in soup.find_all("a", href=True):
        href = str(anchor["href"]).strip()
        if UNSUBSCRIBE_PATTERN.search(str(anchor)) and is_safe_link(href) and href not in links:
            links.append(href)
    for url in URL_PATTERN.findall(text):
        if UNSUBSCRIBE_PATTERN.search(url) and url not in links:
            links.append(url)
    return links


def connect() -> imaplib.IMAP4_SSL:
    mail = imaplib.IMAP4_SSL(EMAIL_SERVER, EMAIL_PORT)
    mail.login(EMAIL_USERNAME, EMAIL_PASSWORD)
    mail.select(EMAIL_FOLDER)
    return mail


def collect_links(mail) -> dict:
    """{uid: {"subject": ..., "links": [...]}} for all mails with unsubscribe links (UIDs stay stable)."""
    _, data = mail.uid("search", None, "ALL")
    result = {}
    for uid in data[0].split():
        _, msg_data = mail.uid("fetch", uid, "(BODY.PEEK[])")
        if not msg_data or not isinstance(msg_data[0], tuple):
            continue
        message = email.message_from_bytes(msg_data[0][1])
        body = get_text_from_email(message)
        links = find_unsubscribe_links(body) if body else []
        if links:
            result[uid.decode()] = {"subject": get_subject(message), "links": links}
    return result


def move_emails_to_trash(uids: list[str]) -> int:
    uids = [u for u in uids if UID_PATTERN.match(u)]
    if not uids:
        return 0
    mail = connect()
    try:
        for uid in uids:
            # Mailcow/Dovecot: copy to trash, flag deleted, expunge
            mail.uid("copy", uid, TRASH_FOLDER)
            mail.uid("store", uid, "+FLAGS", "(\\Deleted)")
        mail.expunge()
    finally:
        mail.logout()
    logger.info("Moved %d mail(s) to %s", len(uids), TRASH_FOLDER)
    return len(uids)


@app.route("/", methods=["GET"])
def index():
    return render_template("index.html", email_address=EMAIL_USERNAME)


@app.route("/result")
def result():
    mail = connect()
    try:
        links_data = collect_links(mail)
    finally:
        mail.logout()
    return render_template("result.html", links_data=links_data)


@app.route("/move-to-trash", methods=["POST"])
def move_to_trash():
    email_id = request.form.get("email_id", "")
    move_emails_to_trash([email_id])
    return redirect(url_for("result"))


@app.route("/move-all-to-trash", methods=["POST"])
def move_all_to_trash():
    move_emails_to_trash(request.form.getlist("email_id"))
    return redirect(url_for("index"))


@app.route("/health")
def health():
    return {"status": "ok"}


if __name__ == "__main__":
    # local development only; the container runs gunicorn (no debugger)
    app.run(host="127.0.0.1", port=5000, debug=os.environ.get("FLASK_DEBUG") == "1")