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

1from datetime import date, datetime, timedelta 

2from email.headerregistry import Address 

3from typing import Literal 

4from zoneinfo import ZoneInfo 

5 

6from icalendar import Calendar, Event 

7 

8from couchers import urls 

9from couchers.config import config 

10from couchers.email.locales import get_emails_i18next 

11from couchers.i18n import LocalizationContext 

12from couchers.markup import markdown_to_plaintext 

13from couchers.proto import events_pb2, messages_pb2, requests_pb2 

14from couchers.proto.internal.jobs_pb2 import EmailPart 

15from couchers.utils import now, to_aware_datetime 

16 

17HOST_REQUEST_ICS_FILENAME = "host_request.ics" 

18 

19 

20def create_host_request_attachment( 

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

22) -> EmailPart: 

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

24 return ics_calendar_to_attachment(calendar, HOST_REQUEST_ICS_FILENAME) 

25 

26 

27def create_host_request_ics_calendar( 

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

29) -> Calendar: 

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

31 

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

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

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

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

36 

37 

38def create_host_request_ics_event( 

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

40) -> Event: 

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

42 

43 event = Event() # type: ignore[no-untyped-call] 

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

45 _set_sequence_timestamp(event, now()) 

46 

47 title: str 

48 if hosting: 

49 title = loc_context.localize_string( 

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

51 ) 

52 else: 

53 title = loc_context.localize_string( 

54 "calendar_events.host_requests.title_surfer", 

55 i18next=get_emails_i18next(), 

56 substitutions={"name": other_name}, 

57 ) 

58 

59 event.add( 

60 "summary", 

61 _final_title( 

62 title, 

63 loc_context, 

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

65 ), 

66 ) 

67 

68 # Our to_date is inclusive, iCalendar's DTEND is exclusive (for full-day events), hence the +1 day. 

69 event.add("dtstart", date.fromisoformat(host_request.from_date)) 

70 event.add("dtend", date.fromisoformat(host_request.to_date) + timedelta(days=1)) 

71 

72 event.add("location", host_request.hosting_city) 

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

74 event.add("url", url) 

75 

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

77 event.add("description", url) 

78 

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

80 event.add("status", "CANCELLED") 

81 

82 return event 

83 

84 

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

86 ics_event = create_event_ics_event(event, loc_context) 

87 

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

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

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

91 

92 

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

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

95 

96 ics_event = Event() # type: ignore[no-untyped-call] 

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

98 ics_event.add("summary", _final_title(event.title, loc_context, is_cancelled=event.is_cancelled)) 

99 

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

101 ics_event.add("last-modified", last_update_datetime) 

102 _set_sequence_timestamp(ics_event, last_update_datetime) 

103 

104 timezone = ZoneInfo(event.timezone) 

105 ics_event.add("dtstart", to_aware_datetime(event.start_time).astimezone(timezone)) 

106 ics_event.add("dtend", to_aware_datetime(event.end_time).astimezone(timezone)) 

107 

108 ics_event.add("location", event.location.address) 

109 ics_event.add("geo", (event.location.lat, event.location.lng)) 

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

111 ics_event.add("url", url) 

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

113 ics_event.add("description", markdown_to_plaintext(event.content) + "\n\n" + url) 

114 

115 if event.is_cancelled: 

116 ics_event.add("status", "CANCELLED") 

117 

118 return ics_event 

119 

120 

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

122 if is_cancelled: 

123 title = loc_context.localize_string( 

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

125 ) 

126 return title 

127 

128 

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

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

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

132 

133 

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

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

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

137 timestamp = round(dt.timestamp()) 

138 event.add("sequence", timestamp) 

139 

140 

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

142 calendar = Calendar() # type: ignore[no-untyped-call] 

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

144 calendar.add("prodid", f"-//Couchers.org//Couchers//{loc_context.preferred_locale.upper()}") 

145 calendar.add("version", "2.0") 

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

147 calendar.add("method", method) 

148 calendar.add_component(event) 

149 return calendar 

150 

151 

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

153 data = calendar.to_ical() 

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

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

156 method = calendar.get("method") 

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

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

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

160 content_type += f"; method={method}" 

161 

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