hnzio/putzliga
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: improve scheduled push notifications

Browse Source
This commit is contained in:
Florian Heinz
2026-04-15 14:28:31 +02:00
parent dba87ebcf2
commit e4589df111
11 changed files with 195 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
app/services/notifications.py
View File

@@ -2,14 +2,14 @@ from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import timedelta
from urllib.parse import urljoin
from flask import current_app
from pywebpush import WebPushException, webpush
from sqlalchemy import or_
from ..extensions import db
from ..models import NotificationLog, PushSubscription, TaskInstance, User
from ..models import NotificationLog, PushSubscription, TaskInstance, User, UserBadge
from .monthly import archive_months_missing_up_to_previous, get_snapshot_rows
from .dates import local_now, previous_month
@@ -60,56 +60,96 @@ def _send_subscription(subscription: PushSubscription, payload: dict) -> bool:
return False
def _subscriptions_for_user(user_id: int) -> list[PushSubscription]:
return PushSubscription.query.filter_by(user_id=user_id).all()
def _send_payload_to_user(user: User, notification_type: str, marker: dict, payload: dict) -> NotificationResult:
result = NotificationResult()
subscriptions = _subscriptions_for_user(user.id)
if not subscriptions:
result.skipped += 1
return result
if _notification_exists(user.id, notification_type, marker):
result.skipped += 1
return result
sent_any = False
for subscription in subscriptions:
if _send_subscription(subscription, payload):
sent_any = True
result.sent += 1
else:
result.failed += 1
if sent_any:
_log_notification(user.id, notification_type, marker)
db.session.commit()
elif result.failed == 0:
result.skipped += 1
return result
def _merge_results(base: NotificationResult, extra: NotificationResult) -> NotificationResult:
base.sent += extra.sent
base.skipped += extra.skipped
base.failed += extra.failed
return base
def send_due_notifications() -> NotificationResult:
result = NotificationResult()
if not push_enabled():
result.skipped += 1
return result
today = local_now().date()
now = local_now()
if now.hour < 9:
result.skipped += 1
return result
today = now.date()
relevant_tasks = TaskInstance.query.filter(
TaskInstance.completed_at.is_(None),
TaskInstance.assigned_user_id.isnot(None),
TaskInstance.due_date <= today + timedelta(days=1),
TaskInstance.due_date == today,
or_(
TaskInstance.assigned_user_id.isnot(None),
TaskInstance.assigned_user_secondary_id.isnot(None),
),
).all()
tasks_by_user: dict[int, list[TaskInstance]] = {}
for task in relevant_tasks:
user = task.assigned_user
if not user or not user.notification_task_due_enabled:
for assigned_user in task.assigned_users:
tasks_by_user.setdefault(assigned_user.id, []).append(task)
for user in User.query.order_by(User.name.asc()).all():
if not user.notification_task_due_enabled:
result.skipped += 1
continue
payload_marker = {"task_instance_id": task.id, "due_date": task.due_date.isoformat()}
if _notification_exists(user.id, "task_due", payload_marker):
personal_tasks = tasks_by_user.get(user.id, [])
if not personal_tasks:
result.skipped += 1
continue
subscriptions = PushSubscription.query.filter_by(user_id=user.id).all()
if not subscriptions:
result.skipped += 1
continue
body = "Heute ist ein guter Tag für Punkte." if task.due_date <= today else "Morgen wird's fällig."
task_count = len(personal_tasks)
payload = {
"title": f"Putzliga erinnert: {task.title}",
"body": body,
"title": "Putzliga für heute",
"body": (
"Heute wartet 1 offene Aufgabe auf dich. Zeit zum Punkte sammeln."
if task_count == 1
else f"Heute warten {task_count} offene Aufgaben auf dich. Zeit zum Punkte sammeln."
),
"icon": _absolute_url("/static/images/pwa-icon-192.png"),
"badge": _absolute_url("/static/images/pwa-badge.png"),
"url": _absolute_url("/my-tasks"),
"tag": f"task-{task.id}",
"tag": f"due-{today.isoformat()}-{user.id}",
}
sent_any = False
for subscription in subscriptions:
if _send_subscription(subscription, payload):
sent_any = True
result.sent += 1
else:
result.failed += 1
if sent_any:
_log_notification(user.id, "task_due", payload_marker)
db.session.commit()
marker = {"date": today.isoformat()}
_merge_results(result, _send_payload_to_user(user, "task_due_daily", marker, payload))
return result
@@ -140,14 +180,6 @@ def send_monthly_winner_notifications() -> NotificationResult:
if not user.notification_monthly_winner_enabled:
result.skipped += 1
continue
if _notification_exists(user.id, "monthly_winner", marker):
result.skipped += 1
continue
subscriptions = PushSubscription.query.filter_by(user_id=user.id).all()
if not subscriptions:
result.skipped += 1
continue
payload = {
"title": "Der Haushalts-Champion des letzten Monats steht fest",
@@ -157,16 +189,43 @@ def send_monthly_winner_notifications() -> NotificationResult:
"url": _absolute_url(f"/scoreboard?archive={target_year}-{target_month:02d}"),
"tag": f"winner-{target_year}-{target_month}",
}
sent_any = False
for subscription in subscriptions:
if _send_subscription(subscription, payload):
sent_any = True
result.sent += 1
else:
result.failed += 1
if sent_any:
_log_notification(user.id, "monthly_winner", marker)
db.session.commit()
_merge_results(result, _send_payload_to_user(user, "monthly_winner", marker, payload))
return result
def send_badge_notifications_for_awards(awards: list[UserBadge]) -> NotificationResult:
result = NotificationResult()
if not push_enabled():
result.skipped += len(awards) or 1
return result
for award in awards:
user = award.user
definition = award.badge_definition
if not user or not definition:
result.skipped += 1
continue
if not user.notification_badge_enabled:
result.skipped += 1
continue
marker = {"user_badge_id": award.id}
payload = {
"title": "Neues Badge freigeschaltet",
"body": f"{definition.name}: {definition.description}",
"icon": _absolute_url("/static/images/pwa-icon-192.png"),
"badge": _absolute_url("/static/images/pwa-badge.png"),
"url": _absolute_url("/settings"),
"tag": f"badge-{award.id}",
}
_merge_results(result, _send_payload_to_user(user, "badge_award", marker, payload))
return result
def run_scheduled_notifications() -> dict[str, NotificationResult]:
return {
"daily_due": send_due_notifications(),
"monthly_winner": send_monthly_winner_notifications(),
}