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
|
|
|
|