369 lines
13 KiB
Python
369 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright(C) 2010 Romain Bignon, Christophe Benz
|
|
#
|
|
# 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, version 3 of the License.
|
|
#
|
|
# 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, write to the Free Software
|
|
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
|
|
|
|
|
import sys, os
|
|
import logging
|
|
import optparse
|
|
from optparse import OptionGroup, OptionParser
|
|
|
|
from weboob.capabilities.base import NotAvailable, NotLoaded
|
|
from weboob.core import Weboob, CallErrors
|
|
from weboob.tools.config.iconfig import ConfigError
|
|
from weboob.tools.backend import ObjectNotAvailable
|
|
from weboob.tools.log import createColoredFormatter, getLogger
|
|
|
|
|
|
__all__ = ['BackendNotFound', 'BaseApplication']
|
|
|
|
|
|
class BackendNotFound(Exception):
|
|
pass
|
|
|
|
|
|
class ApplicationStorage(object):
|
|
def __init__(self, name, storage):
|
|
self.name = name
|
|
self.storage = storage
|
|
|
|
def set(self, *args):
|
|
if self.storage:
|
|
return self.storage.set('applications', self.name, *args)
|
|
|
|
def delete(self, *args):
|
|
if self.storage:
|
|
return self.storage.delete('applications', self.name, *args)
|
|
|
|
def get(self, *args, **kwargs):
|
|
if self.storage:
|
|
return self.storage.get('applications', self.name, *args, **kwargs)
|
|
else:
|
|
return kwargs.get('default', None)
|
|
|
|
def load(self, default):
|
|
if self.storage:
|
|
return self.storage.load('applications', self.name, default)
|
|
|
|
def save(self):
|
|
if self.storage:
|
|
return self.storage.save('applications', self.name)
|
|
|
|
class BaseApplication(object):
|
|
"""
|
|
Base application.
|
|
|
|
This class can be herited to have some common code within weboob
|
|
applications.
|
|
"""
|
|
|
|
# ------ Class attributes --------------------------------------
|
|
|
|
# Application name
|
|
APPNAME = ''
|
|
# Configuration and work directory (default: ~/.weboob/)
|
|
CONFDIR = os.path.join(os.path.expanduser('~'), '.weboob')
|
|
# Default configuration dict (can only contain key/values)
|
|
CONFIG = {}
|
|
# Default storage tree
|
|
STORAGE = {}
|
|
# Synopsis
|
|
SYNOPSIS = 'Usage: %prog [-h] [-dqv] [-b backends] ...\n'
|
|
SYNOPSIS += ' %prog [--help] [--version]'
|
|
# Description
|
|
DESCRIPTION = None
|
|
# Version
|
|
VERSION = None
|
|
# Copyright
|
|
COPYRIGHT = None
|
|
|
|
# ------ Abstract methods --------------------------------------
|
|
def create_weboob(self):
|
|
return Weboob()
|
|
|
|
def _get_completions(self):
|
|
"""
|
|
Overload this method in subclasses if you want to enrich shell completion.
|
|
@return a set object
|
|
"""
|
|
return set()
|
|
|
|
def _handle_options(self):
|
|
"""
|
|
Overload this method in application type subclass
|
|
if you want to handle options defined in subclass constructor.
|
|
"""
|
|
pass
|
|
|
|
def add_application_options(self, group):
|
|
"""
|
|
Overload this method if your application needs extra options.
|
|
|
|
These options will be displayed in an option group.
|
|
"""
|
|
pass
|
|
|
|
def handle_application_options(self):
|
|
"""
|
|
Overload this method in your application if you want to handle options defined in add_application_options.
|
|
"""
|
|
pass
|
|
|
|
# ------ BaseApplication methods -------------------------------
|
|
|
|
def __init__(self, option_parser=None):
|
|
self.logger = getLogger(self.APPNAME)
|
|
self.weboob = self.create_weboob()
|
|
self.config = None
|
|
self.options = None
|
|
if option_parser is None:
|
|
self._parser = OptionParser(self.SYNOPSIS, version=self._get_optparse_version())
|
|
else:
|
|
self._parser = option_parser
|
|
if self.DESCRIPTION:
|
|
self._parser.description = self.DESCRIPTION
|
|
app_options = OptionGroup(self._parser, '%s Options' % self.APPNAME.capitalize())
|
|
self.add_application_options(app_options)
|
|
if len(app_options.option_list) > 0:
|
|
self._parser.add_option_group(app_options)
|
|
self._parser.add_option('-b', '--backends', help='what backend(s) to enable (comma separated)')
|
|
logging_options = OptionGroup(self._parser, 'Logging Options')
|
|
logging_options.add_option('-d', '--debug', action='store_true', help='display debug messages')
|
|
logging_options.add_option('-q', '--quiet', action='store_true', help='display only error messages')
|
|
logging_options.add_option('-v', '--verbose', action='store_true', help='display info messages')
|
|
logging_options.add_option('--logging-file', action='store', type='string', dest='logging_file', help='file to save logs')
|
|
logging_options.add_option('-a', '--save-responses', action='store_true', help='save every response')
|
|
self._parser.add_option_group(logging_options)
|
|
self._parser.add_option('--shell-completion', action='store_true', help=optparse.SUPPRESS_HELP)
|
|
|
|
def deinit(self):
|
|
self.weboob.want_stop()
|
|
self.weboob.deinit()
|
|
|
|
def create_storage(self, path=None, klass=None, localonly=False):
|
|
"""
|
|
Create a storage object.
|
|
|
|
@param path [str] an optional specific path.
|
|
@param klass [IStorage] what klass to instance.
|
|
@param localonly [bool] if True, do not set it on the Weboob object.
|
|
@return a IStorage object
|
|
"""
|
|
if klass is None:
|
|
from weboob.tools.storage import StandardStorage
|
|
klass = StandardStorage
|
|
|
|
if path is None:
|
|
path = os.path.join(self.CONFDIR, self.APPNAME + '.storage')
|
|
elif not path.startswith('/'):
|
|
path = os.path.join(self.CONFDIR, path)
|
|
|
|
storage = klass(path)
|
|
self.storage = ApplicationStorage(self.APPNAME, storage)
|
|
self.storage.load(self.STORAGE)
|
|
|
|
if not localonly:
|
|
self.weboob.storage = storage
|
|
|
|
return storage
|
|
|
|
def load_config(self, path=None, klass=None):
|
|
"""
|
|
Load a configuration file and get his object.
|
|
|
|
@param path [str] an optional specific path.
|
|
@param klass [IConfig] what klass to instance.
|
|
@return a IConfig object
|
|
"""
|
|
if klass is None:
|
|
from weboob.tools.config.iniconfig import INIConfig
|
|
klass = INIConfig
|
|
|
|
if path is None:
|
|
path = os.path.join(self.CONFDIR, self.APPNAME)
|
|
elif not path.startswith('/'):
|
|
path = os.path.join(self.CONFDIR, path)
|
|
|
|
self.config = klass(path)
|
|
self.config.load(self.CONFIG)
|
|
|
|
def main(self, argv):
|
|
"""
|
|
Main method
|
|
|
|
Called by run
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
def load_backends(self, caps=None, names=None, *args, **kwargs):
|
|
if names is None and self.options.backends:
|
|
names = self.options.backends.split(',')
|
|
loaded = self.weboob.load_backends(caps, names, *args, **kwargs)
|
|
if not loaded:
|
|
logging.warning(u'No backend loaded')
|
|
return loaded
|
|
|
|
def _get_optparse_version(self):
|
|
version = None
|
|
if self.VERSION:
|
|
if self.COPYRIGHT:
|
|
version = '%s v%s %s' % (self.APPNAME, self.VERSION, self.COPYRIGHT)
|
|
else:
|
|
version = '%s v%s' % (self.APPNAME, self.VERSION)
|
|
return version
|
|
|
|
def _do_complete_obj(self, backend, fields, obj):
|
|
if fields:
|
|
try:
|
|
backend.fillobj(obj, fields)
|
|
except ObjectNotAvailable, e:
|
|
logging.warning(u'Could not retrieve required fields (%s): %s' % (','.join(fields), e))
|
|
for field in fields:
|
|
if getattr(obj, field) is NotLoaded:
|
|
setattr(obj, field, NotAvailable)
|
|
return obj
|
|
|
|
def _do_complete_iter(self, backend, count, fields, res):
|
|
for i, sub in enumerate(res):
|
|
if count and i == count:
|
|
break
|
|
sub = self._do_complete_obj(backend, fields, sub)
|
|
yield sub
|
|
|
|
def _do_complete(self, backend, count, selected_fields, function, *args, **kwargs):
|
|
assert count is None or count > 0
|
|
if callable(function):
|
|
res = function(backend, *args, **kwargs)
|
|
else:
|
|
res = getattr(backend, function)(*args, **kwargs)
|
|
|
|
if hasattr(res, '__iter__'):
|
|
return self._do_complete_iter(backend, count, selected_fields, res)
|
|
else:
|
|
return self._do_complete_obj(backend, selected_fields, res)
|
|
|
|
def bcall_error_handler(self, backend, error, backtrace):
|
|
"""
|
|
Handler for an exception inside the CallErrors exception.
|
|
|
|
This method can be overrided to support more exceptions types.
|
|
"""
|
|
print >>sys.stderr, u'Error(%s): %s' % (backend.name, error)
|
|
if logging.root.level == logging.DEBUG:
|
|
print >>sys.stderr, backtrace
|
|
|
|
def bcall_errors_handler(self, errors):
|
|
"""
|
|
Handler for the CallErrors exception.
|
|
"""
|
|
for backend, error, backtrace in errors.errors:
|
|
self.bcall_error_handler(backend, error, backtrace)
|
|
if logging.root.level != logging.DEBUG:
|
|
print >>sys.stderr, 'Use --debug option to print backtraces.'
|
|
|
|
def parse_args(self, args):
|
|
self.options, args = self._parser.parse_args(args)
|
|
|
|
if self.options.shell_completion:
|
|
items = set()
|
|
for option in self._parser.option_list:
|
|
if not option.help is optparse.SUPPRESS_HELP:
|
|
items.update(str(option).split('/'))
|
|
items.update(self._get_completions())
|
|
print ' '.join(items)
|
|
sys.exit(0)
|
|
|
|
if self.options.save_responses:
|
|
from weboob.tools.browser import BaseBrowser
|
|
BaseBrowser.SAVE_RESPONSES = True
|
|
|
|
for handler in logging.root.handlers:
|
|
logging.root.removeHandler(handler)
|
|
|
|
if self.options.debug:
|
|
level = logging.DEBUG
|
|
elif self.options.verbose:
|
|
level = logging.INFO
|
|
elif self.options.quiet:
|
|
level = logging.ERROR
|
|
else:
|
|
level = logging.WARNING
|
|
|
|
logging.root.setLevel(level)
|
|
|
|
# file logger
|
|
if self.options.logging_file:
|
|
try:
|
|
stream = open(self.options.logging_file, 'w')
|
|
except IOError, e:
|
|
self.logger.error('Unable to create the logging file: %s' % e)
|
|
sys.exit(1)
|
|
else:
|
|
format = '%(asctime)s:%(levelname)s:%(name)s:%(pathname)s:%(lineno)d:%(funcName)s %(message)s'
|
|
handler = logging.StreamHandler(stream)
|
|
handler.setLevel(level)
|
|
handler.setFormatter(logging.Formatter(format))
|
|
logging.root.addHandler(handler)
|
|
else:
|
|
# stdout logger
|
|
format = '%(asctime)s:%(levelname)s:%(name)s:%(filename)s:%(lineno)d:%(funcName)s %(message)s'
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setFormatter(createColoredFormatter(sys.stdout, format))
|
|
handler.setLevel(level)
|
|
logging.root.addHandler(handler)
|
|
|
|
self._handle_options()
|
|
self.handle_application_options()
|
|
|
|
return args
|
|
|
|
@classmethod
|
|
def run(klass, args=None):
|
|
"""
|
|
This static method can be called to run the application.
|
|
|
|
It creates the application object, handles options, setups logging, calls
|
|
the main() method, and catches common exceptions.
|
|
|
|
You can't do anything after this call, as it *always* finishes with
|
|
a call to sys.exit().
|
|
|
|
For example:
|
|
>>> from weboob.application.myapplication import MyApplication
|
|
>>> MyApplication.run()
|
|
"""
|
|
|
|
if args is None:
|
|
args = [(sys.stdin.encoding and arg.decode(sys.stdin.encoding) or arg) for arg in sys.argv]
|
|
app = klass()
|
|
|
|
try:
|
|
try:
|
|
args = app.parse_args(args)
|
|
sys.exit(app.main(args))
|
|
except KeyboardInterrupt:
|
|
print 'Program killed by SIGINT'
|
|
sys.exit(0)
|
|
except EOFError:
|
|
sys.exit(0)
|
|
except ConfigError, e:
|
|
print 'Configuration error: %s' % e
|
|
sys.exit(1)
|
|
except CallErrors, e:
|
|
app.bcall_errors_handler(e)
|
|
sys.exit(1)
|
|
finally:
|
|
app.deinit()
|