def main(filename):
    weboob = Weboob()
    try:
        hds = weboob.build_backend('hds')
    except ModuleLoadError, e:
        print >>sys.stderr, 'Unable to load "hds" module: %s' % e
        return 1

class BackendTest(TestCase):
    MODULE = None

    def __init__(self, *args, **kwargs):
        super(BackendTest, self).__init__(*args, **kwargs)

        self.backends = {}
        self.backend_instance = None
        self.backend = None
        self.weboob = Weboob()

        # Skip tests when passwords are missing
        self.weboob.requests.register('login', self.login_cb)

        if self.weboob.load_backends(modules=[self.MODULE]):
            # provide the tests with all available backends
            self.backends = self.weboob.backend_instances

    def login_cb(self, backend_name, value):
        raise SkipTest('missing config \'%s\' is required for this test' % value.label)

    def run(self, result):
        """
        Call the parent run() for each backend instance.
        Skip the test if we have no backends.
        """
        # This is a hack to fix an issue with nosetests running
        # with many tests. The default is 1000.
        sys.setrecursionlimit(10000)
        try:
            if not len(self.backends):
                self.backend = self.weboob.build_backend(self.MODULE, nofail=True)
                TestCase.run(self, result)
            else:
                # Run for all backend
                for backend_instance in self.backends.keys():
                    print(backend_instance)
                    self.backend = self.backends[backend_instance]
                    TestCase.run(self, result)
        finally:
            self.weboob.deinit()

    def shortDescription(self):
        """
        Generate a description with the backend instance name.
        """
        # do not use TestCase.shortDescription as it returns None
        return '%s [%s]' % (str(self), self.backend_instance)

    def is_backend_configured(self):
        """
        Check if the backend is in the user configuration file
        """
        return self.weboob.backends_config.backend_exists(self.backend.config.instname)

def get_accounts(bname):
    w = Weboob()

    w.load_backends(names=[bname])
    backend = w.get_backend(bname)

    results = {}
    for account in backend.iter_accounts():
        # a unicode bigger than 8 characters used as key of the table make some bugs in the C++ code
        # convert to string before to use it
        results[str(account.id)] = {'name':    account.label,
                               'balance': int(account.balance * 100),
                               'type':    int(account.type),
                              }
    return results

    def update(self):
        """
        Update Weboob modules.
        """
        self.copy_fakemodules()

        # Weboob has an offending print statement when it "Rebuilds index",
        # which happen at every run if the user has a local repository. We need
        # to silence it, hence the temporary redirect of stdout.
        sys.stdout = open(os.devnull, "w")
        try:
            self.weboob.update(progress=DummyProgress())
        except ConnectionError as exc:
            # Do not delete the repository if there is a connection error.
            raise exc
        except Exception:
            # Try to remove the data directory, to see if it changes a thing.
            # This is especially useful when a new version of Weboob is
            # published and/or the keyring changes.
            shutil.rmtree(self.weboob_data_path)
            os.makedirs(self.weboob_data_path)

            # Recreate the Weboob object as the directories are created
            # on creating the Weboob object.
            self.weboob = Weboob(workdir=self.weboob_data_path,
                                 datadir=self.weboob_data_path)

            # Rewrite sources.list file
            self.write_weboob_sources_list()

            # Retry update
            self.weboob.update(progress=DummyProgress())
        finally:
            # Restore stdout
            sys.stdout = sys.__stdout__

    def __init__(self):
        self.ind = appindicator.Indicator.new(APPINDICATOR_ID,
                                              os.path.abspath(resource_filename('boobank_indicator.data',
                                                                                'indicator-boobank.png')),
                                              appindicator.IndicatorCategory.APPLICATION_STATUS)

        self.menu = Gtk.Menu()
        self.ind.set_menu(self.menu)

        logging.basicConfig()
        if 'weboob_path' in os.environ:
            self.weboob = Weboob(os.environ['weboob_path'])
        else:
            self.weboob = Weboob()

        self.weboob.load_backends(CapBank)

    def __init__(self, *args, **kwargs):
        TestCase.__init__(self, *args, **kwargs)

        self.backend = None
        self.weboob = Weboob()

        if self.weboob.load_backends(modules=[self.BACKEND]):
            self.backend = choice(self.weboob.backend_instances.values())

 def __init__(self, backend_name, download_directory, links_directory):
     self.download_directory = download_directory
     self.links_directory = links_directory
     self.backend_name = backend_name
     self.backend = None
     self.weboob = Weboob()
     self.weboob.load_backends(modules=[self.backend_name])
     self.backend=self.weboob.get_backend(self.backend_name)

