nordabiz/blueprints/unsubscribe/routes.py
Maciej Pienczyn dcbf8b5db6
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
feat(email): one-click unsubscribe w mailach powiadomień (RFC 8058)
Każdy e-mail powiadomieniowy ma teraz:
(1) link w stopce "Wyłącz ten typ powiadomień jednym kliknięciem"
(2) nagłówki List-Unsubscribe + List-Unsubscribe-Post dla klientów
    pocztowych (Gmail/Apple Mail pokażą natywny przycisk Unsubscribe)

Implementacja:
- utils/unsubscribe_tokens.py: signed token (itsdangerous, SECRET_KEY)
  niosący user_id + notification_type, bez wygasania
- blueprints/unsubscribe: GET /unsubscribe?t=TOKEN → strona potwierdzenia,
  POST /unsubscribe → faktyczne wyłączenie flagi notify_email_<type>
- email_service.send_email() dostał parametr notification_type. Jeśli
  przekazany razem z user_id, footer + headery są doklejane
- Aktualizowane wywołania: message_notification (messages),
  classified_question/answer (B2B Q&A), classified_expiry (skrypt cron)

Prefetch safety: GET pokazuje stronę z przyciskiem "Tak, wyłącz",
wyłączenie następuje po POST. RFC 8058 One-Click (POST bez formularza
z Content-Type application/x-www-form-urlencoded + body
"List-Unsubscribe=One-Click") obsługuje klientów pocztowych.

D.2/D.3 dorzucą kolejne notification_type (forum, broadcast, events).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-14 17:56:36 +02:00

92 lines
3.2 KiB
Python

"""Unsubscribe endpoints — bez logowania, weryfikacja przez signed token.
GET /unsubscribe?t=<token> — strona potwierdzenia (klik z maila)
POST /unsubscribe — faktyczne wyłączenie (formularz lub
List-Unsubscribe-Post z klienta pocztowego)
"""
import logging
from flask import request, render_template, jsonify, abort
from database import SessionLocal, User
from utils.unsubscribe_tokens import (
verify_unsubscribe_token,
column_for_type,
label_for_type,
)
from . import bp
logger = logging.getLogger(__name__)
def _apply_unsubscribe(token: str):
"""Apply the unsubscribe. Returns (ok: bool, label: str | None, user_email: str | None)."""
parsed = verify_unsubscribe_token(token)
if not parsed:
return False, None, None
user_id, ntype = parsed
col = column_for_type(ntype)
if not col:
return False, None, None
db = SessionLocal()
try:
user = db.query(User).filter(User.id == user_id).first()
if not user or not hasattr(user, col):
return False, None, None
setattr(user, col, False)
db.commit()
logger.info("unsubscribe applied user=%s type=%s", user_id, ntype)
return True, label_for_type(ntype), user.email
except Exception as e:
db.rollback()
logger.error("unsubscribe error: %s", e)
return False, None, None
finally:
db.close()
@bp.route('/unsubscribe', methods=['GET'])
def unsubscribe_view():
"""Strona potwierdzenia (lub od razu wyłączenie dla List-Unsubscribe-Post)."""
token = request.args.get('t', '').strip()
parsed = verify_unsubscribe_token(token) if token else None
if not parsed:
return render_template('unsubscribe.html',
status='invalid',
label=None,
token=None), 400
user_id, ntype = parsed
return render_template('unsubscribe.html',
status='confirm',
label=label_for_type(ntype),
token=token)
@bp.route('/unsubscribe', methods=['POST'])
def unsubscribe_apply():
"""Faktyczne wyłączenie — formularz z GET lub RFC 8058 One-Click."""
token = (request.form.get('t') or request.args.get('t') or '').strip()
# RFC 8058 One-Click: body = "List-Unsubscribe=One-Click"
if not token:
body = request.get_data(as_text=True) or ''
if 'List-Unsubscribe=One-Click' in body:
token = request.args.get('t', '').strip()
ok, label, _ = _apply_unsubscribe(token)
# RFC 8058 One-Click expects plain 200
if request.headers.get('Content-Type', '').startswith('application/x-www-form-urlencoded') \
and 'List-Unsubscribe=One-Click' in (request.get_data(as_text=True) or ''):
return ('OK', 200) if ok else ('Invalid token', 400)
return render_template('unsubscribe.html',
status='done' if ok else 'invalid',
label=label,
token=None), 200 if ok else 400
def exempt_from_csrf(app):
csrf = app.extensions.get('csrf')
if csrf:
csrf.exempt(bp)