-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
189 lines (157 loc) · 6.48 KB
/
app.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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
from flask import Flask, render_template, request, jsonify, Response, send_from_directory, redirect, url_for
from pprint import pprint
import requests
import pandas as pd
from io import StringIO
import json
app = Flask(__name__)
# Constants for API calls
SERVER_URL = 'https://ibt-hsg.herokuapp.com'
REST_KEY = 'Torstrasse25' # fill this with your actual REST key
@app.route('/')
def index():
# This route serves the index.html template
return render_template('index.html')
@app.route('/docs/', defaults={'filename': 'index.html'})
@app.route('/docs/<path:filename>')
def serve_docs(filename):
try:
return send_from_directory('templates/docs', filename)
except FileNotFoundError:
return redirect(url_for('serve_docs', filename='index.html'))
# Helper function to call the external API
def call_api(method, *path_parts, **params):
path_parts = '/'.join(str(part) for part in path_parts)
url = f'{SERVER_URL}/api/{path_parts}/'
headers = {'otree-rest-key': REST_KEY}
resp = method(url, json=params, headers=headers)
if not resp.ok:
msg = (
f'Request to "{url}" failed '
f'with status code {resp.status_code}: {resp.text}'
)
raise Exception(msg)
return resp.json()
# Add the CSV validation logic here
def check_delimiter(content_url, expected_delimiter=';'):
try:
df = pd.read_csv(content_url, sep=expected_delimiter, engine='python')
if df.shape[1] == 1:
return False, None
return True, df
except Exception:
return False, None
@app.route('/validate_csv', methods=['POST'])
def validate_csv():
data = request.json
content_url = data.get('content_url')
expected_delimiter = data.get('delimiter')
delimiter_ok, df = check_delimiter(content_url, expected_delimiter)
if not delimiter_ok:
return jsonify({"error": "CSV file seems to use a wrong delimiter."}), 400
try:
required_columns = ['datetime', 'text', 'replies', 'reposts', 'likes', 'media', 'username']
if not all(column in df.columns for column in required_columns):
return jsonify({"error": "CSV file is missing one or more required columns."}), 400
pd.to_datetime(df['datetime'])
for column in ['replies', 'reposts', 'likes']:
if not pd.api.types.is_numeric_dtype(df[column]):
return jsonify({"error": f"Column '{column}' contains non-numeric values."}), 400
return jsonify({"message": "CSV validation passed."})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route('/create_session', methods=['POST'])
def create_session():
# This route handles the API call based on form data
data = request.json
url_param = 'None'
if data.get('recruitment_platform') == 'Prolific':
url_param = 'PROLIFIC_PID'
try:
response = call_api(
requests.post,
'sessions',
session_config_name='Twitter',
num_participants=data.get('participant_number'),
modified_session_config_fields=dict(
title=data.get('title'),
full_name=data.get('full_name'),
eMail=data.get('eMail'),
study_name=data.get('study_name'),
channel_type=data.get('channel_type'),
data_path=data.get('content_url'),
delimiter=data.get('delimiter'),
topics=not data.get('display_skyscraper'),
url_param=url_param,
survey_link=data.get('survey_url'),
dwell_threshold=data.get('dwell_threshold'),
search_term=data.get('search_term'),
sort_by=data.get('sort_by'),
condition_col=data.get('condition_col'),
briefing=data.get('briefing')
)
)
# print(response)
return jsonify(response)
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route('/submit_completion_code', methods=['POST'])
def submit_completion_code():
data = request.json
print("Received data:", data) # Print the received data
completion_code = data.get('completion_code')
session_code = data.get('session_code')
try:
# Use the call_api function to process the completion_code and session_code
api_response = call_api(
requests.post,
'session_vars',
session_code,
vars=dict(completion_code=completion_code)
)
# Construct the response
response = {
"status": "success",
"api_response": api_response, # The response from the external API
"completion_code": completion_code,
"session_code": session_code
}
print("Response data:", response) # Print the response data
return jsonify(response)
except Exception as e:
error_response = {"error": str(e)}
print("Error response:", error_response) # Print the error response
return jsonify(error_response), 500
@app.route('/api/sessions/<session_code>')
def get_session_data(session_code):
try:
data = call_api(requests.get, 'sessions', session_code)
# You may need to process data here before sending it to the frontend
# return jsonify(data)
pprint(data)
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route('/create_replication_package', methods=['POST'])
def create_replication_package():
data = request.json
csv_url = data['content_url']
configurations = data.get('configurations', {})
try:
response = requests.get(csv_url)
response.raise_for_status()
csv_data = pd.read_csv(StringIO(response.text), delimiter=data.get('delimiter'))
replication_package = {
"configurations": configurations,
"csv_data": csv_data.to_dict(orient='records')
}
package_json = json.dumps(replication_package, indent=4)
download_filename = 'replication_package.json'
response = Response(package_json, mimetype='application/json')
response.headers['Content-Disposition'] = f'attachment; filename={download_filename}'
return response
except requests.RequestException as e:
return jsonify({'status': 'error', 'message': 'Failed to fetch CSV data: ' + str(e)})
except Exception as e:
return jsonify({'status': 'error', 'message': 'An error occurred: ' + str(e)})
if __name__ == '__main__':
app.run(debug=True)