def get_transactions(bname, accid, maximum):
    w = Weboob()

    w.load_backends(names=[bname])
    backend = w.get_backend(bname)

    acc = backend.get_account(accid)
    results = {}
    results['id'] = acc.id
    results['name'] = acc.label
    results['balance'] = int(acc.balance * 100)
    results['type'] = int(acc.type)
    results['transactions'] = []

    try:
        count = int(maximum)
        if count < 1:
            count = 0
    except:
        count = 0
    i = 0
    first = True
    rewriteid = False
    seen = set()
    for tr in backend.iter_history(acc):
        if first:
            if tr.id == u'0' or tr.id == u'':
                rewriteid = True
            first = False
        if rewriteid:
            tr.id = tr.unique_id(seen)
        t = {'id':          tr.id,
             'date':        tr.date.strftime('%Y-%m-%d'),
             'rdate':       tr.rdate.strftime('%Y-%m-%d'),
             'type':        int(tr.type),
             'raw':         tr.raw,
             'category':    tr.category,
             'label':       tr.label,
             'amount':      int(tr.amount * 100),
        }
        results['transactions'].append(t)
        i += 1
        if count != 0 and i >= count:
            break

    return results

 def __init__(self, name, backend_name, my_download_directory,
              my_links_directory):
     self.download_directory = my_download_directory
     self.links_directory = my_links_directory
     self.backend_name = backend_name
     self.backend = None
     self.weboob = Weboob()
     self.weboob.load_backends(modules=self.backend_name, )
     self.backend = self.weboob.get_backend(self.backend_name)
     self.name = name

class BackendTest(TestCase):
    MODULE = None

    def __init__(self, *args, **kwargs):
        TestCase.__init__(self, *args, **kwargs)

        self.backends = {}
        self.backend_instance = None
        self.backend = None
        self.weboob = Weboob()

        if self.weboob.load_backends(modules=[self.MODULE]):
            # provide the tests with all available backends
            self.backends = self.weboob.backend_instances
            # chose one backend (enough for most tests)
            self.backend_instance = choice(self.backends.keys())
            self.backend = self.backends[self.backend_instance]

    def run(self, result):
        """
        Call the parent run() for each backend instance.
        Skip the test if we have no backends.
        """
        # This is a hack to fix an issue with nosetests running
        # with many tests. The default is 1000.
        sys.setrecursionlimit(10000)
        try:
            if not len(self.backends):
                result.startTest(self)
                result.stopTest(self)
                raise SkipTest('No backends configured for this module.')
            TestCase.run(self, result)
        finally:
            self.weboob.deinit()

    def shortDescription(self):
        """
        Generate a description with the backend instance name.
        """
        # do not use TestCase.shortDescription as it returns None
        return '%s [%s]' % (str(self), self.backend_instance)

    def __init__(self, *args, **kwargs):
        TestCase.__init__(self, *args, **kwargs)

        self.backends = {}
        self.backend_instance = None
        self.backend = None
        self.weboob = Weboob()

        if self.weboob.load_backends(modules=[self.BACKEND]):
            # provide the tests with all available backends
            self.backends = self.weboob.backend_instances
            # chose one backend (enough for most tests)
            self.backend_instance = choice(self.backends.keys())
            self.backend = self.backends[self.backend_instance]

    def __init__(self, *args, **kwargs):
        super(BackendTest, self).__init__(*args, **kwargs)

        self.backends = {}
        self.backend_instance = None
        self.backend = None
        self.weboob = Weboob()

        # Skip tests when passwords are missing
        self.weboob.requests.register('login', self.login_cb)

        if self.weboob.load_backends(modules=[self.MODULE]):
            # provide the tests with all available backends
            self.backends = self.weboob.backend_instances

