#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
This program 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.
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 General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, see .
@author: RaNaN
"""
from Queue import Queue
from threading import Thread
from os import listdir, stat
from os.path import join
from time import sleep, time, strftime, gmtime
from traceback import print_exc, format_exc
from pprint import pformat
from sys import exc_info, exc_clear
from copy import copy
from types import MethodType
from pycurl import error
from PyFile import PyFile
from plugins.Plugin import Abort, Fail, Reconnect, Retry, SkipDownload
from common.packagetools import parseNames
from utils import save_join
from Api import OnlineStatus
class PluginThread(Thread):
"""abstract base class for thread types"""
#----------------------------------------------------------------------
def __init__(self, manager):
"""Constructor"""
Thread.__init__(self)
self.setDaemon(True)
self.m = manager #thread manager
def writeDebugReport(self, pyfile):
""" writes a
:return:
"""
dump_name = "debug_%s_%s.zip" % (pyfile.pluginname, strftime("%d-%m-%Y_%H-%M-%S"))
dump = self.getDebugDump(pyfile)
try:
import zipfile
zip = zipfile.ZipFile(dump_name, "w")
for f in listdir(join("tmp", pyfile.pluginname)):
try:
# avoid encoding errors
zip.write(join("tmp", pyfile.pluginname, f), save_join(pyfile.pluginname, f))
except:
pass
info = zipfile.ZipInfo(save_join(pyfile.pluginname, "debug_Report.txt"), gmtime())
info.external_attr = 0644 << 16L # change permissions
zip.writestr(info, dump)
zip.close()
if not stat(dump_name).st_size:
raise Exception("Empty Zipfile")
except Exception, e:
self.m.log.debug("Error creating zip file: %s" % e)
dump_name = dump_name.replace(".zip", ".txt")
f = open(dump_name, "wb")
f.write(dump)
f.close()
self.m.core.log.info("Debug Report written to %s" % dump_name)
def getDebugDump(self, pyfile):
dump = "pyLoad %s Debug Report of %s %s \n\nTRACEBACK:\n %s \n\nFRAMESTACK:\n" % (
self.m.core.api.getServerVersion(), pyfile.pluginname, pyfile.plugin.__version__, format_exc())
tb = exc_info()[2]
stack = []
while tb:
stack.append(tb.tb_frame)
tb = tb.tb_next
for frame in stack[1:]:
dump += "\nFrame %s in %s at line %s\n" % (frame.f_code.co_name,
frame.f_code.co_filename,
frame.f_lineno)
for key, value in frame.f_locals.items():
dump += "\t%20s = " % key
try:
dump += pformat(value) + "\n"
except Exception, e:
dump += " " + str(e) + "\n"
del frame
del stack #delete it just to be sure...
dump += "\n\nPLUGIN OBJECT DUMP: \n\n"
for name in dir(pyfile.plugin):
attr = getattr(pyfile.plugin, name)
if not name.endswith("__") and type(attr) != MethodType:
dump += "\t%20s = " % name
try:
dump += pformat(attr) + "\n"
except Exception, e:
dump += " " + str(e) + "\n"
dump += "\nPYFILE OBJECT DUMP: \n\n"
for name in dir(pyfile):
attr = getattr(pyfile, name)
if not name.endswith("__") and type(attr) != MethodType:
dump += "\t%20s = " % name
try:
dump += pformat(attr) + "\n"
except Exception, e:
dump += " " + str(e) + "\n"
if pyfile.pluginname in self.m.core.config.plugin:
dump += "\n\nCONFIG: \n\n"
dump += pformat(self.m.core.config.plugin[pyfile.pluginname]) + "\n"
return dump
def clean(self, pyfile):
""" set thread unactive and release pyfile """
self.active = False
pyfile.release()
class DownloadThread(PluginThread):
"""thread for downloading files from 'real' hoster plugins"""
#----------------------------------------------------------------------
def __init__(self, manager):
"""Constructor"""
PluginThread.__init__(self, manager)
self.queue = Queue() # job queue
self.active = False
self.start()
#----------------------------------------------------------------------
def run(self):
"""run method"""
pyfile = None
while True:
del pyfile
self.active = self.queue.get()
pyfile = self.active
if self.active == "quit":
self.active = False
self.m.threads.remove(self)
return True
try:
if not pyfile.hasPlugin(): continue
#this pyfile was deleted while queueing
pyfile.plugin.checkForSameFiles(starting=True)
self.m.log.info(_("Download starts: %s" % pyfile.name))
# start download
self.m.core.hookManager.downloadPreparing(pyfile)
pyfile.plugin.preprocessing(self)
self.m.log.info(_("Download finished: %s") % pyfile.name)
self.m.core.hookManager.downloadFinished(pyfile)
self.m.core.files.checkPackageFinished(pyfile)
except NotImplementedError:
self.m.log.error(_("Plugin %s is missing a function.") % pyfile.pluginname)
pyfile.setStatus("failed")
pyfile.error = "Plugin does not work"
self.clean(pyfile)
continue
except Abort:
try:
self.m.log.info(_("Download aborted: %s") % pyfile.name)
except:
pass
pyfile.setStatus("aborted")
self.clean(pyfile)
continue
except Reconnect:
self.queue.put(pyfile)
#pyfile.req.clearCookies()
while self.m.reconnecting.isSet():
sleep(0.5)
continue
except Retry, e:
reason = e.args[0]
self.m.log.info(_("Download restarted: %(name)s | %(msg)s") % {"name": pyfile.name, "msg": reason})
self.queue.put(pyfile)
continue
except Fail, e:
msg = e.args[0]
if msg == "offline":
pyfile.setStatus("offline")
self.m.log.warning(_("Download is offline: %s") % pyfile.name)
elif msg == "temp. offline":
pyfile.setStatus("temp. offline")
self.m.log.warning(_("Download is temporary offline: %s") % pyfile.name)
else:
pyfile.setStatus("failed")
self.m.log.warning(_("Download failed: %(name)s | %(msg)s") % {"name": pyfile.name, "msg": msg})
pyfile.error = msg
self.m.core.hookManager.downloadFailed(pyfile)
self.clean(pyfile)
continue
except error, e:
if len(e.args) == 2:
code, msg = e.args
else:
code = 0
msg = e.args
self.m.log.debug("pycurl exception %s: %s" % (code, msg))
if code in (7, 18, 28, 52, 56):
self.m.log.warning(_("Couldn't connect to host or connection reset, waiting 1 minute and retry."))
wait = time() + 60
pyfile.waitUntil = wait
pyfile.setStatus("waiting")
while time() < wait:
sleep(1)
if pyfile.abort:
break
if pyfile.abort:
self.m.log.info(_("Download aborted: %s") % pyfile.name)
pyfile.setStatus("aborted")
self.clean(pyfile)
else:
self.queue.put(pyfile)
continue
else:
pyfile.setStatus("failed")
self.m.log.error("pycurl error %s: %s" % (code, msg))
if self.m.core.debug:
print_exc()
self.writeDebugReport(pyfile)
self.m.core.hookManager.downloadFailed(pyfile)
self.clean(pyfile)
continue
except SkipDownload, e:
pyfile.setStatus("skipped")
self.m.log.info(
_("Download skipped: %(name)s due to %(plugin)s") % {"name": pyfile.name, "plugin": e.message})
self.clean(pyfile)
self.m.core.files.checkPackageFinished(pyfile)
self.active = False
self.m.core.files.save()
continue
except Exception, e:
pyfile.setStatus("failed")
self.m.log.warning(_("Download failed: %(name)s | %(msg)s") % {"name": pyfile.name, "msg": str(e)})
pyfile.error = str(e)
if self.m.core.debug:
print_exc()
self.writeDebugReport(pyfile)
self.m.core.hookManager.downloadFailed(pyfile)
self.clean(pyfile)
continue
finally:
self.m.core.files.save()
pyfile.checkIfProcessed()
exc_clear()
#pyfile.plugin.req.clean()
self.active = False
pyfile.finishIfDone()
self.m.core.files.save()
def put(self, job):
"""assing job to thread"""
self.queue.put(job)
def stop(self):
"""stops the thread"""
self.put("quit")
class DecrypterThread(PluginThread):
"""thread for decrypting"""
def __init__(self, manager, pyfile):
"""constructor"""
PluginThread.__init__(self, manager)
self.active = pyfile
manager.localThreads.append(self)
pyfile.setStatus("decrypting")
self.start()
def getActiveFiles(self):
return [self.active]
def run(self):
"""run method"""
pyfile = self.active
retry = False
try:
self.m.log.info(_("Decrypting starts: %s") % self.active.name)
self.active.plugin.preprocessing(self)
except NotImplementedError:
self.m.log.error(_("Plugin %s is missing a function.") % self.active.pluginname)
return
except Fail, e:
msg = e.args[0]
if msg == "offline":
self.active.setStatus("offline")
self.m.log.warning(_("Download is offline: %s") % self.active.name)
else:
self.active.setStatus("failed")
self.m.log.error(_("Decrypting failed: %(name)s | %(msg)s") % {"name": self.active.name, "msg": msg})
self.active.error = msg
return
except Abort:
self.m.log.info(_("Download aborted: %s") % pyfile.name)
pyfile.setStatus("aborted")
return
except Retry:
self.m.log.info(_("Retrying %s") % self.active.name)
retry = True
return self.run()
except Exception, e:
self.active.setStatus("failed")
self.m.log.error(_("Decrypting failed: %(name)s | %(msg)s") % {"name": self.active.name, "msg": str(e)})
self.active.error = str(e)
if self.m.core.debug:
print_exc()
self.writeDebugReport(pyfile)
return
finally:
if not retry:
self.active.release()
self.active = False
self.m.core.files.save()
self.m.localThreads.remove(self)
exc_clear()
#self.m.core.hookManager.downloadFinished(pyfile)
#self.m.localThreads.remove(self)
#self.active.finishIfDone()
if not retry:
pyfile.delete()
class HookThread(PluginThread):
"""thread for hooks"""
#----------------------------------------------------------------------
def __init__(self, m, function, args, kwargs):
"""Constructor"""
PluginThread.__init__(self, m)
self.f = function
self.args = args
self.kwargs = kwargs
self.active = []
m.localThreads.append(self)
self.start()
def getActiveFiles(self):
return self.active
def addActive(self, pyfile):
""" Adds a pyfile to active list and thus will be displayed on overview"""
if pyfile not in self.active:
self.active.append(pyfile)
def finishFile(self, pyfile):
if pyfile in self.active:
self.active.remove(pyfile)
pyfile.finishIfDone()
def run(self):
try:
try:
self.kwargs["thread"] = self
self.f(*self.args, **self.kwargs)
except TypeError, e:
#dirty method to filter out exceptions
if "unexpected keyword argument 'thread'" not in e.args[0]:
raise
del self.kwargs["thread"]
self.f(*self.args, **self.kwargs)
finally:
local = copy(self.active)
for x in local:
self.finishFile(x)
self.m.localThreads.remove(self)
class InfoThread(PluginThread):
def __init__(self, manager, data, pid=-1, rid=-1, add=False):
"""Constructor"""
PluginThread.__init__(self, manager)
self.data = data
self.pid = pid # package id
# [ .. (name, plugin) .. ]
self.rid = rid #result id
self.add = add #add packages instead of return result
self.cache = [] #accumulated data
self.start()
def run(self):
"""run method"""
plugins = {}
container = []
for url, plugin in self.data:
if plugin in plugins:
plugins[plugin].append(url)
else:
plugins[plugin] = [url]
# filter out container plugins
for name in self.m.core.pluginManager.containerPlugins:
if name in plugins:
container.extend([(name, url) for url in plugins[name]])
del plugins[name]
#directly write to database
if self.pid > -1:
for pluginname, urls in plugins.iteritems():
plugin = self.m.core.pluginManager.getPlugin(pluginname, True)
if hasattr(plugin, "getInfo"):
self.fetchForPlugin(pluginname, plugin, urls, self.updateDB)
self.m.core.files.save()
elif self.add:
for pluginname, urls in plugins.iteritems():
plugin = self.m.core.pluginManager.getPlugin(pluginname, True)
if hasattr(plugin, "getInfo"):
self.fetchForPlugin(pluginname, plugin, urls, self.updateCache, True)
else:
#generate default result
result = [(url, 0, 3, url) for url in urls]
self.updateCache(pluginname, result)
packs = parseNames([(name, url) for name, x, y, url in self.cache])
self.m.log.debug("Fetched and generated %d packages" % len(packs))
for k, v in packs:
self.m.core.api.addPackage(k, v)
#empty cache
del self.cache[:]
else: #post the results
for name, url in container:
#attach container content
try:
data = self.decryptContainer(name, url)
except:
print_exc()
self.m.log.error("Could not decrypt container.")
data = []
for url, plugin in data:
if plugin in plugins:
plugins[plugin].append(url)
else:
plugins[plugin] = [url]
self.m.infoResults[self.rid] = {}
for pluginname, urls in plugins.iteritems():
plugin = self.m.core.pluginManager.getPlugin(pluginname, True)
if hasattr(plugin, "getInfo"):
self.fetchForPlugin(pluginname, plugin, urls, self.updateResult, True)
#force to process cache
if self.cache:
self.updateResult(pluginname, [], True)
else:
#generate default result
result = [(url, 0, 3, url) for url in urls]
self.updateResult(pluginname, result, True)
self.m.infoResults[self.rid]["ALL_INFO_FETCHED"] = {}
self.m.timestamp = time() + 5 * 60
def updateDB(self, plugin, result):
self.m.core.files.updateFileInfo(result, self.pid)
def updateResult(self, plugin, result, force=False):
#parse package name and generate result
#accumulate results
self.cache.extend(result)
if len(self.cache) >= 20 or force:
#used for package generating
tmp = [(name, (url, OnlineStatus(name, plugin, "unknown", status, int(size))))
for name, size, status, url in self.cache]
data = parseNames(tmp)
result = {}
for k, v in data.iteritems():
for url, status in v:
status.packagename = k
result[url] = status
self.m.setInfoResults(self.rid, result)
self.cache = []
def updateCache(self, plugin, result):
self.cache.extend(result)
def fetchForPlugin(self, pluginname, plugin, urls, cb, err=None):
try:
result = [] #result loaded from cache
process = [] #urls to process
for url in urls:
if url in self.m.infoCache:
result.append(self.m.infoCache[url])
else:
process.append(url)
if result:
self.m.log.debug("Fetched %d values from cache for %s" % (len(result), pluginname))
cb(pluginname, result)
if process:
self.m.log.debug("Run Info Fetching for %s" % pluginname)
for result in plugin.getInfo(process):
#result = [ .. (name, size, status, url) .. ]
if not type(result) == list: result = [result]
for res in result:
self.m.infoCache[res[3]] = res
cb(pluginname, result)
self.m.log.debug("Finished Info Fetching for %s" % pluginname)
except Exception, e:
self.m.log.warning(_("Info Fetching for %(name)s failed | %(err)s") %
{"name": pluginname, "err": str(e)})
if self.m.core.debug:
print_exc()
# generate default results
if err:
result = [(url, 0, 3, url) for url in urls]
cb(pluginname, result)
def decryptContainer(self, plugin, url):
data = []
# only works on container plugins
self.m.log.debug("Pre decrypting %s with %s" % (url, plugin))
# dummy pyfile
pyfile = PyFile(self.m.core.files, -1, url, url, 0, 0, "", plugin, -1, -1)
pyfile.initPlugin()
# little plugin lifecycle
try:
pyfile.plugin.setup()
pyfile.plugin.loadToDisk()
pyfile.plugin.decrypt(pyfile)
pyfile.plugin.deleteTmp()
for pack in pyfile.plugin.packages:
pyfile.plugin.urls.extend(pack[1])
data = self.m.core.pluginManager.parseUrls(pyfile.plugin.urls)
self.m.log.debug("Got %d links." % len(data))
except Exception, e:
self.m.log.debug("Pre decrypting error: %s" % str(e))
finally:
pyfile.release()
return data