-
Notifications
You must be signed in to change notification settings - Fork 0
/
api.py
232 lines (187 loc) · 7.67 KB
/
api.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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
import firebase_admin
import logging
import os
import re
from flask import Flask, request, jsonify, render_template_string, abort
from flask_httpauth import HTTPBasicAuth
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime, timedelta, timezone
from fetch_query import get_or_fetch_notams, fetch_notam_by_ids, fetch_notams_with_interpretations
from gpt_notam import fetch_interpret_and_insert_notams, generate_briefing
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_cors import CORS, cross_origin
from flask_caching import Cache
from firebase_admin import credentials, db
from firebase_auth import auth_required
app = Flask(__name__)
CORS(app)
app.config['CACHE_TYPE'] = 'simple'
cache = Cache(app)
DEAFULT_CACHE_TIMEOUT = 900
limiter = Limiter(key_func=get_remote_address)
limiter.init_app(app)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
RTDB_URL = os.getenv('RTDB_URL')
DEFAULT_USER_POINTS = 5
def is_valid_icao(codes):
"""Check if the given codes are valid ICAO codes."""
pattern = re.compile(r'^[A-Z]{4}$')
for code in codes.split(','):
code = code.strip() # Remove any leading or trailing spaces
if not pattern.match(code):
return False
return True
def is_valid_date(date_str):
"""Check if the given date string is in the format YYYY-MM-DD."""
try:
datetime.strptime(date_str, '%Y-%m-%d')
return True
except ValueError:
return False
@app.route('/api/notams', methods=['GET'])
@auth_required
@limiter.limit("30 per day")
# @cache.memoize(timeout=DEAFULT_CACHE_TIMEOUT)
def get_notams():
"""
This endpoint fetches NOTAMs for the given locations and date range.
It accepts 'locations', 'start_date', and 'end_date' as query parameters.
It returns a list of NOTAM IDs.
"""
batch_load_str = request.args.get('batch_load', default='False').lower()
batch_load = batch_load_str == 'true'
today = datetime.now(timezone.utc).strftime('%Y-%m-%d')
uid = request.headers.get('uid')
# Fetch the user's data from Firebase RTDB
ref = db.reference(f'/users/{uid}')
user_data = ref.get()
# Check if user exists
if not user_data and not batch_load:
return jsonify({'error': 'User not found'}), 404
# Check and update points
if not batch_load:
first_time_use = user_data.get('first_time_use')
current_time = datetime.utcnow()
if first_time_use is None or (current_time - datetime.fromisoformat(first_time_use)) > timedelta(hours=24):
ref.update({
'points': user_data['maximum_points'],
'first_time_use': current_time.isoformat()
})
elif user_data['points'] <= 0:
return jsonify({'error': 'You have exceeded your request limit'}), 429
locations = request.args.getlist('locations')
if not locations:
return jsonify({'error': 'Missing or empty locations parameter'}), 400
for location in locations:
if not is_valid_icao(location):
return jsonify({'error': f'Invalid ICAO code: {location}'}), 400
start_date = request.args.get('start_date') or logger.info(f"Using default start_date {today} for UID {uid}") or today
end_date = request.args.get('end_date') or logger.info(f"Using default end_date {today} for UID {uid}") or today
if not is_valid_date(start_date) or not is_valid_date(end_date):
return jsonify({'error': 'Invalid date format. Expected format: YYYY-MM-DD'}), 400
notams, airports_fetched = get_or_fetch_notams(locations, start_date, end_date)
if not batch_load:
ref.update({
'points': user_data['points'] - airports_fetched
})
if batch_load:
notam_ids = [notam['notam_id'] for notam in notams]
fetch_interpret_and_insert_notams(notams, notam_ids)
return '', 204 # No Content
return jsonify([notam['notam_id'] for notam in notams])
@app.route('/api/notams/<notams_id>', methods=['GET'])
@auth_required
@limiter.limit("30 per day")
# @cache.memoize(timeout=DEAFULT_CACHE_TIMEOUT)
def get_notam(notams_id):
"""
This endpoint fetches a specific NOTAM by its ID.
It also triggers the interpretation of the NOTAM if it hasn't been interpreted yet.
It returns the NOTAM data as a JSON object or as an HTML table based on the output_type parameter.
"""
notams = fetch_notam_by_ids(notams_id)
if notams is None:
return jsonify({'error': 'NOTAM not found'}), 404
fetch_interpret_and_insert_notams(notams, notams_id)
final_notams = fetch_notams_with_interpretations(notams_id)
return jsonify(final_notams), 200
@app.route('/api/briefing/<notams_id>', methods=['GET'])
@auth_required
@limiter.limit("30 per day")
@cache.memoize(timeout=DEAFULT_CACHE_TIMEOUT)
def get_briefing(notams_id):
"""
This endpoint fetches a specific NOTAM by its ID.
It also triggers the interpretation of the NOTAM if it hasn't been interpreted yet.
It returns the NOTAM data as a JSON object.
"""
# output_type = request.args.get('output_type', 'json')
role = request.args.get('role', 'flight distpacher')
if notams_id is None:
return jsonify({'error': 'notams_id parameter not found'}), 404
# fetch_interpret_and_insert_notams(notams, notams_id)
final_notams = generate_briefing(notams_id, role)
return jsonify(final_notams), 200
@app.route('/api/clear_cache', methods=['POST'])
def clear_cache():
# Get the UID from the request headers or body
uid = request.headers.get('uid') or request.json.get('uid')
if not uid:
abort(400, "Missing UID")
# Fetch the user's data from Firebase RTDB
ref = db.reference(f'/users/{uid}')
user_data = ref.get()
# Check if user exists
if not user_data:
abort(404, "User not found")
# Check if the user is an admin
if user_data.get('role') != "admin":
abort(403, "Forbidden: User is not an admin")
# Clear the cache for specific functions
cache.clear()
return jsonify({'message': 'Cache cleared successfully'}), 200
# User Get Post
cred = credentials.Certificate('notamify-firebase-adminsdk-j4kwm-0a46563068.json')
firebase_admin.initialize_app(cred, {
'databaseURL': RTDB_URL
})
@app.route('/api/save_data', methods=['POST'])
def save_data():
uid = request.json.get('uid') # Assuming you're receiving the UID directly in the request
data = request.json.get('data')
ref = db.reference(f'/users/{uid}')
ref.set(data)
return jsonify({'message': 'Data saved successfully'}), 200
@app.route('/api/get_data/<uid>', methods=['GET'])
def get_data(uid):
ref = db.reference(f'/users/{uid}')
data = ref.get()
return jsonify(data), 200
@app.route('/api/post_signup', methods=['POST'])
def post_signup():
try:
# Try to get JSON data from the request
data = request.get_json(force=True)
except:
return jsonify({'error': 'Invalid JSON data'}), 400
# Get the user's UID and name from the data
uid = data.get('uid')
name = data.get('name')
# Check if UID and name are provided
if not uid or not name:
return jsonify({'error': 'UID and name are required'}), 400
ref = db.reference(f'/users/{uid}')
# Set the user's data
ref.set({
'name': name,
'maximum_points': DEFAULT_USER_POINTS,
'points': DEFAULT_USER_POINTS,
'first_time_use': None
})
return jsonify({'message': 'User created successfully'}), 200
if __name__ == "__main__":
port = int(os.environ.get("PORT", 5000))
app.run(debug=True, host='0.0.0.0', port=port)
# app.run(debug=True, host='127.0.0.1', port=8080)