Coverage for app/backend/src/couchers/tasks.py: 94%
76 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import logging
2from collections.abc import Sequence
4from sqlalchemy import RowMapping, insert, select
5from sqlalchemy.orm import Session
6from sqlalchemy.sql import func
8from couchers import urls
9from couchers.config import config
10from couchers.constants import SIGNUP_EMAIL_TOKEN_VALIDITY
11from couchers.context import CouchersContext
12from couchers.crypto import urlsafe_secure_token
13from couchers.db import session_scope
14from couchers.email.blocks import EmailBase
15from couchers.email.emails import EmailChangeConfirmationEmail, SignupContinueEmail, SignupVerifyEmail
16from couchers.email.queuing import queue_system_email, queue_userless_email
17from couchers.models import (
18 Cluster,
19 ClusterRole,
20 ClusterSubscription,
21 ContentReport,
22 ContributorForm,
23 EventCommunityInviteRequest,
24 Node,
25 RateLimitAction,
26 RateLimitViolation,
27 Reference,
28 SignupFlow,
29 StrongVerificationAttempt,
30 User,
31)
32from couchers.rate_limits.definitions import RATE_LIMIT_HOURS
33from couchers.utils import now
35logger = logging.getLogger(__name__)
38def send_signup_email(context: CouchersContext, session: Session, flow: SignupFlow) -> None:
39 logger.info(f"Sending signup email to {flow.email=}:")
41 # whether we've sent an email at all yet
42 email_sent_before = flow.email_sent
43 if flow.email_verified:
44 # we just send a link to continue, not a verification link
45 signup_link = urls.signup_link(token=flow.flow_token)
46 elif flow.email_token and flow.token_is_valid:
47 # if the verification email was sent and still is not expired, just resend the verification email
48 signup_link = urls.signup_link(token=flow.email_token)
49 else:
50 # otherwise send a fresh email with a new token
51 token = urlsafe_secure_token()
52 flow.email_verified = False
53 flow.email_token = token
54 flow.email_token_expiry = now() + SIGNUP_EMAIL_TOKEN_VALIDITY
55 signup_link = urls.signup_link(token=flow.email_token)
57 flow.email_sent = True
59 email: EmailBase
60 if email_sent_before:
61 email = SignupContinueEmail(user_name=flow.name, continue_url=signup_link)
62 else:
63 email = SignupVerifyEmail(user_name=flow.name, verify_url=signup_link)
65 queue_userless_email(
66 context,
67 session,
68 flow.email,
69 email,
70 source_data_header=f"signup; initial={not email_sent_before}",
71 )
74def send_email_changed_confirmation_to_new_email(context: CouchersContext, session: Session, user: User) -> None:
75 """
76 Send an email to the user's new email address requesting confirmation of email change
77 """
78 logger.info(
79 f"Sending email changed (confirmation) email to {user=}'s new email address, "
80 f"(old email: {user.email}, new email: {user.new_email=})"
81 )
83 if not user.new_email_token: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 raise ValueError(f"No new email token for {user.id}")
85 elif not user.new_email: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true
86 raise ValueError(f"No new email for {user.id}")
88 email = EmailChangeConfirmationEmail(
89 user_name=user.name,
90 old_email=user.email,
91 confirm_url=urls.change_email_link(confirmation_token=user.new_email_token),
92 )
94 queue_userless_email(context, session, user.new_email, email, source_data_header="email_changed_confirmation")
97def send_content_report_email(session: Session, content_report: ContentReport) -> None:
98 logger.info("Sending content report email")
99 queue_system_email(
100 session,
101 config.REPORTS_EMAIL_RECIPIENT,
102 "content_report",
103 template_args={"report": content_report},
104 )
107def maybe_send_reference_report_email(session: Session, reference: Reference) -> None:
108 if reference.should_report:
109 logger.info("Sending reference report email")
110 queue_system_email(
111 session,
112 config.REPORTS_EMAIL_RECIPIENT,
113 "reference_report",
114 template_args={"reference": reference},
115 )
118def send_rate_limit_violation_report_email(
119 session: Session,
120 rate_limit_violation: RateLimitViolation,
121 events: dict[RateLimitAction, Sequence[RowMapping]],
122 threshold: int,
123) -> None:
124 """Send a report email to the moderation team if a user exceeds a rate limit within a given time frame."""
125 logger.info(
126 f"Sending rate limit moderation email for user '{rate_limit_violation.user_id}' ({rate_limit_violation.action})"
127 )
128 user = session.get_one(User, rate_limit_violation.user_id)
129 queue_system_email(
130 session,
131 config.REPORTS_EMAIL_RECIPIENT,
132 "rate_limit_violation_report",
133 template_args={
134 "user": user,
135 "action": rate_limit_violation.action,
136 "threshold": threshold,
137 "hours": RATE_LIMIT_HOURS,
138 "is_hard_limit": rate_limit_violation.is_hard_limit,
139 "events": events,
140 },
141 )
144def send_duplicate_strong_verification_email(
145 session: Session, old_attempt: StrongVerificationAttempt, new_attempt: StrongVerificationAttempt
146) -> None:
147 logger.info("Sending duplicate SV email")
148 queue_system_email(
149 session,
150 config.REPORTS_EMAIL_RECIPIENT,
151 "duplicate_strong_verification_report",
152 template_args={
153 "new_user": new_attempt.user,
154 "new_attempt_id": new_attempt.id,
155 "old_user": old_attempt.user,
156 "old_attempt_id": old_attempt.id,
157 },
158 )
161def maybe_send_contributor_form_email(session: Session, form: ContributorForm) -> None:
162 if form.should_notify: 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 queue_system_email(
164 session,
165 config.CONTRIBUTOR_FORM_EMAIL_RECIPIENT,
166 "contributor_form",
167 template_args={"form": form},
168 )
171def send_event_community_invite_request_email(session: Session, request: EventCommunityInviteRequest) -> None:
172 queue_system_email(
173 session,
174 config.MODS_EMAIL_RECIPIENT,
175 "event_community_invite_request",
176 template_args={
177 "event_link": urls.event_link(occurrence_id=request.occurrence.id, slug=request.occurrence.event.slug),
178 "user_link": urls.user_link(username=request.user.username),
179 "view_link": urls.console_link(page="tools/community-invites"),
180 },
181 )
184def enforce_community_memberships() -> None:
185 """
186 Go through all communities and make sure every user in the polygon is also a member
187 """
188 with session_scope() as session:
189 for node in session.execute(select(Node)).scalars().all():
190 existing_users = select(ClusterSubscription.user_id).where(
191 ClusterSubscription.cluster == node.official_cluster
192 )
193 node_geom = select(Node.geom).where(Node.id == node.id)
194 user_ids_needing_adding = (
195 session.execute(
196 select(User.id)
197 .where(User.is_visible)
198 .where(func.ST_Contains(node_geom, User.geom))
199 .where(~User.id.in_(existing_users))
200 )
201 .scalars()
202 .all()
203 )
204 if user_ids_needing_adding:
205 session.execute(
206 insert(ClusterSubscription),
207 [
208 {"user_id": user_id, "cluster_id": node.official_cluster.id, "role": ClusterRole.member}
209 for user_id in user_ids_needing_adding
210 ],
211 )
212 session.commit()
215def enforce_community_memberships_for_user(session: Session, user: User) -> None:
216 """
217 Adds a given user to all the communities they belong in based on their location.
218 """
219 cluster_ids = (
220 session.execute(
221 select(Cluster.id)
222 .join(Node, Node.id == Cluster.parent_node_id)
223 .where(Cluster.is_official_cluster)
224 .where(func.ST_Contains(Node.geom, user.geom))
225 )
226 .scalars()
227 .all()
228 )
230 for cluster_id in cluster_ids:
231 session.add(
232 ClusterSubscription(
233 user_id=user.id,
234 cluster_id=cluster_id,
235 role=ClusterRole.member,
236 )
237 )
238 session.commit()