mirror of
https://github.com/SickGear/SickGear.git
synced 2024-12-18 08:43:37 +00:00
410 lines
17 KiB
Python
410 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
#
|
|
# This file is part of SickGear.
|
|
#
|
|
# SickGear 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.
|
|
#
|
|
# SickGear 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 SickGear. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
from time import sleep
|
|
|
|
import platform
|
|
import re
|
|
|
|
try:
|
|
import urllib.request as urllib2
|
|
except ImportError:
|
|
import urllib2
|
|
|
|
from sickgear import logger
|
|
from sickgear.helpers import get_url, try_int, parse_xml
|
|
|
|
from _23 import unquote, urlencode
|
|
from six import iteritems
|
|
|
|
|
|
class Plex(object):
|
|
def __init__(self, settings=None):
|
|
|
|
settings = settings or {}
|
|
self._plex_host = settings.get('plex_host') or '127.0.0.1'
|
|
self.plex_port = settings.get('plex_port') or '32400'
|
|
|
|
self.username = settings.get('username', '')
|
|
self.password = settings.get('password', '')
|
|
self.token = settings.get('token', '')
|
|
|
|
self.device_name = settings.get('device_name', '')
|
|
self.client_id = settings.get('client_id') or '5369636B47656172'
|
|
self.machine_client_identifier = ''
|
|
|
|
self.default_home_users = settings.get('default_home_users', '')
|
|
|
|
# Progress percentage to consider video as watched
|
|
# if set to anything > 0, videos with watch progress greater than this will be considered watched
|
|
self.default_progress_as_watched = settings.get('default_progress_as_watched', 0)
|
|
|
|
# Sections to scan. If empty all sections will be looked at,
|
|
# the section id should be used which is the number found be in the url on PlexWeb after /section/[ID]
|
|
self.section_list = settings.get('section_list', [])
|
|
|
|
# Sections to skip scanning, for use when Settings['section_list'] is not specified,
|
|
# the same as section_list, the section id should be used
|
|
self.ignore_sections = settings.get('ignore_sections', [])
|
|
|
|
# Filter sections by paths that are in this array
|
|
self.section_filter_path = settings.get('section_filter_path', [])
|
|
|
|
# Results
|
|
self.show_states = {}
|
|
self.file_count = 0
|
|
|
|
# Conf
|
|
self.config_version = 2.0
|
|
self.use_logger = False
|
|
self.test = None
|
|
self.home_user_tokens = {}
|
|
|
|
if self.username and '' == self.token:
|
|
self.token = self.get_token(self.username, self.password)
|
|
|
|
@property
|
|
def plex_host(self):
|
|
|
|
host = self._plex_host
|
|
if not host.startswith('http'):
|
|
host = 'http://%s' % host
|
|
return host
|
|
|
|
@plex_host.setter
|
|
def plex_host(self, value):
|
|
|
|
self._plex_host = value
|
|
|
|
def log(self, msg, debug=True):
|
|
|
|
try:
|
|
if self.use_logger:
|
|
msg = 'Plex:: ' + msg
|
|
if debug:
|
|
logger.debug(msg)
|
|
else:
|
|
logger.log(msg)
|
|
# else:
|
|
# print(msg.encode('ascii', 'replace').decode())
|
|
except (BaseException, Exception):
|
|
pass
|
|
|
|
def get_token(self, user, passw):
|
|
|
|
auth = ''
|
|
try:
|
|
auth = get_url('https://plex.tv/users/sign_in.json',
|
|
headers={'X-Plex-Device-Name': 'SickGear',
|
|
'X-Plex-Platform': platform.system(), 'X-Plex-Device': platform.system(),
|
|
'X-Plex-Platform-Version': platform.release(),
|
|
'X-Plex-Provides': 'Python', 'X-Plex-Product': 'Python',
|
|
'X-Plex-Client-Identifier': self.client_id,
|
|
'X-Plex-Version': str(self.config_version),
|
|
'X-Plex-Username': user
|
|
},
|
|
parse_json=True,
|
|
failure_monitor=False,
|
|
post_data=urlencode({b'user[login]': user, b'user[password]': passw}).encode('utf-8')
|
|
)['user']['authentication_token']
|
|
except TypeError:
|
|
self.log('Error in response from plex.tv auth server')
|
|
except IndexError:
|
|
self.log('Error getting Plex Token')
|
|
|
|
return auth
|
|
|
|
def get_access_token(self, token):
|
|
|
|
resources = self.get_url_x('https://plex.tv/api/resources?includeHttps=1', token=token)
|
|
if None is resources:
|
|
return ''
|
|
|
|
devices = resources.findall('Device')
|
|
for device in devices:
|
|
if 1 == len(devices) \
|
|
or self.machine_client_identifier == device.get('clientIdentifier') \
|
|
or (self.device_name
|
|
and (self.device_name.lower() in device.get('name').lower()
|
|
or self.device_name.lower() in device.get('clientIdentifier').lower())):
|
|
access_token = device.get('accessToken')
|
|
if not access_token:
|
|
return ''
|
|
return access_token
|
|
|
|
connections = device.findall('Connection')
|
|
for connection in connections:
|
|
if self.plex_host == connection.get('address'):
|
|
access_token = device.get('accessToken')
|
|
if not access_token:
|
|
return ''
|
|
uri = connection.get('uri')
|
|
match = re.compile(r'(http[s]?://.*?):(\d*)').match(uri)
|
|
if match:
|
|
self.plex_host = match.group(1)
|
|
self.plex_port = match.group(2)
|
|
return access_token
|
|
return ''
|
|
|
|
def get_plex_home_user_tokens(self):
|
|
|
|
user_tokens = {}
|
|
|
|
# check Plex is contactable
|
|
home_users = self.get_url_x('https://plex.tv/api/home/users')
|
|
if None is not home_users:
|
|
for user in home_users.findall('User'):
|
|
user_id = user.get('id')
|
|
switch_page = self.get_url_x('https://plex.tv/api/home/users/%s/switch' % user_id, post_data=True)
|
|
if None is not switch_page:
|
|
home_token = 'user' == switch_page.tag and switch_page.get('authenticationToken')
|
|
if home_token:
|
|
username = switch_page.get('title')
|
|
user_tokens[username] = self.get_access_token(home_token)
|
|
return user_tokens
|
|
|
|
def get_url_x(self, url, token=None, **kwargs):
|
|
|
|
if not token:
|
|
token = self.token
|
|
if not url.startswith('http'):
|
|
url = 'http://' + url
|
|
|
|
for x in range(0, 3):
|
|
if 0 < x:
|
|
sleep(0.5)
|
|
try:
|
|
headers = {'X-Plex-Device-Name': 'SickGear',
|
|
'X-Plex-Platform': platform.system(), 'X-Plex-Device': platform.system(),
|
|
'X-Plex-Platform-Version': platform.release(),
|
|
'X-Plex-Provides': 'controller', 'X-Plex-Product': 'Python',
|
|
'X-Plex-Client-Identifier': self.client_id,
|
|
'X-Plex-Version': str(self.config_version),
|
|
'X-Plex-Token': token,
|
|
'Accept': 'application/xml'
|
|
}
|
|
if self.username:
|
|
headers.update({'X-Plex-Username': self.username})
|
|
page = get_url(url, headers=headers, failure_monitor=False, **kwargs)
|
|
if page:
|
|
parsed = parse_xml(page)
|
|
if None is not parsed and len(parsed):
|
|
return parsed
|
|
return None
|
|
|
|
except Exception as e:
|
|
self.log('Error requesting page: %s' % e)
|
|
continue
|
|
return None
|
|
|
|
# uses the Plex API to delete files instead of system functions, useful for remote installations
|
|
def delete_file(self, media_id=0):
|
|
|
|
try:
|
|
endpoint = ('/library/metadata/%s' % str(media_id))
|
|
req = urllib2.Request('%s:%s%s' % (self.plex_host, self.plex_port, endpoint),
|
|
None, {'X-Plex-Token': self.token})
|
|
req.get_method = lambda: 'DELETE'
|
|
urllib2.urlopen(req)
|
|
except (BaseException, Exception):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def get_media_info(video_node):
|
|
|
|
progress = 0
|
|
if None is not video_node.get('viewOffset') and None is not video_node.get('duration'):
|
|
progress = try_int(video_node.get('viewOffset')) * 100 / try_int(video_node.get('duration'))
|
|
|
|
for media in video_node.findall('Media'):
|
|
for part in media.findall('Part'):
|
|
file_name = part.get('file')
|
|
# if '3' > sys.version: # remove HTML quoted characters, only works in python < 3
|
|
# file_name = urllib2.unquote(file_name.encode('utf-8', errors='replace'))
|
|
# else:
|
|
file_name = unquote(file_name)
|
|
|
|
return {'path_file': file_name, 'media_id': video_node.get('ratingKey'),
|
|
'played': int(video_node.get('viewCount') or 0), 'progress': progress}
|
|
|
|
def check_users_watched(self, users, media_id):
|
|
|
|
if not self.home_user_tokens:
|
|
self.home_user_tokens = self.get_plex_home_user_tokens()
|
|
|
|
result = {}
|
|
if 'all' in users:
|
|
users = self.home_user_tokens.keys()
|
|
|
|
for user in users:
|
|
user_media_page = self.get_url_pms('/library/metadata/%s' % media_id, token=self.home_user_tokens[user])
|
|
if None is not user_media_page:
|
|
video_node = user_media_page.find('Video')
|
|
|
|
progress = 0
|
|
if None is not video_node.get('viewOffset') and None is not video_node.get('duration'):
|
|
progress = try_int(video_node.get('viewOffset')) * 100 / try_int(video_node.get('duration'))
|
|
|
|
played = int(video_node.get('viewCount') or 0)
|
|
if not progress and not played:
|
|
continue
|
|
|
|
date_watched = 0
|
|
if (0 < try_int(video_node.get('viewCount'))) or (0 < self.default_progress_as_watched < progress):
|
|
last_viewed_at = video_node.get('lastViewedAt')
|
|
if last_viewed_at and last_viewed_at not in ('', '0'):
|
|
date_watched = last_viewed_at
|
|
|
|
if date_watched:
|
|
result[user] = dict(played=played, progress=progress, date_watched=date_watched)
|
|
else:
|
|
self.log('Do not have the token for %s.' % user)
|
|
|
|
return result
|
|
|
|
def get_url_pms(self, endpoint=None, **kwargs):
|
|
|
|
return endpoint and self.get_url_x(
|
|
'%s:%s%s' % (self.plex_host, self.plex_port, endpoint), **kwargs)
|
|
|
|
# parse episode information from season pages
|
|
def stat_show(self, node):
|
|
|
|
ep_nodes = []
|
|
if 'directory' == node.tag.lower() and 'show' == node.get('type'):
|
|
show = self.get_url_pms(node.get('key'))
|
|
if None is show: # Check if show page is None or empty
|
|
self.log('Failed to load show page. Skipping...')
|
|
return None
|
|
|
|
for season_node in show.findall('Directory'): # Each directory is a season
|
|
if 'season' != season_node.get('type'): # skips Specials
|
|
continue
|
|
|
|
season_node_key = season_node.get('key')
|
|
season_node = self.get_url_pms(season_node_key)
|
|
if None is not season_node:
|
|
ep_nodes += [season_node]
|
|
|
|
elif 'mediacontainer' == node.tag.lower() and 'episode' == node.get('viewGroup'):
|
|
ep_nodes = [node]
|
|
|
|
check_users = []
|
|
if self.default_home_users:
|
|
check_users = self.default_home_users.strip(' ,').lower().split(',')
|
|
for k in range(0, len(check_users)): # Remove extra spaces and commas
|
|
check_users[k] = check_users[k].strip(', ')
|
|
|
|
for episode_node in ep_nodes:
|
|
for video_node in episode_node.findall('Video'):
|
|
|
|
media_info = self.get_media_info(video_node)
|
|
|
|
if check_users:
|
|
user_info = self.check_users_watched(check_users, media_info['media_id'])
|
|
for user_name, user_media_info in user_info.items():
|
|
self.show_states.update({len(self.show_states): dict(
|
|
path_file=media_info['path_file'],
|
|
media_id=media_info['media_id'],
|
|
played=(100 * user_media_info['played']) or user_media_info['progress'] or 0,
|
|
label=user_name,
|
|
date_watched=user_media_info['date_watched'])})
|
|
else:
|
|
self.show_states.update({len(self.show_states): dict(
|
|
path_file=media_info['path_file'],
|
|
media_id=media_info['media_id'],
|
|
played=(100 * media_info['played']) or media_info['progress'] or 0,
|
|
label=self.username,
|
|
date_watched=video_node.get('lastViewedAt'))})
|
|
|
|
self.file_count += 1
|
|
|
|
return True
|
|
|
|
def fetch_show_states(self, fetch_all=False):
|
|
|
|
error_log = []
|
|
self.show_states = {}
|
|
|
|
server_check = self.get_url_pms('/')
|
|
if None is server_check or 'MediaContainer' != server_check.tag:
|
|
error_log.append('Cannot reach server!')
|
|
|
|
else:
|
|
if not self.device_name:
|
|
self.device_name = server_check.get('friendlyName')
|
|
|
|
if not self.machine_client_identifier:
|
|
self.machine_client_identifier = server_check.get('machineIdentifier')
|
|
|
|
access_token = None
|
|
if self.token:
|
|
access_token = self.get_access_token(self.token)
|
|
if access_token:
|
|
self.token = access_token
|
|
if not self.home_user_tokens:
|
|
self.home_user_tokens = self.get_plex_home_user_tokens()
|
|
else:
|
|
error_log.append('Access Token not found')
|
|
|
|
resp_sections = None
|
|
if None is access_token or len(access_token):
|
|
resp_sections = self.get_url_pms('/library/sections/')
|
|
|
|
if None is not resp_sections:
|
|
|
|
unpather = []
|
|
for loc in self.section_filter_path:
|
|
loc = re.sub(r'[/\\]+', '/', loc.lower())
|
|
loc = re.sub(r'^(.{,2})[/\\]', '', loc)
|
|
unpather.append(loc)
|
|
self.section_filter_path = unpather
|
|
|
|
for section in resp_sections.findall('Directory'):
|
|
if 'show' != section.get('type') or not section.findall('Location'):
|
|
continue
|
|
|
|
section_path = re.sub(r'[/\\]+', '/', section.find('Location').get('path').lower())
|
|
section_path = re.sub(r'^(.{,2})[/\\]', '', section_path)
|
|
if not any(section_path in path for path in self.section_filter_path):
|
|
continue
|
|
|
|
if section.get('key') not in self.ignore_sections \
|
|
and section.get('title') not in self.ignore_sections:
|
|
section_key = section.get('key')
|
|
|
|
for (user, token) in iteritems(self.home_user_tokens or {'': None}):
|
|
self.username = user
|
|
|
|
resp_section = self.get_url_pms('/library/sections/%s/%s' % (
|
|
section_key, ('recentlyViewed', 'all')[fetch_all]), token=token)
|
|
if None is not resp_section:
|
|
view_group = 'MediaContainer' == resp_section.tag and \
|
|
resp_section.get('viewGroup') or ''
|
|
if 'show' == view_group and fetch_all:
|
|
for DirectoryNode in resp_section.findall('Directory'):
|
|
self.stat_show(DirectoryNode)
|
|
elif 'episode' == view_group and not fetch_all:
|
|
self.stat_show(resp_section)
|
|
|
|
if 0 < len(error_log):
|
|
self.log('Library errors...')
|
|
for item in error_log:
|
|
self.log(item)
|
|
|
|
return 0 < len(error_log)
|