Coverage for app/backend/src/couchers/email/calendar_events.py: 96%

1from datetime import UTC, datetime 

2from email.headerregistry import Address 

3from typing import Literal 

4from zoneinfo import ZoneInfo 

5 

6from ics import Calendar, Event, Geo 

7from ics.grammar.parse import ContentLine # type: ignore[import-untyped] 

8 

9from couchers import urls 

10from couchers.config import config 

11from couchers.email.locales import get_emails_i18next 

12from couchers.i18n import LocalizationContext 

13from couchers.markup import markdown_to_plaintext 

14from couchers.proto import events_pb2, messages_pb2, requests_pb2 

15from couchers.proto.internal.jobs_pb2 import EmailPart 

16from couchers.utils import now, to_aware_datetime 

17 

18HOST_REQUEST_ICS_FILENAME = "host_request.ics" 

19 

20 

21def create_host_request_attachment( 

22 host_request: requests_pb2.HostRequest, other_name: str, hosting: bool, loc_context: LocalizationContext 

23) -> EmailPart: 

24 calendar = create_host_request_ics_calendar(host_request, other_name, hosting, loc_context) 

25 return ics_calendar_to_attachment(calendar, HOST_REQUEST_ICS_FILENAME) 

26 

27 

28def create_host_request_ics_calendar( 

29 host_request: requests_pb2.HostRequest, other_name: str, hosting: bool, loc_context: LocalizationContext 

30) -> Calendar: 

31 event = create_host_request_ics_event(host_request, other_name, hosting, loc_context) 

32 

33 # METHOD:PUBLISH means this is part of a stream of calendar event information. 

34 # It allows for later cancellation, and doesn't expose accept/decline functionality. 

35 # METHOD:CANCEL might leave the event in cancelled state or not work. 

36 return ics_event_to_calendar(event, "PUBLISH", loc_context) 

37 

38 

39def create_host_request_ics_event( 

40 host_request: requests_pb2.HostRequest, other_name: str, hosting: bool, loc_context: LocalizationContext 

41) -> Event: 

42 """Creates an ics event for a host request.""" 

43 

44 event = Event() 

45 event.uid = _event_uid(host_request.host_request_id, kind="host_request") 

46 _set_sequence_timestamp(event, now()) 

47 

48 title: str 

49 if hosting: 

50 title = loc_context.localize_string( 

51 "calendar_events.host_requests.title_host", i18next=get_emails_i18next(), substitutions={"name": other_name} 

52 ) 

53 else: 

54 title = loc_context.localize_string( 

55 "calendar_events.host_requests.title_surfer", 

56 i18next=get_emails_i18next(), 

57 substitutions={"name": other_name}, 

58 ) 

59 

60 event.name = _final_title( 

61 title, 

62 loc_context, 

63 is_cancelled=host_request.status == messages_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CANCELLED, 

64 ) 

65 

66 # Our to_date is inclusive, iCalendar's DTEND is exclusive (for full-day events) 

67 # make_all_day will adjust the end date by one day accordingly. 

68 event.begin = host_request.from_date 

69 event.end = host_request.to_date 

70 event.make_all_day() 

71 

72 event.location = host_request.hosting_city 

73 event.url = urls.host_request(host_request_id=str(host_request.host_request_id)) 

74 

75 # Google Calendar™ will hide the URL if there is a location, so also include it in the description 

76 event.description = event.url 

77 

78 if host_request.status == messages_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CANCELLED: 

79 event.status = "CANCELLED" 

80 

81 return event 

82 

83 

84def create_event_ics_calendar(event: events_pb2.Event, loc_context: LocalizationContext) -> Calendar: 

85 ics_event = create_event_ics_event(event, loc_context) 

86 

87 # METHOD:PUBLISH means this is part of a stream of calendar event information. 

88 # It allows for later cancellation, and doesn't expose accept/decline functionality. 

89 return ics_event_to_calendar(ics_event, "PUBLISH", loc_context) 

90 

91 

92def create_event_ics_event(event: events_pb2.Event, loc_context: LocalizationContext) -> Event: 

93 """Creates an ics event for a host request.""" 

94 

95 ics_event = Event() 

96 ics_event.uid = _event_uid(event.event_id, kind="event") 

97 ics_event.name = _final_title(event.title, loc_context, is_cancelled=event.is_cancelled) 

98 

99 last_update_datetime = to_aware_datetime(event.created if event.last_edited.seconds == 0 else event.last_edited) 

100 ics_event.last_modified = last_update_datetime 

101 _set_sequence_timestamp(ics_event, last_update_datetime) 

102 

103 _set_datetime_with_timezone(ics_event, "DTSTART", to_aware_datetime(event.start_time, event.timezone)) 

104 _set_datetime_with_timezone(ics_event, "DTEND", to_aware_datetime(event.end_time, event.timezone)) 

105 

106 ics_event.location = event.location.address 

107 ics_event.geo = Geo(event.location.lat, event.location.lng) 

108 url = urls.event_link(occurrence_id=event.event_id, slug=event.slug) 

109 ics_event.url = url 

110 # Google Calendar™ will hide the URL if there is a location, so also include it in the description 

111 ics_event.description = markdown_to_plaintext(event.content) + "\n\n" + url 

112 

113 if event.is_cancelled: 

114 ics_event.status = "CANCELLED" 

115 

116 return ics_event 

117 

118 

119def _final_title(title: str, loc_context: LocalizationContext, *, is_cancelled: bool) -> str: 

120 if is_cancelled: 

121 title = loc_context.localize_string( 

122 "calendar_events.title_cancelled", i18next=get_emails_i18next(), substitutions={"title": title} 

123 ) 

124 return title 

125 

126 

127def _event_uid(item_id: int, *, kind: Literal["host_request"] | Literal["event"]) -> str: 

128 uid_domain = Address(addr_spec=config.NOTIFICATION_EMAIL_ADDRESS).domain 

129 return f"{kind}.{item_id}@{uid_domain}" 

130 

131 

132def _set_sequence_timestamp(event: Event, dt: datetime) -> None: 

133 # SEQUENCE is 32-bit, so only support second granularity to avoid overflows 

134 # A better implementation would need to reply on a stored sequence number. 

135 timestamp = round(dt.timestamp()) 

136 event.extra.append(ContentLine(name="SEQUENCE", value=str(timestamp))) 

137 

138 

139def _set_datetime_with_timezone(event: Event, name: str, dt: datetime) -> None: 

140 """ 

141 Adds a property whose value is a datetime with a timezone. 

142 Working around that ics's native Event.begin/end properties don't encode the timezone. 

143 """ 

144 datetime_format = "%Y%m%dT%H%M%S" 

145 params: dict[str, list[str]] = {} 

146 value: str 

147 if isinstance(dt.tzinfo, ZoneInfo) and dt.tzinfo.key != "Etc/UTC": 147 ↛ 151line 147 didn't jump to line 151 because the condition on line 147 was always true

148 params["TZID"] = [dt.tzinfo.key] 

149 value = dt.strftime(datetime_format) 

150 else: 

151 value = dt.astimezone(UTC).strftime(datetime_format) + "Z" 

152 event.extra.append(ContentLine(name, params, value=value)) 

153 

154 

155def ics_event_to_calendar(event: Event, method: str | None, loc_context: LocalizationContext) -> Calendar: 

156 # PRODID is mandatory and generally follows "-//[Organization]//[Product Name]//[Language]" 

157 calendar = Calendar(creator=f"-//Couchers.org//Couchers//{loc_context.locale.upper()}") 

158 if method: 158 ↛ 160line 158 didn't jump to line 160 because the condition on line 158 was always true

159 calendar.method = method 

160 calendar.events.add(event) 

161 return calendar 

162 

163 

164def ics_calendar_to_attachment(calendar: Calendar, filename: str) -> EmailPart: 

165 data = calendar.serialize().encode("utf-8") 

166 content_disposition = f'attachment; filename="{filename}"' 

167 content_type = 'text/calendar; charset="utf-8"' 

168 if calendar.method: 168 ↛ 173line 168 didn't jump to line 173 because the condition on line 168 was always true

169 # The SMTP Content-Type "method" parameter must match the value in the ics file. 

170 # AI recommends avoiding quotes on this parameter for backwards compatibility with old email clients. 

171 content_type += f"; method={calendar.method}" 

172 

173 return EmailPart(data=data, content_disposition=content_disposition, content_type=content_type)