Coverage for app/logic/campusGroups.py: 63%

1from datetime import datetime 

2from os import abort 

3from flask import abort, jsonify 

4import requests, xmltodict 

5import xml.etree.ElementTree as ET 

6 

7from app import app 

8from app.models.event import Event 

9 

10EVENT_FIELDS = [ 

11 "cg_event_id", # Integer — 0 to create, existing id to update 

12 "cg_group_acronym", # String 

13 "external_event_id", # String[100] 

14 "event_coordinator", # String[300], email 

15 "event_name", # String 

16 "quick_description", # String[4294967295] 

17 "event_type", # String[255] 

18 "event_start_date", # String[16] 

19 "event_start_time", # String[16] 

20 "event_end_date", # String[16] 

21 "event_end_time", # String[16] 

22 "event_location", # String[255] 

23 "room_id", # Integer 

24 "event_display_to", # Integer - see docs for magic numbers 

25 "allow_rsvp", # Integer, 0 or 1 

26 "event_open_to", # Integer - see docs for magic numbers 

27 "delete_event", # Integer, 0 or 1 

28 "hide_from_events_slider", # Integer, 0 or 1 

29 "force_display_on_rooms_schedule", # Integer, 0 or 1 

30 "location_type", # Integer, 0/2/3 

31 "event_audience", # Integer 

32] 

33 

34REQUIRED_FIELDS = [ 

35 "cg_group_acronym", 

36 "event_name", 

37 "event_start_date", 

38 "event_start_time", 

39 "event_end_date", 

40 "event_end_time" 

41] 

42 

43# The following are also required fields, but are NOT passed into CampusGroups objects (they are retrieved from the config files): 

44 # "api_key", # String[100], Required 

45 # "timestamp", # String[20], yyyy-MM-ddTHH:mm:ssZ (UTC) 

46 # "school", # String[30] 

47 

48class CampusGroups: 

49 def __init__(self, campusGroupsEnv = "sandbox"): 

50 self.url = app.config["campusgroups"][campusGroupsEnv]["url"] 

51 self.secret = app.config["campusgroups"][campusGroupsEnv]["secret"] 

52 self.key = app.config["campusgroups"][campusGroupsEnv]["key"] 

53 self.school = app.config["campusgroups"][campusGroupsEnv]["school"] 

54 self.headers = {'X-CG-API-Secret': self.secret} 

55 self.event = None 

56 

57 def getEvents(self): 

58 """ 

59 Retrieve events from CampusGroups using the RSS feed. 

60 """ 

61 now = datetime.now() 

62 ts_now = now.isoformat().replace('+00:00', 'Z') 

63 rss_events_url = f'{self.url}/rss_events?ts={ts_now}&preauth={self.key}' 

64 response = requests.get(rss_events_url, headers = self.headers) 

65 response.raise_for_status() 

66 data_dict = xmltodict.parse(response.text) 

67 return jsonify(data_dict) 

68 

69 

70 def addEvent(self, eventData): 

71 """ 

72 Add or update an event in CampusGroups using the CreateUpdateEvent SOAP API. 

73 """ 

74 createUpdateEvent_url = "https://berea-sandbox.campusgroups.com/WebServices/campusgroups.asmx?op=CreateUpdateEvent" 

75 xmlOut = self.build_event_xml(eventData) 

76 

77 self.headers["SOAPAction"] = "http://campusgroups.com/CreateUpdateEvent" 

78 self.headers["Content-Type"] = "text/xml; charset=utf-8" 

79 

80 

81 response = requests.post(createUpdateEvent_url, data=xmlOut, headers=self.headers, timeout=15) 

82 response.raise_for_status() 

83 

84 return self.parse_event_response(response.content) 

85 

86 

87 def build_event_xml(self, payload): 

88 """ 

89 Build the SOAP XML body for CreateUpdateEvent from a dictionary payload. 

90 """ 

91 missing = [f for f in REQUIRED_FIELDS if not payload.get(f)] 

92 if missing: 

93 raise ValueError(f"Missing required field(s): {', '.join(missing)}") 

