nordabiz/utils/unsubscribe_tokens.py
Maciej Pienczyn c9985ba51a
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(notifications): D.2+D.3 — forum, broadcasty Izby, wydarzenia, cron 24h
Rozszerzenie powiadomień o kolejne typy zdarzeń, z symetrycznymi togglami
e-mail i push w /konto/prywatnosc.

Migracje 103 + 104 — 6 nowych kolumn preferencji e-mail + NordaEvent.reminder_24h_sent_at.

Triggery:
- Forum odpowiedź → push do autora wątku (notify_push_forum_reply)
- Forum cytat (> **Imię** napisał(a):) → push + email do cytowanego
  (notify_push/email_forum_quote)
- Admin publikuje aktualność → broadcast push (ON) + email (OFF)
  do aktywnych członków (notify_push/email_announcements)
- Board: utworzenie / publikacja programu / publikacja protokołu
  → broadcast push + opt-in email (notify_push/email_board_meetings)
- Nowe wydarzenie w kalendarzu → broadcast push + email (oba ON)
  (notify_push/email_event_invites)
- Cron scripts/event_reminders_cron.py co godzinę — wydarzenia za 23-25h,
  dla zapisanych (EventAttendee.status != 'declined') push + email,
  znacznik NordaEvent.reminder_24h_sent_at żeby nie dublować.

Email defaults dobrane, by nie zalać inbox: broadcast OFF (announcements,
board, forum_reply), personalne/actionable ON (forum_quote, event_invites,
event_reminders).

Wszystkie nowe e-maile mają jednym-kliknięciem unsubscribe (RFC 8058
+ link w stopce) — unsubscribe_tokens.py rozszerzony o nowe typy.

Cron entry do dodania na prod (osobny krok, bo to edycja crontaba):
  0 * * * * cd /var/www/nordabiznes && venv/bin/python3 scripts/event_reminders_cron.py

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

71 lines
2.6 KiB
Python

"""Signed tokens do one-click unsubscribe z maili powiadomień.
Token niesie user_id + notification_type + podpis HMAC (przez itsdangerous).
Bez wygasania — user może kliknąć stary mail po tygodniach.
URL: /unsubscribe?t=<token>
"""
import os
from itsdangerous import URLSafeSerializer, BadSignature
# Mapa: notification_type -> nazwa kolumny User
NOTIFICATION_TYPE_TO_COLUMN = {
'messages': 'notify_email_messages',
'classified_question': 'notify_email_classified_question',
'classified_answer': 'notify_email_classified_answer',
'classified_expiry': 'notify_email_classified_expiry',
'forum_reply': 'notify_email_forum_reply',
'forum_quote': 'notify_email_forum_quote',
'announcements': 'notify_email_announcements',
'board_meetings': 'notify_email_board_meetings',
'event_invites': 'notify_email_event_invites',
'event_reminders': 'notify_email_event_reminders',
}
# Friendly labels dla strony potwierdzenia
NOTIFICATION_TYPE_LABELS = {
'messages': 'Nowe wiadomości prywatne',
'classified_question': 'Pytanie do Twojego ogłoszenia B2B',
'classified_answer': 'Odpowiedź na Twoje pytanie B2B',
'classified_expiry': 'Przypomnienie o wygasającym ogłoszeniu B2B',
'forum_reply': 'Odpowiedź w Twoim wątku forum',
'forum_quote': 'Cytat Twojego wpisu na forum',
'announcements': 'Aktualności Izby',
'board_meetings': 'Posiedzenia Rady Izby',
'event_invites': 'Nowe wydarzenia w kalendarzu Izby',
'event_reminders': 'Przypomnienia 24h przed wydarzeniem',
}
def _serializer():
secret = os.getenv('SECRET_KEY') or os.getenv('FLASK_SECRET_KEY') or 'nordabiz-dev-fallback'
return URLSafeSerializer(secret, salt='unsub-v1')
def generate_unsubscribe_token(user_id: int, notification_type: str) -> str:
if notification_type not in NOTIFICATION_TYPE_TO_COLUMN:
raise ValueError(f"Unknown notification_type: {notification_type}")
return _serializer().dumps({'u': user_id, 't': notification_type})
def verify_unsubscribe_token(token: str):
"""Return (user_id: int, notification_type: str) or None if invalid."""
try:
payload = _serializer().loads(token)
uid = int(payload.get('u', 0))
ntype = payload.get('t', '')
if not uid or ntype not in NOTIFICATION_TYPE_TO_COLUMN:
return None
return uid, ntype
except (BadSignature, ValueError, TypeError):
return None
def column_for_type(notification_type: str):
return NOTIFICATION_TYPE_TO_COLUMN.get(notification_type)
def label_for_type(notification_type: str):
return NOTIFICATION_TYPE_LABELS.get(notification_type, notification_type)