Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
filename = os.path.basename(filename)
fullpath = os.path.join(savepath, filename)
# append file.ext from content-type if not already there
extension = os.path.splitext(fullpath)[-1]
if not extension:
contenttype = response.headers.get('content-type')
if contenttype and 'image' in contenttype:
fullpath += contenttype.split('/')[1]
# check this is a mocked download (testing)
if mocked:
log.debug('Mocked download %s', fullpath)
return fullpath
# save the file to disk
log.info('Downloading: %s', fullpath)
if showstatus: # pragma: no cover
total = int(response.headers.get('content-length', 0))
bar = tqdm(unit='B', unit_scale=True, total=total, desc=filename)
with open(fullpath, 'wb') as handle:
for chunk in response.iter_content(chunk_size=chunksize):
handle.write(chunk)
if showstatus:
bar.update(len(chunk))
if showstatus: # pragma: no cover
bar.close()
# check we want to unzip the contents
if fullpath.endswith('zip') and unpack:
with zipfile.ZipFile(fullpath, 'r') as handle:
handle.extractall(savepath)
filename = os.path.basename(filename)
fullpath = os.path.join(savepath, filename)
# append file.ext from content-type if not already there
extension = os.path.splitext(fullpath)[-1]
if not extension:
contenttype = response.headers.get('content-type')
if contenttype and 'image' in contenttype:
fullpath += contenttype.split('/')[1]
# check this is a mocked download (testing)
if mocked:
log.debug('Mocked download %s', fullpath)
return fullpath
# save the file to disk
log.info('Downloading: %s', fullpath)
if showstatus: # pragma: no cover
total = int(response.headers.get('content-length', 0))
bar = tqdm(unit='B', unit_scale=True, total=total, desc=filename)
with open(fullpath, 'wb') as handle:
for chunk in response.iter_content(chunk_size=chunksize):
handle.write(chunk)
if showstatus:
bar.update(len(chunk))
if showstatus: # pragma: no cover
bar.close()
# check we want to unzip the contents
if fullpath.endswith('zip') and unpack:
with zipfile.ZipFile(fullpath, 'r') as handle:
handle.extractall(savepath)
def setWebhooks(self, urls):
log.info('Setting webhooks: %s' % urls)
data = self.query(self.WEBHOOKS, self._session.post, data={'urls[]': urls})
self._webhooks = self.listAttrs(data, 'url', etag='webhook')
return self._webhooks
def connect(self, timeout=None):
""" Returns a new :class:`~plexapi.client.PlexClient` or :class:`~plexapi.server.PlexServer`
Sometimes there is more than one address specified for a server or client.
After trying to connect to all available addresses for this client and assuming
at least one connection was successful, the PlexClient object is built and returned.
Raises:
:class:`~plexapi.exceptions.NotFound`: When unable to connect to any addresses for this device.
"""
cls = PlexServer if 'server' in self.provides else PlexClient
listargs = [[cls, url, self.token, timeout] for url in self.connections]
log.info('Testing %s device connections..', len(listargs))
results = utils.threaded(_connect, listargs)
return _chooseConnection('Device', self.name, results)
def _chooseConnection(ctype, name, results):
""" Chooses the first (best) connection from the given _connect results. """
# At this point we have a list of result tuples containing (url, token, PlexServer, runtime)
# or (url, token, None, runtime) in the case a connection could not be established.
for url, token, result, runtime in results:
okerr = 'OK' if result else 'ERR'
log.info('%s connection %s (%ss): %s?X-Plex-Token=%s', ctype, okerr, runtime, url, token)
results = [r[2] for r in results if r and r[2] is not None]
if results:
log.info('Connecting to %s: %s?X-Plex-Token=%s', ctype, results[0]._baseurl, results[0]._token)
return results[0]
raise NotFound('Unable to connect to %s: %s' % (ctype.lower(), name))
def _chooseConnection(ctype, name, results):
""" Chooses the first (best) connection from the given _connect results. """
# At this point we have a list of result tuples containing (url, token, PlexServer, runtime)
# or (url, token, None, runtime) in the case a connection could not be established.
for url, token, result, runtime in results:
okerr = 'OK' if result else 'ERR'
log.info('%s connection %s (%ss): %s?X-Plex-Token=%s', ctype, okerr, runtime, url, token)
results = [r[2] for r in results if r and r[2] is not None]
if results:
log.info('Connecting to %s: %s?X-Plex-Token=%s', ctype, results[0]._baseurl, results[0]._token)
return results[0]
raise NotFound('Unable to connect to %s: %s' % (ctype.lower(), name))
def setWebhooks(self, urls):
log.info('Setting webhooks: %s' % urls)
data = {'urls[]': urls} if len(urls) else {'urls': ''}
data = self.query(self.WEBHOOKS, self._session.post, data=data)
self._webhooks = self.listAttrs(data, 'url', etag='webhook')
return self._webhooks
def connect(self, timeout=None):
""" Returns a new :class:`~plexapi.client.PlexClient` or :class:`~plexapi.server.PlexServer`
Sometimes there is more than one address specified for a server or client.
After trying to connect to all available addresses for this client and assuming
at least one connection was successful, the PlexClient object is built and returned.
Raises:
:class:`~plexapi.exceptions.NotFound`: When unable to connect to any addresses for this device.
"""
cls = PlexServer if 'server' in self.provides else PlexClient
listargs = [[cls, url, self.token, timeout] for url in self.connections]
log.info('Testing %s device connections..', len(listargs))
results = utils.threaded(_connect, listargs)
_chooseConnection('Device', self.name, results)
"""
# Sort connections from (https, local) to (http, remote)
# Only check non-local connections unless we own the resource
connections = sorted(self.connections, key=lambda c: c.local, reverse=True)
owned_or_unowned_non_local = lambda x: self.owned or (not self.owned and not x.local)
https = [c.uri for c in connections if owned_or_unowned_non_local(c)]
http = [c.httpuri for c in connections if owned_or_unowned_non_local(c)]
cls = PlexServer if 'server' in self.provides else PlexClient
# Force ssl, no ssl, or any (default)
if ssl is True: connections = https
elif ssl is False: connections = http
else: connections = https + http
# Try connecting to all known resource connections in parellel, but
# only return the first server (in order) that provides a response.
listargs = [[cls, url, self.accessToken, timeout] for url in connections]
log.info('Testing %s resource connections..', len(listargs))
results = utils.threaded(_connect, listargs)
return _chooseConnection('Resource', self.name, results)
def _chooseConnection(ctype, name, results):
""" Chooses the first (best) connection from the given _connect results. """
# At this point we have a list of result tuples containing (url, token, PlexServer, runtime)
# or (url, token, None, runtime) in the case a connection could not be established.
for url, token, result, runtime in results:
okerr = 'OK' if result else 'ERR'
log.info('%s connection %s (%ss): %s?X-Plex-Token=%s', ctype, okerr, runtime, url, token)
results = [r[2] for r in results if r and r[2] is not None]
if results:
log.info('Connecting to %s: %s?X-Plex-Token=%s', ctype, results[0]._baseurl, results[0]._token)
return results[0]
raise NotFound('Unable to connect to %s: %s' % (ctype.lower(), name))