#
# Aura Engine
#
# Copyright (C) 2020 David Trattnig <david.trattnig@subsquare.at>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# Meta
__version__ = '0.0.1'
__license__ = "GNU Affero General Public License (AGPL) Version 3"
__version_info__ = (0, 0, 1)
__author__ = 'David Trattnig <david.trattnig@subsquare.at>'
import logging
import os, os.path
from datetime import datetime, date, timedelta
from flask import Flask, Response
from flask_caching import Cache
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from marshmallow import Schema, fields, post_dump
from flask_restful import Api, Resource
from apispec import APISpec
from apispec.ext.marshmallow import MarshmallowPlugin
from apispec_webframeworks.flask import FlaskPlugin
from libraries.base.logger import AuraLogger
from libraries.base.config import AuraConfig
from libraries.database.broadcasts import AuraDatabaseModel, Schedule, Playlist, PlaylistEntry, PlaylistEntryMetaData, TrackService
#
# Initialize the Aura Web App and API.
#
config = AuraConfig()
app = Flask(__name__,
static_url_path='',
static_folder='contrib/aura-player/public/')
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SQLALCHEMY_DATABASE_URI"] = config.get_database_uri()
app.config["CACHE_TYPE"] = "simple"
app.config["CACHE_DEFAULT_TIMEOUT"] = 0
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
cache = Cache(app)
cors = CORS(app, resources={r"/*": {"origins": "*"}})
db = SQLAlchemy(app)
ma = Marshmallow(app)
api = Api(app)
class EngineApi:
"""
Provides the Aura Engine API services.
"""
config = None
api = None
logger = None
trackservice_schema = None
def __init__(self, config, api):
"""
Initializes the API.
Args:
config (AuraConfig): The Engine configuration.
api (Api): The Flask restful API object.
"""
self.config = config
self.logger = AuraLogger(self.config, "engine-api")
self.logger = logging.getLogger("engine-api")
self.api = api
spec.components.schema("TrackService", schema=TrackServiceSchema)
# Schema instances
EngineApi.trackservice_schema = TrackServiceSchema(many=True)
EngineApi.track_schema = TrackServiceSchema()
EngineApi.report_schema = ReportSchema(many=True)
EngineApi.schedule_schema = ScheduleSchema(many=True)
# Define API routes
self.api.add_resource(TrackServiceResource, config.api_prefix + "/trackservice/")
self.api.add_resource(TrackResource, config.api_prefix + "/trackservice/<int:track_id>")
self.api.add_resource(CurrentTrackResource, config.api_prefix + "/trackservice/current")
self.api.add_resource(TracksByDayResource, config.api_prefix + "/trackservice/date/<string:date_string>")
self.api.add_resource(ReportResource, config.api_prefix + "/report/<string:year_month>")
self.api.add_resource(UpcomingSchedulesResource, config.api_prefix + "/schedule/upcoming")
self.logger.info("Engine API routes successfully set!")
# Static resources
@app.route('/trackservice', methods=['GET'])
def trackservice():
content = open(os.path.join("contrib/aura-player/public/", "index.html"))
return Response(content, mimetype="text/html")
# Print the API Spec
# TODO Generates HTML for specification
# self.logger.info(spec.to_dict())
# self.logger.info(spec.to_yaml())
def run(self):
"""
Starts the API server.
"""
# TODO Extend to provide production-grade app server
# Set debug=False if you want to use your native IDE debugger
self.api.app.run(port=self.config.api_port, debug=False)
#
# API SPEC
#
spec = APISpec(
title="Swagger API Specification for Aura Engine",
version="1.0.0",
openapi_version="3.0.2",
plugins=[FlaskPlugin(), MarshmallowPlugin()],
)
#
# API SCHEMA
#
class TrackServiceSchema(ma.Schema):
class Meta:
fields = (
"id",
"schedule.schedule_id",
"schedule.schedule_start",
"schedule.schedule_end",
"schedule.languages",
"schedule.type",
"schedule.category",
"schedule.topic",
"schedule.musicfocus",
"schedule.is_repetition",
"schedule.show_id",
"schedule.show_name",
"schedule.show_hosts",
"track",
"track_start"
)
class ScheduleSchema(ma.Schema):
class Meta:
fields = (
"id",
"schedule_id",
"schedule_start",
"schedule",
"show_id",
"show_name",
"show_hosts",
"show_type",
)
class ReportSchema(ma.Schema):
class Meta:
fields = (
"id",
"schedule.schedule_id",
"schedule.schedule_start",
"schedule.schedule_end",
"schedule.languages",
"schedule.type",
"schedule.category",
"schedule.topic",
"schedule.musicfocus",
"schedule.is_repetition",
"schedule.show_id",
"schedule.show_name",
"schedule.show_hosts",
"schedule.show_type",
"schedule.show_funding_category",
"track",
"track_start",
"playlist_id",
"fallback_type",
"schedule_fallback_id",
"show_fallback_id",
"station_fallback_id"
)
#
# API RESOURCES
#
class TrackServiceResource(Resource):
logger = None
def __init__(self):
self.logger = logging.getLogger("engine-api")
def get(self):
today = date.today()
today = datetime(today.year, today.month, today.day)
tracks = TrackService.select_by_day(today)
return EngineApi.trackservice_schema.dump(tracks)
class TrackResource(Resource):
logger = None
def __init__(self):
self.logger = logging.getLogger("engine-api")
def get(self, track_id):
track = TrackService.select_one(track_id)
return EngineApi.track_schema.dump(track)
class CurrentTrackResource(Resource):
logger = None
def __init__(self):
self.logger = logging.getLogger("engine-api")
def get(self, track_id):
track = TrackService.select_current()
return EngineApi.track_schema.dump(track)
class TracksByDayResource(Resource):
logger = None
def __init__(self):
self.logger = logging.getLogger("engine-api")
def get(self, date_string):
date = datetime.strptime(date_string, "%Y-%m-%d")
self.logger.debug("Query track-service by day: %s" % str(date))
tracks = TrackService.select_by_day(date)
return EngineApi.trackservice_schema.dump(tracks)
class UpcomingSchedulesResource(Resource):
logger = None
def __init__(self):
self.logger = logging.getLogger("engine-api")
def get(self):
now = datetime.now()
self.logger.debug("Query upcoming schedules after %s" % str(now))
schedules = Schedule.select_upcoming(3)
return EngineApi.schedule_schema.dump(schedules)
class ReportResource(Resource):
logger = None
def __init__(self):
self.logger = logging.getLogger("engine-api")
def get(self, year_month):
year = int(year_month.split("-")[0])
month = int(year_month.split("-")[1])
first_day = datetime(year, month, 1)
next_month = first_day.replace(day=28) + timedelta(days=4)
next_month - timedelta(days=next_month.day)
self.logger.debug("Query report for month: %s - %s" % (str(first_day), str(next_month)))
report = TrackService.select_by_range(first_day, next_month)
return EngineApi.report_schema.dump(report)
# Initialization calls
if __name__ == "__main__":
engine_api = EngineApi(config, api)
engine_api.run()