Passed
Branch main (854eb5)
by Jochen
04:24
created

is_posting_unseen()   A

Complexity

Conditions 3

Size

Total Lines 13
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 5
CRAP Score 3.0416

Importance

Changes 0
Metric Value
cc 3
eloc 6
nop 2
dl 0
loc 13
ccs 5
cts 6
cp 0.8333
crap 3.0416
rs 10
c 0
b 0
f 0
1
"""
2
byceps.blueprints.site.board.service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2021 Jochen Kupperschmidt
6
:License: Revised BSD (see `LICENSE` file for details)
7
"""
8
9 1
from __future__ import annotations
10 1
from datetime import datetime
11 1
from typing import Optional, Sequence
12
13 1
from flask import g
14
15 1
from ....services.authentication.session.models.current_user import CurrentUser
16 1
from ....services.board.dbmodels.topic import Topic as DbTopic
17 1
from ....services.board.dbmodels.posting import Posting as DbPosting
18 1
from ....services.board import (
19
    last_view_service as board_last_view_service,
20
    posting_query_service as board_posting_query_service,
21
)
22 1
from ....services.board.transfer.models import CategoryWithLastUpdate
23 1
from ....services.party import service as party_service
24 1
from ....services.site import settings_service as site_settings_service
25 1
from ....services.ticketing import ticket_service
26 1
from ....services.user import service as user_service
27 1
from ....services.user_badge import awarding_service as badge_awarding_service
28 1
from ....services.user_badge.transfer.models import Badge
29 1
from ....util.authorization import has_current_user_permission
30 1
from ....typing import BrandID, PartyID, UserID
31
32 1
from .authorization import (
33
    BoardPermission,
34
    BoardPostingPermission,
35
    BoardTopicPermission,
36
)
37 1
from .models import CategoryWithLastUpdateAndUnseenFlag, Creator, Ticket
38
39
40 1
DEFAULT_POSTINGS_PER_PAGE = 10
41 1
DEFAULT_TOPICS_PER_PAGE = 10
42
43
44 1
def add_unseen_postings_flag_to_categories(
45
    categories: Sequence[CategoryWithLastUpdate], user: CurrentUser
46
) -> Sequence[CategoryWithLastUpdateAndUnseenFlag]:
47
    """Add flag to each category stating if it contains postings unseen
48
    by the user.
49
    """
50 1
    categories_with_flag = []
51
52 1
    for category in categories:
53 1
        contains_unseen_postings = (
54
            user.authenticated
55
            and board_last_view_service.contains_category_unseen_postings(
56
                category, user.id
57
            )
58
        )
59
60 1
        category_with_flag = (
61
            CategoryWithLastUpdateAndUnseenFlag.from_category_with_last_update(
62
                category, contains_unseen_postings
63
            )
64
        )
65
66 1
        categories_with_flag.append(category_with_flag)
67
68 1
    return categories_with_flag
69
70
71 1
def add_topic_creators(topics: Sequence[DbTopic]) -> None:
72
    """Add each topic's creator as topic attribute."""
73 1
    creator_ids = {t.creator_id for t in topics}
74 1
    creators = user_service.find_users(creator_ids, include_avatars=True)
75 1
    creators_by_id = user_service.index_users_by_id(creators)
76
77 1
    for topic in topics:
78 1
        topic.creator = creators_by_id[topic.creator_id]
79
80
81 1
def add_topic_unseen_flag(topics: Sequence[DbTopic], user: CurrentUser) -> None:
82
    """Add `unseen` flag to topics."""
83 1
    for topic in topics:
84 1
        topic.contains_unseen_postings = (
85
            user.authenticated
86
            and board_last_view_service.contains_topic_unseen_postings(
87
                topic, user.id
88
            )
89
        )
90
91
92 1
def add_unseen_flag_to_postings(
93
    postings: Sequence[DbPosting], last_viewed_at: datetime
94
) -> None:
95
    """Add the attribute 'unseen' to each post."""
96 1
    for posting in postings:
97 1
        posting.unseen = is_posting_unseen(posting, last_viewed_at)
98
99
100 1
def is_posting_unseen(posting: DbPosting, last_viewed_at: datetime) -> bool:
101
    """Return `True` if the posting has not yet been seen by the current
102
    user.
103
    """
