import atexit import json from datetime import datetime from slugify import slugify from .config import Config config = Config() SPACES = int(config['json_indent']) if config['json_indent'] else None from tinydb import TinyDB, Query, operations dbfile = config['db_file'] or '/tmp/ftracker-db.json' db = TinyDB(dbfile, indent=SPACES) from flask import Flask, request, redirect app = Flask(__name__, static_folder='../web') def shutdown(): print('\rReceived stop signal, stopping threads...') db.close() atexit.register(shutdown) @app.route('/guidelines') def get_guidelines(): dest = config['guideline_url'] or None if dest: return redirect(dest) return "No guideline document was configured.", 404 @app.route('/checkin', methods=['POST']) def post_arrival(): try: payload = request.data.decode('UTF-8') data = json.loads(payload) except ValueError as e: return 'Error: JSON decode error:\n' + str(e), 400 if not ('checkpoint' in data and 'name' in data): return "Error: Key missing. See docs/API.md for reference.", 400 name = slugify(data['name']) checkpoint = slugify(data['checkpoint']) now = datetime.utcnow().isoformat() + 'Z' db.insert({ 'name': name, 'checkpoint': checkpoint, 'arrival': now, }) return 'OK', 200 @app.route('/data') def get_data(): if not 'Authorization' in request.headers: return 'Error: No Authorization', 401, {'WWW-Authenticate': 'Basic'} if request.authorization.username != config['admin_user']: return "Wrong username", 403 if request.authorization.password != config['admin_pass']: return "Wrong password", 403 start = request.args.get('start', default = None, type = str) end = request.args.get('end' , default = None, type = str) room = request.args.get('room' , default = None, type = str) # Skip DB query if no parameters given # (Which is currently every request) if not (start or end or room): return json.dumps(db.all(), indent=SPACES), 200 def is_after(val, iso): return (val >= iso if val else True ) if iso else True def is_before(val, iso): return (val <= iso if val else False) if iso else True Entry = Query() r = db.search( (Entry.departure.test(is_after, start)) & (Entry.arrival.test(is_before, end)) & (Entry.room.search(room or ".*")) ) return json.dumps(r, indent=SPACES), 200