class RboorrentDownload(object):
    def __init__(self, _id, no_tracker):
        self.id, self.backend_name = _id.split("@")
        self.no_tracker = no_tracker
        self.weboob = Weboob()
        self.backend = self.weboob.load_backends(modules=[self.backend_name])[self.backend_name]

    def get_magnet(self, torrent):
        if self.no_tracker:
            return "&".join([_ for _ in torrent.magnet.split("&") if not _.startswith("tr=")])
        else:
            return torrent.magnet

    def write_meta(self, torrent):
        dest = "meta-%s-%s.torrent" % (torrent.id, torrent.name)
        magnet = self.get_magnet(torrent)
        buf = "d10:magnet-uri%d:%se" % (len(magnet), magnet)
        try:
            with open(dest, 'w') as f:
                f.write(buf)
        except IOError as e:
            print('Unable to write "%s": %s' % (dest, e.message))

    def write_torrent(self, torrent):
        dest = "%s-%s.torrent" % (torrent.id, torrent.name)
        try:
            buf = self.backend.get_torrent_file(torrent.id)
            if buf:
                try:
                    with open(dest, 'w') as f:
                        f.write(buf)
                except IOError as e:
                    print('Unable to write "%s": %s' % (dest, e))
        except Exception as e:
            print("Could not get torrent file for %[email protected]%s" % (self.id, self.backend_name))

    def run(self):
        try:
            torrent = self.backend.get_torrent(self.id)
            if torrent.magnet:
                self.write_meta(torrent)
            else:
                self.write_torrent(torrent)
        except HTTPNotFound:
            print("Could not find %[email protected]%s" % (self.id, self.backend_name))

class BackendTest(TestCase):
    BACKEND = None

    def __init__(self, *args, **kwargs):
        TestCase.__init__(self, *args, **kwargs)

        self.backend = None
        self.weboob = Weboob()

        if self.weboob.load_backends(modules=[self.BACKEND]):
            self.backend = choice(self.weboob.backend_instances.values())

    def run(self, result):
        if not self.backend:
            result.startTest(self)
            result.stopTest(self)
            raise SkipTest()

        return TestCase.run(self, result)

    def __init__(self, weboob_data_path, fakemodules_path, sources_list_content, is_prod):
        """
        Create a Weboob instance.

        :param weboob_data_path: Weboob path to use.
        :param fakemodules_path: Path to the fake modules directory in user
        data.
        :param sources_list_content: Optional content of the sources.list file,
        as an array of lines, or None if not present.
        :param is_prod: whether we're running in production or not.
        """
        # By default, consider we don't need to update the repositories.
        self.needs_update = False

        self.fakemodules_path = fakemodules_path
        self.sources_list_content = sources_list_content

        if not os.path.isdir(weboob_data_path):
            os.makedirs(weboob_data_path)

        # Set weboob data directory and sources.list file.
        self.weboob_data_path = weboob_data_path
        self.write_weboob_sources_list()

        # Create a Weboob object.
        self.weboob = Weboob(workdir=weboob_data_path,
                             datadir=weboob_data_path)
        self.backend = None
        self.storage = None

        # To make development more pleasant, always copy the fake modules in
        # non-production modes.
        if not is_prod:
            self.copy_fakemodules()

        # Update the weboob repos only if new repos are included.
        if self.needs_update:
            self.update()