94 

95 envelope = ET.Element( 

96 "soap12:Envelope", 

97 { 

98 "xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance", 

99 "xmlns:xsd": "http://www.w3.org/2001/XMLSchema", 

100 "xmlns:soap12": "http://www.w3.org/2003/05/soap-envelope", 

101 }, 

102 ) 

103 body = ET.SubElement(envelope, "soap12:Body") 

104 op = ET.SubElement(body, "CreateUpdateEvent", {"xmlns": "http://campusgroups.com/"}) 

105 

106 # Always-required auth/context fields 

107 now = datetime.now() 

108 ts_now = now.isoformat().replace('+00:00', 'Z') 

109 

110 ET.SubElement(op, "timestamp").text = (ts_now) 

111 ET.SubElement(op, "school").text = self.school 

112 ET.SubElement(op, "api_key").text = self.key 

113 # Event-specific fields — only send what's provided so partial 

114 # updates don't clobber existing values. 

115 for field in EVENT_FIELDS: 

116 value = payload.get(field) 

117 if value is not None and value != "": 

118 ET.SubElement(op, field).text = str(value) 

119 

120 xml_bytes = ET.tostring(envelope, encoding="utf-8") 

121 return b'<?xml version="1.0" encoding="utf-8"?>' + xml_bytes 

122 

123 def parseEventData(self, eventID): 

124 """ 

125 Parse the event data from Celts-link database into a dictionary. 

126 """ 

127 

128 # TODO handle recurring events  

129 try: 

130 self.event = Event.get_by_id(eventID) 

131 except Event.DoesNotExist: 

132 abort(404, description=f"No events with ID {eventID} found.") 

133 data = {} 

134 

135 if self.event.campusGroupsId is None: 

136 data["cg_event_id"] = 0 # create a new event 

137 else: 

138 data["cg_event_id"] = self.event.campusGroupsId #update an existing event 

139 

140 data["cg_group_acronym"] = "Celts" # event.program? 

141 data["external_event_id"] = self.event.id 

142 data["event_coordinator"] = self.event.contactEmail 

143 data["event_name"] = self.event.name 

144 data["quick_description"] = self.event.description 

145 data["event_type"] = "Academic" 

146 data["event_start_date"] = self.event.startDate.strftime("%Y-%m-%d") 

147 data["event_start_time"] = self.event.timeStart.strftime("%H:%M") 

148 data["event_end_date"] = self.event.startDate.strftime("%Y-%m-%d") 

149 data["event_end_time"] = self.event.timeEnd.strftime("%H:%M") 

150 data["event_location"] = self.event.location 

151 

152 # Check CampusGroups API documentation for magic numbers: 

153 data["event_display_to"] = 0 

154 data["allow_rsvp"] = 1 

155 data["event_open_to"] = 1 

156 data["delete_event"] = 0 

157 data["hide_from_events_slider"] = 0 

158 data["force_display_on_rooms_schedule"] = 0 

159 data["location_type"] = 0 

160 

161 return data 

162 

163 def parse_event_response(self, xml_content): 

164 """ 

165 Pull cg_event_id / message / message_code out of the SOAP response, 

166 tolerating either the namespaced or bare tag form. 

167 """ 

168 

169 responseData = xmltodict.parse(xml_content) 

170 

171 def find_text(tag): 

172 el = responseData['soap:Envelope']['soap:Body']['CreateUpdateEventResponse']['CreateUpdateEventResult'].get(tag) 

173 return el if el is not None else None 

174 

175 response = { 

176 "cg_event_id": find_text("cg_event_id"), 

177 "message": find_text("message"), 

178 "message_code": find_text("message_code"), 

179 } 

180 

181 if response["cg_event_id"]: 

182 self.event.campusGroupsId = response["cg_event_id"] 

183 self.event.campusGroupsURL = f"{self.url}/celts/rsvp_boot?id={response['cg_event_id']}" 

184 self.event.isCampusGroupsSynced = True 

185 self.event.save() 

186 

187 return response