104
    # Don't display any posting as new to a guest.
105 1
    if not g.user.authenticated:
106 1
        return False
107
108
    # Don't display the author's own posting as new to them.
109 1
    if posting.creator_id == g.user.id:
110
        return False
111
112 1
    return (last_viewed_at is None) or (posting.created_at > last_viewed_at)
113
114
115 1
def enrich_creators(
116
    postings: Sequence[DbPosting],
117
    brand_id: BrandID,
118
    party_id: Optional[PartyID],
119
) -> None:
120
    """Enrich creators with their badges."""
121 1
    creator_ids = {posting.creator_id for posting in postings}
122
123 1
    badges_by_user_id = _get_badges_for_users(creator_ids, brand_id)
124
125 1
    if party_id is not None:
126 1
        party = party_service.get_party(party_id)
127 1
        ticket_users = ticket_service.select_ticket_users_for_party(
128
            creator_ids, party.id
129
        )
130
    else:
131
        party = None
132
        ticket_users = set()
133
134 1
    for posting in postings:
135 1
        user_id = posting.creator_id
136
137 1
        badges: set[Badge] = badges_by_user_id.get(user_id, set())
138
139 1
        if user_id in ticket_users:
140
            ticket = Ticket(party.title)
141
        else:
142 1
            ticket = None
143
144 1
        posting.creator = Creator.from_(posting.creator, badges, ticket)
145
146
147 1
def _get_badges_for_users(
148
    user_ids: set[UserID], brand_id: BrandID
149
) -> dict[UserID, set[Badge]]:
150
    """Fetch users' badges that are either global or belong to the brand."""
151 1
    badges_by_user_id = badge_awarding_service.get_badges_awarded_to_users(
152
        user_ids, featured_only=True
153
    )
154
155 1
    def generate_items():
156 1
        for user_id, badges in badges_by_user_id.items():
157
            selected_badges = {
158
                badge for badge in badges if badge.brand_id in {None, brand_id}
159
            }
160
            yield user_id, selected_badges
161
162 1
    return dict(generate_items())
163
164
165 1
def calculate_posting_page_number(posting: DbPosting) -> int:
166
    """Calculate the number of postings to show per page."""
167 1
    include_hidden = may_current_user_view_hidden()
168 1
    postings_per_page = get_postings_per_page_value()
169
170 1
    return board_posting_query_service.calculate_posting_page_number(
171
        posting, include_hidden, postings_per_page
172
    )
173
174
175 1
def get_topics_per_page_value() -> int:
176
    """Return the configured number of topics per page."""
177 1
    return _get_site_setting_int_value(
178
        'board_topics_per_page', DEFAULT_TOPICS_PER_PAGE
179
    )
180
181
182 1
def get_postings_per_page_value() -> int:
183
    """Return the configured number of postings per page."""
184 1
    return _get_site_setting_int_value(
185
        'board_postings_per_page', DEFAULT_POSTINGS_PER_PAGE
186
    )
187
188
189 1
def _get_site_setting_int_value(key, default_value) -> int:
190 1
    value = site_settings_service.find_setting_value(g.site_id, key)
191
192 1
    if value is None:
193 1
        return default_value
194
195
    return int(value)
196
197
198 1
def may_current_user_view_hidden() -> bool:
199
    """Return `True' if the current user may view hidden items."""
200 1
    return has_current_user_permission(BoardPermission.view_hidden)
201
202
203 1
def may_topic_be_updated_by_current_user(topic: DbTopic) -> bool:
204
    """Return `True` if the topic may be updated by the current user."""
205 1
    return (
206
        not topic.locked
207
        and g.user.id == topic.creator_id
208
        and has_current_user_permission(BoardTopicPermission.update)
209
    ) or has_current_user_permission(BoardPermission.update_of_others)
210
211
212 1
def may_posting_be_updated_by_current_user(posting: DbPosting) -> bool:
213
    """Return `True` if the post may be updated by the current user."""
214
    return (
215
        not posting.topic.locked
216
        and g.user.id == posting.creator_id
217
        and has_current_user_permission(BoardPostingPermission.update)
218
    ) or has_current_user_permission(BoardPermission.update_of_others)
219