class DownloadBoob(object):
    """
        Search and download
        :param name:
        :param backend_name:
        :param my_download_directory:
        :param my_links_directory:
    """

    def __init__(self, name, backend_name, my_download_directory,
                 my_links_directory):
        self.download_directory = my_download_directory
        self.links_directory = my_links_directory
        self.backend_name = backend_name
        self.backend = None
        self.weboob = Weboob()
        self.weboob.load_backends(modules=self.backend_name, )
        self.backend = self.weboob.get_backend(self.backend_name)
        self.name = name

    def get_filename(self, video, relative=False, m3u=False):
        """
            Generate filename for the video
            :param relative:
            :param m3u:
            :param video:
            :rtype : string
        """
        if relative:
            directory = os.path.join("..", DOWNLOAD_DIRECTORY,
                                     self.backend_name)
        else:
            directory = os.path.join(self.download_directory, self.backend_name)
            if not os.path.exists(directory.encode('utf8')):
                os.makedirs(directory.encode('utf8'))
        if not m3u:
            ext = video.ext
            if not ext:
                ext = 'avi'
        else:
            ext = 'm3u'
        return "%s/%s.%s" % (directory, removenonascii(video.id), ext)

    def get_linkname(self, video, m3u=False):
        """
            Generate filename for the link
            :param m3u:
            :param video:
            :rtype : string
        """
        if not os.path.exists(self.links_directory.encode('utf8')):
            os.makedirs(self.links_directory.encode('utf8'))
        if not m3u:
            ext = video.ext
            if not ext:
                ext = 'avi'
        else:
            ext = 'm3u'
        if not kodi:
            misc = video.date.strftime("%y-%m-%d")
            if not misc:
                misc = video.id
            return "%s/%s (%s).%s" % (self.links_directory,
                                      removespecial(video.title),
                                      removespecial(misc), ext)
        else:
            return "%s/%s.%s" % (self.links_directory,
                                 removespecial(video.title), ext)

    def is_downloaded(self, video):
        """
            Check if the video has already been downloaded
            :param video:
            :rtype : bool
        """
        if (os.path.isfile(self.get_filename(video).encode('utf8')) or
            os.path.isfile(self.get_filename(video,
                                             m3u=True).encode('utf8'))):
            logging.info("%s Already downloaded : %s" %
                         (video.id, video.title))
            return True
        logging.debug("%s To be downloaded : %s" %
                      (video.id, video.title))
        return False

    def init_dir(self):
        """
            create directory
        """
        if not os.path.isdir(self.links_directory.encode('utf8')):
            logging.debug("  create link directory : %s" % self.links_directory)
            os.makedirs(self.links_directory.encode('utf8'))
        else:
            logging.debug("  existing link directory : %s" % self.links_directory)
        if kodi:
            file_name = os.path.join(self.links_directory, 'tvshow.nfo')
            show_name = self.links_directory.split("/")[-1]
            if not os.path.isfile(file_name.encode('utf8')):
                logging.debug("  create %s" % file_name)
                f = codecs.open(file_name, "w", "utf-8")
#.........这里部分代码省略.........

