mirror of
https://github.com/Tautulli/Tautulli.git
synced 2025-08-14 02:26:58 -07:00
Initial code for a more suitable websocket event handler.
This commit is contained in:
Tim
parent
fa25809e7a
commit
28227a3c35
8 changed files with 290 additions and 71 deletions
|
|
@ -361,7 +361,7 @@ def dbcheck():
|
|||
'audio_channels INTEGER, transcode_protocol TEXT, transcode_container TEXT, '
|
||||
'transcode_video_codec TEXT, transcode_audio_codec TEXT, transcode_audio_channels INTEGER,'
|
||||
'transcode_width INTEGER, transcode_height INTEGER, buffer_count INTEGER DEFAULT 0, '
|
||||
'buffer_last_triggered INTEGER)'
|
||||
'buffer_last_triggered INTEGER, last_paused INTEGER)'
|
||||
)
|
||||
|
||||
# session_history table :: This is a history table which logs essential stream details
|
||||
|
|
@ -609,6 +609,15 @@ def dbcheck():
|
|||
'ALTER TABLE users ADD COLUMN custom_avatar_url TEXT'
|
||||
)
|
||||
|
||||
# Upgrade sessions table from earlier versions
|
||||
try:
|
||||
c_db.execute('SELECT last_paused from sessions')
|
||||
except sqlite3.OperationalError:
|
||||
logger.debug(u"Altering database. Updating database table sessions.")
|
||||
c_db.execute(
|
||||
'ALTER TABLE sessions ADD COLUMN last_paused INTEGER'
|
||||
)
|
||||
|
||||
# Add "Local" user to database as default unauthenticated user.
|
||||
result = c_db.execute('SELECT id FROM users WHERE username = "Local"')
|
||||
if not result.fetchone():
|
||||
|
|
|
|||
160
plexpy/activity_handler.py
Normal file
160
plexpy/activity_handler.py
Normal file
|
|
@ -0,0 +1,160 @@
|
|||
# This file is part of PlexPy.
|
||||
#
|
||||
# PlexPy is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# PlexPy 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 General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with PlexPy. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import time
|
||||
from plexpy import logger, datafactory, pmsconnect, monitor, threading, notification_handler
|
||||
|
||||
|
||||
class ActivityHandler(object):
|
||||
|
||||
def __init__(self, timeline):
|
||||
self.timeline = timeline
|
||||
|
||||
def is_valid_session(self):
|
||||
if 'sessionKey' in self.timeline:
|
||||
if str(self.timeline['sessionKey']).isdigit():
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def get_session_key(self):
|
||||
if self.is_valid_session():
|
||||
return int(self.timeline['sessionKey'])
|
||||
|
||||
return None
|
||||
|
||||
def get_live_session(self):
|
||||
pms_connect = pmsconnect.PmsConnect()
|
||||
session_list = pms_connect.get_current_activity()
|
||||
|
||||
for session in session_list['sessions']:
|
||||
if int(session['session_key']) == self.get_session_key():
|
||||
return session
|
||||
|
||||
return None
|
||||
|
||||
def update_db_session(self):
|
||||
# Update our session temp table values
|
||||
monitor_proc = monitor.MonitorProcessing()
|
||||
monitor_proc.write_session(self.get_live_session())
|
||||
|
||||
def on_start(self):
|
||||
if self.is_valid_session():
|
||||
logger.debug(u"PlexPy ActivityHandler :: Session %s has started." % str(self.get_session_key()))
|
||||
|
||||
# Fire off notifications
|
||||
threading.Thread(target=notification_handler.notify,
|
||||
kwargs=dict(stream_data=self.get_live_session(), notify_action='play')).start()
|
||||
|
||||
# Write the new session to our temp session table
|
||||
self.update_db_session()
|
||||
|
||||
def on_stop(self):
|
||||
if self.is_valid_session():
|
||||
logger.debug(u"PlexPy ActivityHandler :: Session %s has stopped." % str(self.get_session_key()))
|
||||
|
||||
# Set the session last_paused timestamp
|
||||
data_factory = datafactory.DataFactory()
|
||||
data_factory.set_session_last_paused(session_key=self.get_session_key(), timestamp=None)
|
||||
|
||||
# Update the session state and viewOffset
|
||||
data_factory.set_session_state(session_key=self.get_session_key(),
|
||||
state=self.timeline['state'],
|
||||
view_offset=self.timeline['viewOffset'])
|
||||
|
||||
# Retrieve the session data from our temp table
|
||||
db_session = data_factory.get_session_by_key(session_key=self.get_session_key())
|
||||
|
||||
# Fire off notifications
|
||||
threading.Thread(target=notification_handler.notify,
|
||||
kwargs=dict(stream_data=db_session, notify_action='stop')).start()
|
||||
|
||||
# Write it to the history table
|
||||
monitor_proc = monitor.MonitorProcessing()
|
||||
monitor_proc.write_session_history(session=db_session)
|
||||
|
||||
# Remove the session from our temp session table
|
||||
data_factory.delete_session(session_key=self.get_session_key())
|
||||
|
||||
def on_buffer(self):
|
||||
pass
|
||||
|
||||
def on_pause(self):
|
||||
if self.is_valid_session():
|
||||
logger.debug(u"PlexPy ActivityHandler :: Session %s has been paused." % str(self.get_session_key()))
|
||||
|
||||
# Set the session last_paused timestamp
|
||||
data_factory = datafactory.DataFactory()
|
||||
data_factory.set_session_last_paused(session_key=self.get_session_key(), timestamp=int(time.time()))
|
||||
|
||||
# Update the session state and viewOffset
|
||||
data_factory.set_session_state(session_key=self.get_session_key(),
|
||||
state=self.timeline['state'],
|
||||
view_offset=self.timeline['viewOffset'])
|
||||
|
||||
# Retrieve the session data from our temp table
|
||||
db_session = data_factory.get_session_by_key(session_key=self.get_session_key())
|
||||
|
||||
# Fire off notifications
|
||||
threading.Thread(target=notification_handler.notify,
|
||||
kwargs=dict(stream_data=db_session, notify_action='pause')).start()
|
||||
|
||||
def on_resume(self, time_line=None):
|
||||
if self.is_valid_session():
|
||||
logger.debug(u"PlexPy ActivityHandler :: Session %s has been resumed." % str(self.get_session_key()))
|
||||
|
||||
# Set the session last_paused timestamp
|
||||
data_factory = datafactory.DataFactory()
|
||||
data_factory.set_session_last_paused(session_key=self.get_session_key(), timestamp=None)
|
||||
|
||||
# Update the session state and viewOffset
|
||||
data_factory.set_session_state(session_key=self.get_session_key(),
|
||||
state=self.timeline['state'],
|
||||
view_offset=self.timeline['viewOffset'])
|
||||
|
||||
# Retrieve the session data from our temp table
|
||||
db_session = data_factory.get_session_by_key(session_key=self.get_session_key())
|
||||
|
||||
# Fire off notifications
|
||||
threading.Thread(target=notification_handler.notify,
|
||||
kwargs=dict(stream_data=db_session, notify_action='resume')).start()
|
||||
|
||||
# This function receives events from our websocket connection
|
||||
def process(self):
|
||||
if self.is_valid_session():
|
||||
data_factory = datafactory.DataFactory()
|
||||
db_session = data_factory.get_session_by_key(session_key=self.get_session_key())
|
||||
|
||||
# If we already have this session in the temp table, check for state changes
|
||||
if db_session:
|
||||
this_state = self.timeline['state']
|
||||
last_state = db_session['state']
|
||||
|
||||
if this_state != last_state:
|
||||
# logger.debug(u"PlexPy ActivityHandler :: Last state %s :: Current state %s" %
|
||||
# (last_state, this_state))
|
||||
if this_state == 'paused':
|
||||
self.on_pause()
|
||||
elif last_state == 'paused' and this_state == 'playing':
|
||||
self.on_resume()
|
||||
elif this_state == 'stopped':
|
||||
self.on_stop()
|
||||
# else:
|
||||
# logger.debug(u"PlexPy ActivityHandler :: Session %s state has not changed." %
|
||||
# self.get_session_key())
|
||||
else:
|
||||
# We don't have this session in our table yet, start a new one.
|
||||
# logger.debug(u"PlexPy ActivityHandler :: Session %s has started." % self.get_session_key())
|
||||
self.on_start()
|
||||
|
|
@ -778,3 +778,66 @@ class DataFactory(object):
|
|||
return 'Deleted all items for user_id %s.' % user_id
|
||||
else:
|
||||
return 'Unable to delete items. Input user_id not valid.'
|
||||
|
||||
def get_session_by_key(self, session_key=None):
|
||||
monitor_db = database.MonitorDatabase()
|
||||
|
||||
if str(session_key).isdigit():
|
||||
result = monitor_db.select('SELECT started, session_key, rating_key, media_type, title, parent_title, '
|
||||
'grandparent_title, user_id, user, friendly_name, ip_address, player, '
|
||||
'platform, machine_id, parent_rating_key, grandparent_rating_key, state, '
|
||||
'view_offset, duration, video_decision, audio_decision, width, height, '
|
||||
'container, video_codec, audio_codec, bitrate, video_resolution, '
|
||||
'video_framerate, aspect_ratio, audio_channels, transcode_protocol, '
|
||||
'transcode_container, transcode_video_codec, transcode_audio_codec, '
|
||||
'transcode_audio_channels, transcode_width, transcode_height, '
|
||||
'paused_counter, last_paused '
|
||||
'FROM sessions WHERE session_key = ? LIMIT 1', args=[session_key])
|
||||
for session in result:
|
||||
if session:
|
||||
return session
|
||||
|
||||
return None
|
||||
|
||||
def set_session_state(self, session_key=None, state=None, view_offset=0):
|
||||
monitor_db = database.MonitorDatabase()
|
||||
|
||||
if str(session_key).isdigit() and str(view_offset).isdigit():
|
||||
values = {'view_offset': int(view_offset)}
|
||||
if state:
|
||||
values['state'] = state
|
||||
|
||||
keys = {'session_key': session_key}
|
||||
result = monitor_db.upsert('sessions', values, keys)
|
||||
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
def delete_session(self, session_key=None):
|
||||
monitor_db = database.MonitorDatabase()
|
||||
|
||||
if str(session_key).isdigit():
|
||||
monitor_db.action('DELETE FROM sessions WHERE session_key = ?', [session_key])
|
||||
|
||||
def set_session_last_paused(self, session_key=None, timestamp=None):
|
||||
import time
|
||||
|
||||
if str(session_key).isdigit():
|
||||
monitor_db = database.MonitorDatabase()
|
||||
|
||||
result = monitor_db.select('SELECT last_paused, paused_counter '
|
||||
'FROM sessions '
|
||||
'WHERE session_key = ?', args=[session_key])
|
||||
|
||||
paused_counter = 0
|
||||
for session in result:
|
||||
if session['last_paused']:
|
||||
paused_offset = int(time.time()) - int(session['last_paused'])
|
||||
paused_counter = int(session['paused_counter']) + int(paused_offset)
|
||||
|
||||
values = {'state': 'playing',
|
||||
'last_paused': timestamp,
|
||||
'paused_counter': paused_counter}
|
||||
keys = {'session_key': session_key}
|
||||
monitor_db.upsert('sessions', values, keys)
|
||||
|
|
@ -22,7 +22,8 @@ import time
|
|||
|
||||
monitor_lock = threading.Lock()
|
||||
|
||||
def check_active_sessions():
|
||||
|
||||
def check_active_sessions(ws_request=False):
|
||||
|
||||
with monitor_lock:
|
||||
pms_connect = pmsconnect.PmsConnect()
|
||||
|
|
@ -42,7 +43,8 @@ def check_active_sessions():
|
|||
'container, video_codec, audio_codec, bitrate, video_resolution, '
|
||||
'video_framerate, aspect_ratio, audio_channels, transcode_protocol, '
|
||||
'transcode_container, transcode_video_codec, transcode_audio_codec, '
|
||||
'transcode_audio_channels, transcode_width, transcode_height, paused_counter '
|
||||
'transcode_audio_channels, transcode_width, transcode_height, '
|
||||
'paused_counter, last_paused '
|
||||
'FROM sessions')
|
||||
for stream in db_streams:
|
||||
if any(d['session_key'] == str(stream['session_key']) and d['rating_key'] == str(stream['rating_key'])
|
||||
|
|
@ -59,19 +61,22 @@ def check_active_sessions():
|
|||
# Push it on it's own thread so we don't hold up our db actions
|
||||
threading.Thread(target=notification_handler.notify,
|
||||
kwargs=dict(stream_data=stream, notify_action='pause')).start()
|
||||
|
||||
if session['state'] == 'playing' and stream['state'] == 'paused':
|
||||
# Push any notifications -
|
||||
# Push it on it's own thread so we don't hold up our db actions
|
||||
threading.Thread(target=notification_handler.notify,
|
||||
kwargs=dict(stream_data=stream, notify_action='resume')).start()
|
||||
if stream['state'] == 'paused':
|
||||
|
||||
if stream['state'] == 'paused' and not ws_request:
|
||||
# The stream is still paused so we need to increment the paused_counter
|
||||
# Using the set config parameter as the interval, probably not the most accurate but
|
||||
# it will have to do for now.
|
||||
# it will have to do for now. If it's a websocket request don't use this method.
|
||||
paused_counter = int(stream['paused_counter']) + plexpy.CONFIG.MONITORING_INTERVAL
|
||||
monitor_db.action('UPDATE sessions SET paused_counter = ? '
|
||||
'WHERE session_key = ? AND rating_key = ?',
|
||||
[paused_counter, stream['session_key'], stream['rating_key']])
|
||||
|
||||
if session['state'] == 'buffering' and plexpy.CONFIG.BUFFER_THRESHOLD > 0:
|
||||
# The stream is buffering so we need to increment the buffer_count
|
||||
# We're going just increment on every monitor ping,
|
||||
|
|
@ -160,21 +165,6 @@ def check_active_sessions():
|
|||
logger.debug(u"PlexPy Monitor :: Unable to read session list.")
|
||||
|
||||
|
||||
def get_last_state_by_session(session_key=None):
|
||||
monitor_db = database.MonitorDatabase()
|
||||
|
||||
if str(session_key).isdigit():
|
||||
logger.debug(u"PlexPy Monitor :: Checking state for sessionKey %s..." % str(session_key))
|
||||
query = 'SELECT state FROM sessions WHERE session_key = ? LIMIT 1'
|
||||
result = monitor_db.select(query, args=[session_key])
|
||||
|
||||
if result:
|
||||
return result[0]
|
||||
|
||||
logger.debug(u"PlexPy Monitor :: No session with key %s is active." % str(session_key))
|
||||
return False
|
||||
|
||||
|
||||
class MonitorProcessing(object):
|
||||
|
||||
def __init__(self):
|
||||
|
|
@ -184,7 +174,7 @@ class MonitorProcessing(object):
|
|||
|
||||
values = {'session_key': session['session_key'],
|
||||
'rating_key': session['rating_key'],
|
||||
'media_type': session['type'],
|
||||
'media_type': session['media_type'],
|
||||
'state': session['state'],
|
||||
'user_id': session['user_id'],
|
||||
'user': session['user'],
|
||||
|
|
@ -197,7 +187,7 @@ class MonitorProcessing(object):
|
|||
'platform': session['platform'],
|
||||
'parent_rating_key': session['parent_rating_key'],
|
||||
'grandparent_rating_key': session['grandparent_rating_key'],
|
||||
'view_offset': session['progress'],
|
||||
'view_offset': session['view_offset'],
|
||||
'duration': session['duration'],
|
||||
'video_decision': session['video_decision'],
|
||||
'audio_decision': session['audio_decision'],
|
||||
|
|
|
|||
|
|
@ -705,9 +705,9 @@ class PmsConnect(object):
|
|||
'transcode_container': transcode_container,
|
||||
'transcode_protocol': transcode_protocol,
|
||||
'duration': duration,
|
||||
'progress': progress,
|
||||
'view_offset': progress,
|
||||
'progress_percent': str(helpers.get_percent(progress, duration)),
|
||||
'type': 'track',
|
||||
'media_type': 'track',
|
||||
'indexes': 0
|
||||
}
|
||||
|
||||
|
|
@ -826,14 +826,14 @@ class PmsConnect(object):
|
|||
'transcode_container': transcode_container,
|
||||
'transcode_protocol': transcode_protocol,
|
||||
'duration': duration,
|
||||
'progress': progress,
|
||||
'view_offset': progress,
|
||||
'progress_percent': str(helpers.get_percent(progress, duration)),
|
||||
'indexes': use_indexes
|
||||
}
|
||||
if helpers.get_xml_attr(session, 'ratingKey').isdigit():
|
||||
session_output['type'] = helpers.get_xml_attr(session, 'type')
|
||||
session_output['media_type'] = helpers.get_xml_attr(session, 'type')
|
||||
else:
|
||||
session_output['type'] = 'clip'
|
||||
session_output['media_type'] = 'clip'
|
||||
|
||||
elif helpers.get_xml_attr(session, 'type') == 'movie':
|
||||
session_output = {'session_key': helpers.get_xml_attr(session, 'sessionKey'),
|
||||
|
|
@ -882,14 +882,14 @@ class PmsConnect(object):
|
|||
'transcode_container': transcode_container,
|
||||
'transcode_protocol': transcode_protocol,
|
||||
'duration': duration,
|
||||
'progress': progress,
|
||||
'view_offset': progress,
|
||||
'progress_percent': str(helpers.get_percent(progress, duration)),
|
||||
'indexes': use_indexes
|
||||
}
|
||||
if helpers.get_xml_attr(session, 'ratingKey').isdigit():
|
||||
session_output['type'] = helpers.get_xml_attr(session, 'type')
|
||||
session_output['media_type'] = helpers.get_xml_attr(session, 'type')
|
||||
else:
|
||||
session_output['type'] = 'clip'
|
||||
session_output['media_type'] = 'clip'
|
||||
|
||||
elif helpers.get_xml_attr(session, 'type') == 'clip':
|
||||
session_output = {'session_key': helpers.get_xml_attr(session, 'sessionKey'),
|
||||
|
|
@ -938,9 +938,9 @@ class PmsConnect(object):
|
|||
'transcode_container': transcode_container,
|
||||
'transcode_protocol': transcode_protocol,
|
||||
'duration': duration,
|
||||
'progress': progress,
|
||||
'view_offset': progress,
|
||||
'progress_percent': str(helpers.get_percent(progress, duration)),
|
||||
'type': helpers.get_xml_attr(session, 'type'),
|
||||
'media_type': helpers.get_xml_attr(session, 'type'),
|
||||
'indexes': 0
|
||||
}
|
||||
|
||||
|
|
@ -1027,9 +1027,9 @@ class PmsConnect(object):
|
|||
'transcode_container': transcode_container,
|
||||
'transcode_protocol': transcode_protocol,
|
||||
'duration': '',
|
||||
'progress': '',
|
||||
'view_offset': '',
|
||||
'progress_percent': '100',
|
||||
'type': 'photo',
|
||||
'media_type': 'photo',
|
||||
'indexes': 0
|
||||
}
|
||||
|
||||
|
|
@ -1083,7 +1083,6 @@ class PmsConnect(object):
|
|||
}
|
||||
children_list.append(children_output)
|
||||
|
||||
|
||||
output = {'children_count': helpers.get_xml_attr(xml_head[0], 'size'),
|
||||
'children_type': helpers.get_xml_attr(xml_head[0], 'viewGroup'),
|
||||
'title': helpers.get_xml_attr(xml_head[0], 'title2'),
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@
|
|||
|
||||
# Mostly borrowed from https://github.com/trakt/Plex-Trakt-Scrobbler
|
||||
|
||||
from plexpy import logger
|
||||
from plexpy import logger, monitor
|
||||
|
||||
import threading
|
||||
import plexpy
|
||||
|
|
@ -28,6 +28,9 @@ opcode_data = (websocket.ABNF.OPCODE_TEXT, websocket.ABNF.OPCODE_BINARY)
|
|||
|
||||
|
||||
def start_thread():
|
||||
# Check for any existing sessions on start up
|
||||
monitor.check_active_sessions(ws_request=True)
|
||||
# Start the websocket listener on it's own thread
|
||||
threading.Thread(target=run).start()
|
||||
|
||||
|
||||
|
|
@ -53,9 +56,9 @@ def run():
|
|||
ws = create_connection(uri)
|
||||
reconnects = 0
|
||||
ws_connected = True
|
||||
logger.debug(u'PlexPy WebSocket :: Ready')
|
||||
logger.info(u'PlexPy WebSocket :: Ready')
|
||||
except IOError, e:
|
||||
logger.info(u'PlexPy WebSocket :: %s.' % e)
|
||||
logger.error(u'PlexPy WebSocket :: %s.' % e)
|
||||
reconnects += 1
|
||||
time.sleep(5)
|
||||
|
||||
|
|
@ -73,7 +76,7 @@ def run():
|
|||
if reconnects > 1:
|
||||
time.sleep(2 * (reconnects - 1))
|
||||
|
||||
logger.info(u'PlexPy WebSocket :: Connection has closed, reconnecting...')
|
||||
logger.warn(u'PlexPy WebSocket :: Connection has closed, reconnecting...')
|
||||
try:
|
||||
ws = create_connection(uri)
|
||||
except IOError, e:
|
||||
|
|
@ -108,7 +111,7 @@ def receive(ws):
|
|||
|
||||
|
||||
def process(opcode, data):
|
||||
from plexpy import monitor
|
||||
from plexpy import activity_handler
|
||||
|
||||
if opcode not in opcode_data:
|
||||
return False
|
||||
|
|
@ -126,19 +129,14 @@ def process(opcode, data):
|
|||
return False
|
||||
|
||||
if type == 'playing':
|
||||
logger.debug('%s.playing %s' % (name, info))
|
||||
# logger.debug('%s.playing %s' % (name, info))
|
||||
try:
|
||||
time_line = info.get('_children')
|
||||
except:
|
||||
logger.debug(u"PlexPy WebSocket :: Session found but unable to get timeline data.")
|
||||
return False
|
||||
|
||||
last_session_state = monitor.get_last_state_by_session(time_line[0]['sessionKey'])
|
||||
session_state = time_line[0]['state']
|
||||
|
||||
if last_session_state != session_state:
|
||||
monitor.check_active_sessions()
|
||||
else:
|
||||
logger.debug(u"PlexPy WebSocket :: Session %s state has not changed." % time_line[0]['sessionKey'])
|
||||
activity = activity_handler.ActivityHandler(timeline=time_line[0])
|
||||
activity.process()
|
||||
|
||||
return True
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue