"""
Notification Helpers
====================
Functions for creating and managing user notifications.
"""
import logging
from database import SessionLocal, UserNotification, User
logger = logging.getLogger(__name__)
def create_notification(user_id, title, message, notification_type='info',
related_type=None, related_id=None, action_url=None):
"""
Create a notification for a user.
Args:
user_id: ID of the user to notify
title: Notification title
message: Notification message/body
notification_type: Type of notification (news, system, message, event, alert)
related_type: Type of related entity (company_news, event, message, etc.)
related_id: ID of the related entity
action_url: URL to navigate when notification is clicked
Returns:
UserNotification object or None on error
"""
db = SessionLocal()
try:
notification = UserNotification(
user_id=user_id,
title=title,
message=message,
notification_type=notification_type,
related_type=related_type,
related_id=related_id,
action_url=action_url
)
db.add(notification)
db.commit()
db.refresh(notification)
logger.info(f"Created notification for user {user_id}: {title}")
return notification
except Exception as e:
logger.error(f"Error creating notification: {e}")
db.rollback()
return None
finally:
db.close()
def create_news_notification(company_id, news_id, news_title):
"""
Create notification for company owner when their news is approved.
Args:
company_id: ID of the company
news_id: ID of the approved news
news_title: Title of the news
"""
db = SessionLocal()
try:
# Find users associated with this company
users = db.query(User).filter(
User.company_id == company_id,
User.is_active == True
).all()
for user in users:
create_notification(
user_id=user.id,
title="Nowa aktualnosc o Twojej firmie",
message=f"Aktualnosc '{news_title}' zostala zatwierdzona i jest widoczna na profilu firmy.",
notification_type='news',
related_type='company_news',
related_id=news_id,
action_url=f"/company/{company_id}"
)
finally:
db.close()
def create_message_notification(user_id, sender_name, message_id):
"""
Create notification when user receives a private message.
Args:
user_id: ID of the recipient
sender_name: Name of the sender
message_id: ID of the message
"""
create_notification(
user_id=user_id,
title="Nowa wiadomość prywatna",
message=f"Otrzymałeś nową wiadomość od {sender_name}.",
notification_type='message',
related_type='private_message',
related_id=message_id,
action_url=f"/wiadomosci/{message_id}"
)
def create_event_notification(user_id, event_title, event_id):
"""
Create notification for upcoming event reminder.
Args:
user_id: ID of the user to notify
event_title: Title of the event
event_id: ID of the event
"""
create_notification(
user_id=user_id,
title="Przypomnienie o wydarzeniu",
message=f"Zbliża się wydarzenie: {event_title}",
notification_type='event',
related_type='norda_event',
related_id=event_id,
action_url=f"/kalendarz/{event_id}"
)
def notify_all_users_release(version, highlights=None):
"""
Notify all active users about a new system release.
Args:
version: Version string (e.g., 'v1.17.0')
highlights: Optional list of key changes to include in notification
Returns:
Number of notifications created
"""
db = SessionLocal()
try:
# Get all active users
users = db.query(User).filter(User.is_active == True).all()
message = f"Wersja {version} jest już dostępna."
if highlights:
# Include first 2 highlights
message += " Nowości: " + ", ".join(highlights[:2])
if len(highlights) > 2:
message += f" (+{len(highlights) - 2} więcej)"
count = 0
for user in users:
result = create_notification(
user_id=user.id,
title=f"🚀 Nowa wersja systemu {version}",
message=message,
notification_type='system',
related_type='release',
related_id=None,
action_url='/release-notes'
)
if result:
count += 1
logger.info(f"Created {count} release notifications for version {version}")
return count
except Exception as e:
logger.error(f"Error creating release notifications: {e}")
return 0
finally:
db.close()
def notify_all_users_announcement(announcement_id, title, category=None):
"""
Notify all active users about a new announcement.
Args:
announcement_id: ID of the announcement
title: Title of the announcement
category: Optional category for context
Returns:
Number of notifications created
"""
db = SessionLocal()
try:
# Get all active users
users = db.query(User).filter(User.is_active == True).all()
# Category-specific icons
category_icons = {
'general': '📢',
'event': '📅',
'business_opportunity': '💼',
'member_news': '👥',
'partnership': '🤝'
}
icon = category_icons.get(category, '📢')
count = 0
for user in users:
result = create_notification(
user_id=user.id,
title=f"{icon} Nowe ogłoszenie w Aktualnościach",
message=title[:100] + ('...' if len(title) > 100 else ''),
notification_type='news',
related_type='announcement',
related_id=announcement_id,
action_url=f'/ogloszenia/{announcement_id}'
)
if result:
count += 1
logger.info(f"Created {count} announcement notifications for: {title[:50]}")
return count
except Exception as e:
logger.error(f"Error creating announcement notifications: {e}")
return 0
finally:
db.close()
# ============================================================
# FORUM NOTIFICATIONS
# ============================================================
def create_forum_reply_notification(topic_id, topic_title, replier_name, reply_id, subscriber_ids):
"""
Notify topic subscribers about a new reply.
Args:
topic_id: ID of the forum topic
topic_title: Title of the topic
replier_name: Name of the user who replied
reply_id: ID of the new reply
subscriber_ids: List of user IDs to notify
Returns:
Number of notifications created
"""
count = 0
for user_id in subscriber_ids:
result = create_notification(
user_id=user_id,
title="Nowa odpowiedź na forum",
message=f"{replier_name} odpowiedział w temacie: {topic_title[:50]}{'...' if len(topic_title) > 50 else ''}",
notification_type='message',
related_type='forum_reply',
related_id=reply_id,
action_url=f'/forum/{topic_id}#reply-{reply_id}'
)
if result:
count += 1
logger.info(f"Created {count} forum reply notifications for topic {topic_id}")
return count
def send_forum_reply_email(topic_id, topic_title, replier_name, reply_content, subscriber_emails):
"""
Send email notifications to forum topic subscribers about a new reply.
Args:
topic_id: ID of the forum topic
topic_title: Title of the topic
replier_name: Name of the user who replied
reply_content: First 200 chars of the reply content
subscriber_emails: List of dicts with 'email' and 'name' keys
"""
from email_service import send_email
base_url = "https://nordabiznes.pl"
topic_url = f"{base_url}/forum/{topic_id}"
unsubscribe_url = f"{base_url}/forum/{topic_id}/unsubscribe"
preview = reply_content[:200].strip()
if len(reply_content) > 200:
preview += "..."
count = 0
for subscriber in subscriber_emails:
subject = f"Nowa odpowiedz na forum: {topic_title[:60]}"
body_text = f"""{replier_name} odpowiedzial w temacie: {topic_title}
"{preview}"
Zobacz pelna odpowiedz: {topic_url}
---
Aby przestac obserwowac ten watek: {unsubscribe_url}
Norda Biznes Partner - https://nordabiznes.pl
"""
from email_service import _email_v3_wrap
content = f'''
Cześć {subscriber.get('name', '')}!
{replier_name} odpowiedział w temacie, który obserwujesz:
'''
body_html = _email_v3_wrap('Nowa odpowiedź na forum', 'Norda Biznes Partner', content)
try:
result = send_email(
to=[subscriber['email']],
subject=subject,
body_text=body_text,
body_html=body_html,
email_type='forum_notification',
recipient_name=subscriber.get('name', '')
)
if result:
count += 1
except Exception as e:
logger.error(f"Failed to send forum reply email to {subscriber['email']}: {e}")
logger.info(f"Sent {count}/{len(subscriber_emails)} forum reply emails for topic {topic_id}")
return count
def create_forum_reaction_notification(user_id, reactor_name, content_type, content_id, topic_id, emoji):
"""
Notify user when someone reacts to their content.
Args:
user_id: ID of the content author
reactor_name: Name of user who reacted
content_type: 'topic' or 'reply'
content_id: ID of the content
topic_id: ID of the topic
emoji: The reaction emoji
"""
create_notification(
user_id=user_id,
title=f"Nowa reakcja {emoji}",
message=f"{reactor_name} zareagował na Twój{'ą odpowiedź' if content_type == 'reply' else ' temat'}",
notification_type='message',
related_type=f'forum_{content_type}',
related_id=content_id,
action_url=f'/forum/{topic_id}{"#reply-" + str(content_id) if content_type == "reply" else ""}'
)
def create_forum_solution_notification(user_id, topic_id, topic_title):
"""
Notify topic author when their question gets a solution.
Args:
user_id: ID of the topic author
topic_id: ID of the topic
topic_title: Title of the topic
"""
create_notification(
user_id=user_id,
title="Twoje pytanie ma rozwiązanie!",
message=f"Odpowiedź w temacie '{topic_title[:40]}' została oznaczona jako rozwiązanie.",
notification_type='message',
related_type='forum_topic',
related_id=topic_id,
action_url=f'/forum/{topic_id}'
)
def create_forum_report_notification(admin_user_ids, report_id, content_type, reporter_name):
"""
Notify admins about a new forum report.
Args:
admin_user_ids: List of admin user IDs
report_id: ID of the report
content_type: 'topic' or 'reply'
reporter_name: Name of the reporter
"""
for user_id in admin_user_ids:
create_notification(
user_id=user_id,
title="Nowe zgłoszenie na forum",
message=f"{reporter_name} zgłosił {'odpowiedź' if content_type == 'reply' else 'temat'}",
notification_type='alert',
related_type='forum_report',
related_id=report_id,
action_url='/admin/forum/reports'
)
def parse_mentions_and_notify(content, author_id, author_name, topic_id, content_type, content_id):
"""
Parse @mentions in content and send notifications.
Supports formats:
- @jan.kowalski (name with dots)
- @jan_kowalski (name with underscores)
- @jankowalski (name without separators)
Args:
content: Text content to parse
author_id: ID of the content author (won't be notified)
author_name: Name of the author
topic_id: ID of the topic
content_type: 'topic' or 'reply'
content_id: ID of the content
Returns:
List of mentioned user IDs
"""
import re
# Find all @mentions (letters, numbers, dots, underscores, hyphens)
mentions = re.findall(r'@([\w.\-]+)', content)
if not mentions:
return []
db = SessionLocal()
try:
mentioned_user_ids = []
for mention in set(mentions): # Unique mentions
# Try to find user by name (case-insensitive)
mention_lower = mention.lower()
# Try exact name match
user = db.query(User).filter(
User.is_active == True,
User.id != author_id
).filter(
(User.name.ilike(mention)) |
(User.name.ilike(mention.replace('.', ' '))) |
(User.name.ilike(mention.replace('_', ' '))) |
(User.email.ilike(f'{mention}@%'))
).first()
if user:
mentioned_user_ids.append(user.id)
create_notification(
user_id=user.id,
title=f"@{author_name} wspomniał o Tobie",
message=f"Zostałeś wspomniany w {'odpowiedzi' if content_type == 'reply' else 'temacie'} na forum",
notification_type='message',
related_type=f'forum_{content_type}',
related_id=content_id,
action_url=f'/forum/{topic_id}{"#reply-" + str(content_id) if content_type == "reply" else ""}'
)
return mentioned_user_ids
except Exception as e:
logger.error(f"Error parsing mentions: {e}")
return []
finally:
db.close()