class BoobankChecker():
    def __init__(self):
        self.ind = appindicator.Indicator.new(APPINDICATOR_ID,
                                              os.path.abspath(resource_filename('boobank_indicator.data',
                                                                                'indicator-boobank.png')),
                                              appindicator.IndicatorCategory.APPLICATION_STATUS)

        self.menu = Gtk.Menu()
        self.ind.set_menu(self.menu)

        logging.basicConfig()
        if 'weboob_path' in os.environ:
            self.weboob = Weboob(os.environ['weboob_path'])
        else:
            self.weboob = Weboob()

        self.weboob.load_backends(CapBank)

    def clean_menu(self, menu):
        for i in menu.get_children():
            submenu = i.get_submenu()
            if submenu:
                self.clean_menu(i)
            menu.remove(i)

    def check_boobank(self):
        self.ind.set_status(appindicator.IndicatorStatus.ACTIVE)
        self.clean_menu(self.menu)

        total = 0
        currency = ''
        threads = []

        try:
            for account in self.weboob.do('iter_accounts'):

                balance = account.balance
                if account.coming:
                    balance += account.coming

                if account.type != Account.TYPE_LOAN:
                    total += balance
                    image = "green_light.png" if balance > 0 else "red_light.png"
                else:
                    image = "personal-loan.png"

                currency = account.currency_text
                label = "%s: %s%s" % (account.label, balance, account.currency_text)
                account_item = create_image_menu_item(label, image)
                thread = BoobankTransactionsChecker(self.weboob, account_item, account)
                thread.start()
                threads.append(thread)

        except CallErrors as errors:
            self.bcall_errors_handler(errors)

        for thread in threads:
            thread.join()

        for thread in threads:
            self.menu.append(thread.menu)
            thread.menu.show()

        if len(self.menu.get_children()) == 0:
            Notify.Notification.new('<b>Boobank</b>',
                                    'No Bank account found\n Please configure one by running boobank',
                                    'notification-message-im').show()

        sep = Gtk.SeparatorMenuItem()
        self.menu.append(sep)
        sep.show()

        total_item = Gtk.MenuItem("%s: %s%s" % ("Total", total, currency))
        self.menu.append(total_item)
        total_item.show()

        sep = Gtk.SeparatorMenuItem()
        self.menu.append(sep)
        sep.show()

        btnQuit = Gtk.ImageMenuItem()
        image = Gtk.Image()
        image.set_from_stock(Gtk.STOCK_QUIT, Gtk.IconSize.BUTTON)
        btnQuit.set_image(image)
        btnQuit.set_label('Quit')
        btnQuit.set_always_show_image(True)
        btnQuit.connect("activate", self.quit)
        self.menu.append(btnQuit)
        btnQuit.show()

    def quit(self, widget):
        Gtk.main_quit()

    def bcall_errors_handler(self, errors):
        """
        Handler for the CallErrors exception.
        """
        self.ind.set_status(appindicator.IndicatorStatus.ATTENTION)
        for backend, error, backtrace in errors.errors:
            notify = True
#.........这里部分代码省略.........

 def __init__(self, bot):
     Thread.__init__(self)
     self.weboob = Weboob(storage=StandardStorage(STORAGE_FILE))
     self.weboob.load_backends()
     self.bot = bot
     self.bot.set_weboob(self.weboob)

class MyThread(Thread):
    daemon = True

    def __init__(self, bot):
        Thread.__init__(self)
        self.weboob = Weboob(storage=StandardStorage(STORAGE_FILE))
        self.weboob.load_backends()
        self.bot = bot
        self.bot.set_weboob(self.weboob)

    def run(self):
        for ev in self.bot.joined.itervalues():
            ev.wait()

        self.weboob.repeat(300, self.check_board)
        self.weboob.repeat(600, self.check_dlfp)
        self.weboob.repeat(600, self.check_twitter)

        self.weboob.loop()

    def find_keywords(self, text):
        for word in [
            'weboob', 'videoob', 'havesex', 'havedate', 'monboob', 'boobmsg',
            'flatboob', 'boobill', 'pastoob', 'radioob', 'translaboob', 'traveloob', 'handjoob',
            'boobathon', 'boobank', 'boobtracker', 'comparoob', 'wetboobs',
            'webcontentedit', 'weboorrents', 'assnet', 'budget insight', 'budget-insight', 'budgetinsight', 'budgea']:
            if word in text.lower():
                return word
        return None

    def check_twitter(self):
        nb_tweets = 10

        for backend in self.weboob.iter_backends(module='twitter'):
            for thread in list(itertools.islice(backend.iter_resources(None, ['search', 'weboob']),
                                                0,
                                                nb_tweets)):

                if not backend.storage.get('lastpurge'):
                    backend.storage.set('lastpurge', datetime.now() - timedelta(days=60))
                    backend.storage.save()

                if thread.id not in backend.storage.get('seen', default={}) and\
                   thread.date > backend.storage.get('lastpurge'):
                    _item = thread.id.split('#')
                    url = 'https://twitter.com/%s/status/%s' % (_item[0], _item[1])
                    for msg in self.bot.on_url(url):
                        self.bot.send_message('%s: %s' % (_item[0], url))
                        self.bot.send_message(msg)

                    backend.set_message_read(backend.fill_thread(thread, ['root']).root)

    def check_dlfp(self):
        for msg in self.weboob.do('iter_unread_messages', backends=['dlfp']):
            word = self.find_keywords(msg.content)
            if word is not None:
                url = msg.signature[msg.signature.find('https://linuxfr'):]
                self.bot.send_message('[DLFP] %s talks about %s: %s' % (
                    msg.sender, word, url))
            self.weboob[msg.backend].set_message_read(msg)

    def check_board(self):
        def iter_messages(backend):
            with backend.browser:
                return backend.browser.iter_new_board_messages()

        for msg in self.weboob.do(iter_messages, backends=['dlfp']):
            word = self.find_keywords(msg.message)
            if word is not None and msg.login != 'moules':
                message = msg.message.replace(word, '\002%s\002' % word)
                self.bot.send_message('[DLFP] <%s> %s' % (msg.login, message))

    def stop(self):
        self.weboob.want_stop()
        self.weboob.deinit()

