<commit_before>from collections import namedtuple
from itertools import imap

from flask import Blueprint, render_template, abort, request, url_for, redirect, g
from sqlalchemy.orm import joinedload, contains_eager

from skylines import db
from skylines.lib.dbutil import get_requested_record
from skylines.model import Event, Notification

EventGroup = namedtuple('EventGroup', ['grouped', 'type', 'time', 'actor_id', 'link', 'events'])


notifications_blueprint = Blueprint('notifications', 'skylines')


@notifications_blueprint.before_app_request
def inject_notification_count():
    if g.current_user:
        g.notifications = Notification.count_unread(g.current_user)


def _filter_query(query, args):
    if 'type' in args:
        query = query.filter(Event.type == args['type'])
    if 'user' in args:
        query = query.filter(Event.actor_id == args['user'])
    if 'flight' in args:
        query = query.filter(Event.flight_id == args['flight'])

    return query


def _group_events(_events):
    events = []
    for event in _events:
        # add first event if list is empty
        if not events:
            events.append(event)
            continue

        # get last event from list for comparison
        last_event = events[-1]

        # if there are multiple flight events from the same actor -> group them
        if (event.type == Event.Type.FLIGHT and
                last_event.type == event.type and
                last_event.actor_id == event.actor_id):
            if isinstance(last_event, EventGroup):
                last_event.events.append(event)
            else:
                events[-1] = EventGroup(
                    grouped=True, type=event.type, time=last_event.time,
                    actor_id=event.actor_id, events=[last_event, event],
                    link=url_for('.index', type=event.type, user=event.actor_id))
            continue

        events.append(event)

    return events


@notifications_blueprint.route('/')
def index():
    if not g.current_user:
        abort(401)

    query = Notification.query_unread(g.current_user) \
        .join('event') \
        .options(contains_eager('event')) \
        .options(joinedload('event.actor')) \<commit_msg>views/notifications: Converted EventGroup tuple to class<commit_after>from collections import namedtuple
from itertools import imap

from flask import Blueprint, render_template, abort, request, url_for, redirect, g
from sqlalchemy.orm import joinedload, contains_eager

from skylines import db
from skylines.lib.dbutil import get_requested_record
from skylines.model import Event, Notification


class EventGroup:
    grouped = True

    def __init__(self, events, link=None):
        self.events = events
        self.link = link

    def __getattr__(self, name):
        return getattr(self.events[0], name)


notifications_blueprint = Blueprint('notifications', 'skylines')


@notifications_blueprint.before_app_request
def inject_notification_count():
    if g.current_user:
        g.notifications = Notification.count_unread(g.current_user)


def _filter_query(query, args):
    if 'type' in args:
        query = query.filter(Event.type == args['type'])
    if 'user' in args:
        query = query.filter(Event.actor_id == args['user'])
    if 'flight' in args:
        query = query.filter(Event.flight_id == args['flight'])

    return query


def _group_events(_events):
    events = []
    for event in _events:
        # add first event if list is empty
        if not events:
            events.append(event)
            continue

        # get last event from list for comparison
        last_event = events[-1]

        # if there are multiple flight events from the same actor -> group them
        if (event.type == Event.Type.FLIGHT and
                last_event.type == event.type and
                last_event.actor_id == event.actor_id):
            if isinstance(last_event, EventGroup):
                last_event.events.append(event)
            else:
                events[-1] = EventGroup([last_event, event], link=url_for(
                    '.index', type=event.type, user=event.actor_id))
            continue

        events.append(event)

    return events


@notifications_blueprint.route('/')
def index():
    if not g.current_user:
        abort(401)

    query = Notification.query_unread(g.current_user) \
        .join('event') \
        .options(contains_eager('event')) \
        .options(joinedload('event.actor')) \
{'fraction_non_alphanumeric': 0.10049019607843138, 'fraction_numerical': 0.0024509803921568627, 'mean_word_length': 2.7222222222222223, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4205053', 'n_tokens_mistral': 1314, 'n_tokens_neox': 1261, 'n_words': 333}