api.py

#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.

# 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/>.



import logging
import requests
import queue
import threading


from src.base.utils import SimpleUtil as SU
from src.scheduling.utils import TimeslotFilter



class ApiFetcher(threading.Thread):
    """
    Fetches the timeslots, playlists and playlist entries as JSON
    via the API endpoints of Steering and Tank.
    """
    config = None
    logging = None
    queue = None
    has_already_fetched = False
    fetched_timeslot_data = None
    stop_event = None

    # Config for API Endpoints
    steering_calendar_url = None
    tank_playlist_url = None
    tank_session = None
    tank_secret = None



    def __init__(self, config):
        """
        Constructor
        """
        self.config = config
        self.logger = logging.getLogger("AuraEngine")
        self.steering_calendar_url = self.config.get("api_steering_calendar")
        self.tank_playlist_url = self.config.get("api_tank_playlist")
        self.tank_session = self.config.get("api_tank_session")
        self.tank_secret = self.config.get("api_tank_secret")
        self.queue = queue.Queue()
        self.stop_event = threading.Event()
        threading.Thread.__init__(self)



    def run(self):
        """
        Fetch timeslot data from the API.

        Returns
            Timeslot ([dict]):  An array of retrieved timeslots dictionary
        """
        try:
            fetched_timeslots = self.fetch()
            self.logger.debug("Timeslot data fetched from API: " + str(fetched_timeslots))

            # If nothing is fetched, return
            if not fetched_timeslots:
                self.queue.put("fetching_aborted Nothing fetched")
                return None

            # Release the mutex
            self.queue.put(fetched_timeslots)
        except Exception as e:
            # Release the mutex
            self.logger.warning("Fetching aborted due to: %s" % str(e), e)
            self.queue.put("fetching_aborted " + str(e))

        # Terminate the thread
        return



    #
    #   METHODS
    #


    def get_fetched_data(self):
        """
        Retrieves the fetched data from the queue.
        """
        return self.queue.get()



    def fetch(self):
        """
        Retrieve all required data from the API.
        """
        return_data = []

        self.logger.debug("Fetching timeslots from STEERING")
        self.fetched_timeslot_data = self.fetch_timeslot_data()
        if not self.fetched_timeslot_data:
            self.logger.critical(SU.red("No timeslots fetched from API!"))
            return None

        for timeslot in self.fetched_timeslot_data:

            if "schedule_default_playlist_id" in timeslot:
                timeslot["default_schedule_playlist_id"] = timeslot["schedule_default_playlist_id"]
                timeslot["schedule_fallback_id"] = None
            if "show_default_playlist_id" in timeslot:
                timeslot["default_show_playlist_id"] = timeslot["show_default_playlist_id"]
                timeslot["show_fallback_id"] = None

        self.logger.debug("Fetching playlists from TANK")
        self.fetch_playlists()

        try:
            for timeslot in self.fetched_timeslot_data:

                # Skip timeslot if no start or end is given
                if "start" not in timeslot:
                    self.logger.warning("No start of timeslot given. Skipping timeslot: " + str(timeslot))
                    timeslot = None
                if "end" not in timeslot:
                    self.logger.warning("No end of timeslot given. Skipping timeslot: " + str(timeslot))
                    timeslot = None

                if timeslot:
                    return_data.append(timeslot)
        except TypeError:
            self.logger.error(SU.red("Nothing fetched ..."))
            self.fetched_timeslot_data = None
            return None

        return return_data



    def fetch_timeslot_data(self):
        """
        Fetches timeslot data from Steering.

        Returns:
            ([Timeslot]):   An array of timeslots
        """
        timeslots = None
        headers = { "content-type": "application/json" }

        try:
            self.logger.debug("Fetch timeslots from Steering API...")
            response = requests.get(self.steering_calendar_url, data=None, headers=headers)
            if not response.status_code == 200:
                self.logger.critical(SU.red("HTTP Status: %s | Timeslots could not be fetched! Response: %s" % \
                    (str(response.status_code), response.text)))
                return None
            timeslots = response.json()
        except Exception as e:
            self.logger.critical(SU.red("Error while requesting timeslots from Steering!"), e)

        if not timeslots:
            self.logger.error(SU.red("Got no timeslots via Playout API (Steering)!"))
            return None

        return self.polish_timeslots(timeslots)



    def fetch_playlists(self):
        """
        Fetches all playlists including fallback playlists for every timeslot.
        This method used the class member `fetched_timeslot_data`` to iterate
        over and extend timeslot data.
        """
        # store fetched entries => do not have to fetch playlist_id more than once
        fetched_entries=[]

        try:
            for timeslot in self.fetched_timeslot_data:

                # Get IDs of specific, default and fallback playlists
                playlist_id = self.get_playlist_id(timeslot, "playlist_id")
                default_schedule_playlist_id = self.get_playlist_id(timeslot, "default_schedule_playlist_id")
                default_show_playlist_id = self.get_playlist_id(timeslot, "default_show_playlist_id")
                schedule_fallback_id = self.get_playlist_id(timeslot, "schedule_fallback_id")
                show_fallback_id = self.get_playlist_id(timeslot, "show_fallback_id")
                station_fallback_id = self.get_playlist_id(timeslot, "station_fallback_id")

                # Retrieve playlist, default and the fallback playlists for every timeslot.
                # If a playlist (like station_fallback) is already fetched, it is not fetched again but reused
                timeslot["playlist"] = self.fetch_playlist(playlist_id, fetched_entries)
                timeslot["default_schedule_playlist"] = self.fetch_playlist(default_schedule_playlist_id, fetched_entries)
                timeslot["default_show_playlist"] = self.fetch_playlist(default_show_playlist_id, fetched_entries)
                timeslot["schedule_fallback"] = self.fetch_playlist(schedule_fallback_id, fetched_entries)
                timeslot["show_fallback"] = self.fetch_playlist(show_fallback_id, fetched_entries)
                timeslot["station_fallback"]  = self.fetch_playlist(station_fallback_id, fetched_entries)

        except Exception as e:
            self.logger.error("Error while fetching playlists from API endpoints: " + str(e), e)



    def fetch_playlist(self, playlist_id, fetched_playlists):
        """
        Fetches the playlist for a given timeslot.

        Args:
            playlist_id (String):       The ID of the playlist
            fetched_playlists ([dict]): Previously fetched playlists to avoid re-fetching

        Returns:
            (Playlist):             Playlist for `playlist_id`
        """
        if not playlist_id:
            return None

        playlist = None
        url = self.tank_playlist_url.replace("${ID}", playlist_id)
        headers = {
            "Authorization": "Bearer %s:%s" % (self.tank_session, self.tank_secret),
            "content-type": "application/json"
        }

        # If playlist is already fetched in this round, use the existing one
        for playlist in fetched_playlists:
            if playlist["id"] == playlist_id:
                self.logger.debug("Playlist #%s already fetched" % playlist_id)
                return playlist

        try:
            self.logger.debug("Fetch playlist from Tank API...")
            response = requests.get(url, data=None, headers=headers)
            if not response.status_code == 200:
                self.logger.critical(SU.red("HTTP Status: %s | Playlist #%s could not be fetched or is not available! Response: %s" % \
                    (str(response.status_code), str(playlist_id), response.text)))
                return None
            playlist = response.json()
        except Exception as e:
            self.logger.critical(SU.red("Error while requesting playlist #%s from Tank" % str(playlist_id)), e)
            return None

        fetched_playlists.append(playlist)
        return playlist



    def get_playlist_id(self, timeslot, id_name):
        """
        Extracts the playlist ID for a given playlist (fallback) type.

        Args:
            timeslot (dict):    The timeslot dictionary
            id_name (String):   The dictionary key holding the playlist ID

        Returns:
            (Integer):  The playlist ID
        """
        if not id_name in timeslot:
            return None

        playlist_id = str(timeslot[id_name])
        if not playlist_id or playlist_id == "None":
            self.logger.debug("No value defined for '%s' in timeslot '#%s'" % (id_name, timeslot["id"]))
            return None

        return playlist_id



    def polish_timeslots(self, timeslots):
        """
        Removes all timeslots which are not relevant for further processing,
        and transparent timeslot ID assignment for more expressive use.
        """
        count_before = len(timeslots)
        timeslots = TimeslotFilter.filter_24h(timeslots)
        timeslots = TimeslotFilter.filter_past(timeslots)
        count_after = len(timeslots)
        self.logger.debug("Removed %d unnecessary timeslots from response. Timeslots left: %d" % ((count_before - count_after), count_after))

        for t in timeslots:
            t["timeslot_id"] = t["id"]
        return timeslots



    def terminate(self):
        """
        Terminates the thread.
        """
        self.logger.info("Shutting down API fetcher...")
        self.stop_event.set()