class Connector(object):

    """
    Connector is a tool that connects to common websites like bank website,
    phone operator website... and that grabs personal data from there.
    Credentials are required to make this operation.

    Technically, connectors are weboob backend wrappers.
    """

    @staticmethod
    def version():
        """
        Get the version of the installed Weboob.
        """
        return Weboob.VERSION

    def __init__(self, weboob_data_path, fakemodules_path, sources_list_content, is_prod):
        """
        Create a Weboob instance.

        :param weboob_data_path: Weboob path to use.
        :param fakemodules_path: Path to the fake modules directory in user
        data.
        :param sources_list_content: Optional content of the sources.list file,
        as an array of lines, or None if not present.
        :param is_prod: whether we're running in production or not.
        """
        # By default, consider we don't need to update the repositories.
        self.needs_update = False

        self.fakemodules_path = fakemodules_path
        self.sources_list_content = sources_list_content

        if not os.path.isdir(weboob_data_path):
            os.makedirs(weboob_data_path)

        # Set weboob data directory and sources.list file.
        self.weboob_data_path = weboob_data_path
        self.write_weboob_sources_list()

        # Create a Weboob object.
        self.weboob = Weboob(workdir=weboob_data_path,
                             datadir=weboob_data_path)
        self.backend = None
        self.storage = None

        # To make development more pleasant, always copy the fake modules in
        # non-production modes.
        if not is_prod:
            self.copy_fakemodules()

        # Update the weboob repos only if new repos are included.
        if self.needs_update:
            self.update()

    def copy_fakemodules(self):
        """
        Copies the fake modules files into the default fakemodules user-data
        directory.

        When Weboob updates modules, it might want to write within the
        fakemodules directory, which might not be writable by the current
        user. To prevent this, first copy the fakemodules directory in
        a directory we have write access to, and then use that directory
        in the sources list file.
        """
        fakemodules_src = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'fakemodules')
        if os.path.isdir(self.fakemodules_path):
            shutil.rmtree(self.fakemodules_path)
        shutil.copytree(fakemodules_src, self.fakemodules_path)

    def write_weboob_sources_list(self):
        """
        Ensure the Weboob sources.list file contains the required entries from
        Kresus.
        """
        sources_list_path = os.path.join(self.weboob_data_path, 'sources.list')

        # Determine the new content of the sources.list file.
        new_sources_list_content = []
        if self.sources_list_content is not None:
            new_sources_list_content = self.sources_list_content
        else:
            # Default content of the sources.list file.
            new_sources_list_content = [
                unicode('https://updates.weboob.org/%(version)s/main/'),
                unicode('file://%s' % self.fakemodules_path)
            ]

        # Read the content of existing sources.list, if it exists.
        original_sources_list_content = []
        if os.path.isfile(sources_list_path):
            with io.open(sources_list_path, encoding="utf-8") as fh:
                original_sources_list_content = fh.read().splitlines()

        # Update the source.list content and update the repository, only if the
        # content has changed.
        if set(original_sources_list_content) != set(new_sources_list_content):
            with io.open(sources_list_path, 'w', encoding="utf-8") as sources_list_file:
#.........这里部分代码省略.........