plugin.video.viervijfzes/resources/lib/viervijfzes/epg.py

185 lines
6.1 KiB
Python

# -*- coding: utf-8 -*-
""" EPG API """
from __future__ import absolute_import, division, unicode_literals
import json
import logging
from datetime import datetime, timedelta
import dateutil.parser
import dateutil.tz
import requests
_LOGGER = logging.getLogger(__name__)
GENRE_MAPPING = {
'Detective': 0x11,
'Dramaserie': 0x15,
'Fantasy': 0x13,
'Human Interest': 0x00,
'Informatief': 0x20,
'Komedie': 0x14,
'Komische serie': 0x14,
'Kookprogramma': '',
'Misdaadserie': 0x15,
'Politieserie': 0x17,
'Reality': 0x31,
'Science Fiction': 0x13,
'Show': 0x30,
'Thriller': 0x11,
'Voetbal': 0x43,
}
class EpgProgram:
""" Defines a Program in the EPG. """
# pylint: disable=invalid-name
def __init__(self, channel, program_title, episode_title, episode_title_original, number, season, genre, start,
won_id, won_program_id, program_description, description, duration, program_url, video_url, cover,
airing):
self.channel = channel
self.program_title = program_title
self.episode_title = episode_title
self.episode_title_original = episode_title_original
self.number = number
self.season = season
self.genre = genre
self.start = start
self.won_id = won_id
self.won_program_id = won_program_id
self.program_description = program_description
self.description = description
self.duration = duration
self.program_url = program_url
self.video_url = video_url
self.cover = cover
self.airing = airing
if GENRE_MAPPING.get(self.genre):
self.genre_id = GENRE_MAPPING.get(self.genre)
else:
self.genre_id = None
def __repr__(self):
return "%r" % self.__dict__
class EpgApi:
""" VIER/VIJF/ZES EPG API """
EPG_ENDPOINTS = {
'vier': 'https://www.vier.be/api/epg/{date}',
'vijf': 'https://www.vijf.be/api/epg/{date}',
'zes': 'https://www.zestv.be/api/epg/{date}',
}
EPG_NO_BROADCAST = 'Geen uitzending'
def __init__(self):
""" Initialise object """
self._session = requests.session()
def get_epg(self, channel, date):
""" Returns the EPG for the specified channel and date.
:type channel: str
:type date: str
:rtype list[EpgProgram]
"""
if channel not in self.EPG_ENDPOINTS:
raise Exception('Unknown channel %s' % channel)
if date is None:
# Fetch today when no date is specified
date = datetime.today().strftime('%Y-%m-%d')
elif date == 'yesterday':
date = (datetime.today() + timedelta(days=-1)).strftime('%Y-%m-%d')
elif date == 'today':
date = datetime.today().strftime('%Y-%m-%d')
elif date == 'tomorrow':
date = (datetime.today() + timedelta(days=1)).strftime('%Y-%m-%d')
# Request the epg data
response = self._get_url(self.EPG_ENDPOINTS.get(channel).format(date=date))
data = json.loads(response)
# Parse the results
return [self._parse_program(channel, x) for x in data if x.get('program_title') != self.EPG_NO_BROADCAST]
@staticmethod
def _parse_program(channel, data):
""" Parse the EPG JSON data to a EpgProgram object.
:type channel: str
:type data: dict
:rtype EpgProgram
"""
duration = int(data.get('duration')) if data.get('duration') else None
# Check if this broadcast is currently airing
timestamp = datetime.now().replace(tzinfo=dateutil.tz.gettz('CET'))
start = datetime.fromtimestamp(data.get('timestamp')).replace(tzinfo=dateutil.tz.gettz('CET'))
if duration:
airing = bool(start <= timestamp < (start + timedelta(seconds=duration)))
else:
airing = False
# Only allow direct playing if the linked video is the actual program
if data.get('video_node', {}).get('latest_video'):
video_url = (data.get('video_node', {}).get('url') or '').lstrip('/')
cover = data.get('video_node', {}).get('image')
else:
video_url = None
cover = None
return EpgProgram(
channel=channel,
program_title=data.get('program_title'),
episode_title=data.get('episode_title'),
episode_title_original=data.get('original_title'),
number=int(data.get('episode_nr')) if data.get('episode_nr') else None,
season=data.get('season'),
genre=data.get('genre'),
start=start,
won_id=int(data.get('won_id')) if data.get('won_id') else None,
won_program_id=int(data.get('won_program_id')) if data.get('won_program_id') else None,
program_description=data.get('program_concept'),
description=data.get('content_episode'),
duration=duration,
program_url=(data.get('program_node', {}).get('url') or '').lstrip('/'),
video_url=video_url,
cover=cover,
airing=airing,
)
def get_broadcast(self, channel, timestamp):
""" Load EPG information for the specified channel and date.
:type channel: str
:type timestamp: str
:rtype: EpgProgram
"""
# Parse to a real datetime
timestamp = dateutil.parser.parse(timestamp).replace(tzinfo=dateutil.tz.gettz('CET'))
# Load guide info for this date
programs = self.get_epg(channel=channel, date=timestamp.strftime('%Y-%m-%d'))
# Find a matching broadcast
for broadcast in programs:
if timestamp <= broadcast.start < (broadcast.start + timedelta(seconds=broadcast.duration)):
return broadcast
return None
def _get_url(self, url):
""" Makes a GET request for the specified URL.
:type url: str
:rtype str
"""
response = self._session.get(url)
if response.status_code != 200:
raise Exception('Could not fetch data')
return response.text