Coverage for app/logic/campusGroups.py: 63%
91 statements
« prev ^ index » next coverage.py v7.10.2, created at 2026-09-22 13:52 +0000
1from datetime import datetime
2from os import abort
3from flask import abort, jsonify
4import requests, xmltodict
5import xml.etree.ElementTree as ET
7from app import app
8from app.models.event import Event
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]
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]
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]
48class CampusGroups:
49 def __init__(self):
50 self.url = app.config["campusgroups"]["url"]
51 self.api_path = app.config["campusgroups"]["api_path"]
52 self.key = app.config["campusgroups"]["key"]
53 self.school = app.config["campusgroups"]["school"]
54 self.headers = {'X-CG-API-Secret': app.config["campusgroups"]["secret"]}
55 self.event = None
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)
70 def addEvent(self, eventData):
71 """
72 Add or update an event in CampusGroups using the CreateUpdateEvent SOAP API.
73 """
74 createUpdateEvent_url = f'{self.url}/{self.api_path}?op=CreateUpdateEvent'
75 xmlOut = self.build_event_xml(eventData)
77 self.headers["SOAPAction"] = "http://campusgroups.com/CreateUpdateEvent"
78 self.headers["Content-Type"] = "text/xml; charset=utf-8"
80 response = requests.post(createUpdateEvent_url, data=xmlOut, headers=self.headers, timeout=15)
81 response.raise_for_status()
83 return self.parse_event_response(response.content)
86 def build_event_xml(self, payload):
87 """
88 Build the SOAP XML body for CreateUpdateEvent from a dictionary payload.
89 """
90 missing = [f for f in REQUIRED_FIELDS if not payload.get(f)]
91 if missing:
92 raise ValueError(f"Missing required field(s): {', '.join(missing)}")
94 envelope = ET.Element(
95 "soap12:Envelope",
96 {
97 "xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance",
98 "xmlns:xsd": "http://www.w3.org/2001/XMLSchema",
99 "xmlns:soap12": "http://www.w3.org/2003/05/soap-envelope",
100 },
101 )
102 body = ET.SubElement(envelope, "soap12:Body")
103 op = ET.SubElement(body, "CreateUpdateEvent", {"xmlns": "http://campusgroups.com/"})
105 # Always-required auth/context fields
106 now = datetime.now()
107 ts_now = now.isoformat().replace('+00:00', 'Z')
109 ET.SubElement(op, "timestamp").text = (ts_now)
110 ET.SubElement(op, "school").text = self.school
111 ET.SubElement(op, "api_key").text = self.key
112 # Event-specific fields — only send what's provided so partial
113 # updates don't clobber existing values.
114 for field in EVENT_FIELDS:
115 value = payload.get(field)
116 if value is not None and value != "":
117 ET.SubElement(op, field).text = str(value)
119 xml_bytes = ET.tostring(envelope, encoding="utf-8")
120 return b'<?xml version="1.0" encoding="utf-8"?>' + xml_bytes
122 def parseEventData(self, eventID):
123 """
124 Parse the event data from Celts-link database into a dictionary.
125 """
127 # TODO handle recurring events
128 try:
129 self.event = Event.get_by_id(eventID)
130 except Event.DoesNotExist:
131 abort(404, description=f"No events with ID {eventID} found.")
132 data = {}
134 if self.event.campusGroupsId is None:
135 data["cg_event_id"] = 0 # create a new event
136 else:
137 data["cg_event_id"] = self.event.campusGroupsId #update an existing event
139 data["cg_group_acronym"] = "Celts" # event.program?
140 data["external_event_id"] = self.event.id
141 data["event_coordinator"] = self.event.contactEmail
142 data["event_name"] = self.event.name
143 data["quick_description"] = self.event.description
144 data["event_type"] = "Academic"
145 data["event_start_date"] = self.event.startDate.strftime("%Y-%m-%d")
146 data["event_start_time"] = self.event.timeStart.strftime("%H:%M")
147 data["event_end_date"] = self.event.startDate.strftime("%Y-%m-%d")
148 data["event_end_time"] = self.event.timeEnd.strftime("%H:%M")
149 data["event_location"] = self.event.location
151 # Check CampusGroups API documentation for magic numbers:
152 data["event_display_to"] = 0
153 data["allow_rsvp"] = 1
154 data["event_open_to"] = 1
155 data["delete_event"] = 0
156 data["hide_from_events_slider"] = 0
157 data["force_display_on_rooms_schedule"] = 0
158 data["location_type"] = 0
160 return data
162 def parse_event_response(self, xml_content):
163 """
164 Pull cg_event_id / message / message_code out of the SOAP response,
165 tolerating either the namespaced or bare tag form.
166 """
168 responseData = xmltodict.parse(xml_content)
170 def find_text(tag):
171 el = responseData['soap:Envelope']['soap:Body']['CreateUpdateEventResponse']['CreateUpdateEventResult'].get(tag)
172 return el if el is not None else None
174 response = {
175 "cg_event_id": find_text("cg_event_id"),
176 "message": find_text("message"),
177 "message_code": find_text("message_code"),
178 }
180 if response["cg_event_id"]:
181 self.event.campusGroupsId = response["cg_event_id"]
182 self.event.campusGroupsURL = f"{self.url}/celts/rsvp_boot?id={response['cg_event_id']}"
183 self.event.isCampusGroupsSynced = True
184 self.event.save()
186 return response