-
Notifications
You must be signed in to change notification settings - Fork 1
/
eventbrite.py
116 lines (89 loc) · 3.57 KB
/
eventbrite.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""Handles Eventbrite authorization and API requests."""
from datetime import datetime
import json
import os
import urllib
import re
import requests
from config import EVENTBRITE_APP_KEY, EVENTBRITE_CLIENT_SECRET, EVENTBRITE_OAUTH_TOKEN
EVENTBRITE_SEARCH_ENDPOINT='https://www.eventbriteapi.com/v3/events/search'
EVENTBRITE_BATCH_ENDPOINT='https://www.eventbriteapi.com/v3/batch/'
def search_batch_events(artists, city, distance):
"""Search events with a batched request to the Eventbrite API and return a list of events."""
req_payload = []
# creates URLs to add to the batched request
for artist in artists:
name = artist[1]
query_params = {
'q': name,
'location.address': city,
'location.within': distance,
'categories': '103',
'expand': 'venue'
}
url_args = urllib.parse.urlencode(query_params)
url = f"/events/search?{url_args}"
req = {'method': 'GET', 'relative_url': url}
req_payload.append(req)
# converts batch to JSON for post request
req_payload = json.dumps(req_payload)
batch = {"batch": req_payload}
headers = {'Authorization': f"Bearer {EVENTBRITE_OAUTH_TOKEN}"}
response = requests.post(EVENTBRITE_BATCH_ENDPOINT, data=batch, headers=headers)
results = []
response = response.json()
# check validity of responses
for r in response:
body = json.loads(r['body'])
if body.get('status_code') == 400:
return results
else:
break
filtered_events = filter_events(artists, response)
if filtered_events:
formatted_events = format_batch_events(filtered_events)
results.extend(formatted_events)
return results
else:
return results
def filter_events(artists, response):
"""Decode JSON and filter Eventbrite event data for relevant events."""
initial_matches = []
initial_events = []
filtered_events = []
for r in response:
decoded_events = json.loads(r['body'])
if decoded_events['events']:
for event in decoded_events['events']:
initial_matches.append(event)
for match in initial_matches:
for artist in artists:
name = artist[1].lower()
title = match['name']['text'].lower()
if re.search(r'\b{}(?!\w|[?\'.,!])\b'.format(name), title):
initial_events.append(match)
for i, event in enumerate(initial_events):
if event not in initial_events[i+1:]:
filtered_events.append(event)
return filtered_events
def format_batch_events(filtered_events):
"""Return a list of formatted event data."""
formatted_events = []
for event in filtered_events:
address = f"{event['venue']['address']['address_1']}, {event['venue']['address']['city']}, {event['venue']['address']['region']}, {event['venue']['address']['postal_code']}"
iso_starts_at = event['start']['utc']
iso_ends_at = event['end']['utc']
if not event['logo']:
continue
formatted_event = {
'name': event['name']['text'],
'description': event['description']['text'],
'starts_at': datetime.strptime(iso_starts_at, '%Y-%m-%dT%H:%M:%SZ'),
'ends_at': datetime.strptime(iso_ends_at, '%Y-%m-%dT%H:%M:%SZ'),
'venue': event['venue']['name'],
'address': address,
'url': event['url'],
'img': event['logo']['original']['url']
}
formatted_events.append(formatted_event)
return formatted_events