nojhan/weboob-devel
Watch
1
0
Fork
You've already forked weboob-devel
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge branch 'master' into maemoui

Browse source
This commit is contained in:
Juke 2010-04-11 18:36:18 +02:00
commit 350deaafbc
17 changed files with 509 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

26
scripts/havesex Executable file
View file

@ -0,0 +1,26 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# vim: ft=python et softtabstop=4 cinoptions=4 shiftwidth=4 ts=4 ai
"""
Copyright(C) 2010 Romain Bignon
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.
"""
from weboob.frontends.havesex import HaveSex
if __name__ == '__main__':
HaveSex.run()

28
weboob/backend.py
View file

@ -20,6 +20,29 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
import re
class BackendStorage(object):
def __init__(self, name, storage):
self.name = name
self.storage = storage
def set(self, *args):
if self.storage:
return self.storage.set(self.name, *args)
def get(self, *args, **kwargs):
if self.storage:
return self.storage.get(self.name, *args, **kwargs)
else:
return kwargs.get('default', None)
def load(self, default):
if self.storage:
return self.storage.load(self.name, default)
def save(self):
if self.storage:
return self.storage.save(self.name)
class Backend(object):
# Module name.
NAME = None
@ -69,9 +92,8 @@ class Backend(object):
elif isinstance(field.default, float):
value = float(value)
self.config[name] = value
self.storage = storage
if self.storage:
self.storage.load(self.name, self.STORAGE)
self.storage = BackendStorage(self.name, storage)
self.storage.load(self.STORAGE)
def has_caps(self, *caps):
for c in caps:

59
weboob/backends/aum/backend.py
View file

@ -20,10 +20,13 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
from weboob.backend import Backend
from weboob.capabilities.messages import ICapMessages, ICapMessagesReply
from weboob.capabilities.dating import ICapDating
from weboob.tools.browser import BrowserUnavailable
from .adopte import AdopteUnMec
from .optim.profiles_walker import ProfilesWalker
class AuMBackend(Backend, ICapMessages, ICapMessagesReply):
class AuMBackend(Backend, ICapMessages, ICapMessagesReply, ICapDating):
NAME = 'aum'
MAINTAINER = 'Romain Bignon'
EMAIL = 'romain@peerfuse.org'
@ -33,7 +36,11 @@ class AuMBackend(Backend, ICapMessages, ICapMessagesReply):
CONFIG = {'username': Backend.ConfigField(description='Username on website'),
'password': Backend.ConfigField(description='Password of account', is_masked=True),
}
STORAGE = {'profiles_walker': {'viewed': []} }
# Private
_browser = None
_profiles_walker = None
def __getattr__(self, name):
if name == 'browser':
@ -60,28 +67,44 @@ class AuMBackend(Backend, ICapMessages, ICapMessagesReply):
yield message
def _iter_messages(self, thread, only_new):
if not only_new or self.browser.nb_new_mails():
my_name = self.browser.get_my_name()
contacts = self.browser.get_contact_list()
contacts.reverse()
try:
if only_new and not self.browser.nb_new_mails():
my_name = self.browser.get_my_name()
contacts = self.browser.get_contact_list()
contacts.reverse()
for contact in contacts:
if only_new and not contact.is_new() or thread and int(thread) != contact.get_id():
continue
for contact in contacts:
if only_new and not contact.is_new() or thread and int(thread) != contact.get_id():
continue
mails = self.browser.get_thread_mails(contact.get_id())
profile = None
for i in xrange(len(mails)):
mail = mails[i]
if only_new and mail.get_from() == my_name:
break
mails = self.browser.get_thread_mails(contact.get_id())
profile = None
for i in xrange(len(mails)):
mail = mails[i]
if only_new and mail.get_from() == my_name:
break
if not profile:
profile = self.browser.get_profile(contact.get_id())
mail.signature += u'\n%s' % profile.get_profile_text()
yield mail
if not profile:
profile = self.browser.get_profile(contact.get_id())
mail.signature += u'\n%s' % profile.get_profile_text()
yield mail
except BrowserUnavailable:
pass
def post_reply(self, thread_id, reply_id, title, message):
for message in self._iter_messages(thread_id, True):
self.queue_messages.append(message)
return self.browser.post(thread_id, message)
def get_profile(self, _id):
try:
return self.browser.get_profile(_id)
except BrowserUnavailable:
return None
def start_profiles_walker(self):
self._profile_walker = ProfilesWalker(self.weboob.scheduler, self.storage, self.browser)
def stop_profiles_walker(self):
self._profiles_walker.stop()
self._profiles_walker = None

0
weboob/backends/aum/optim/__init__.py Normal file
View file

75
weboob/backends/aum/optim/profiles_walker.py Normal file
View file

@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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.
"""
from logging import debug
from random import randint
from weboob.tools.browser import BrowserUnavailable
class ProfilesWalker(object):
def __init__(self, sched, storage, browser):
self.sched = sched
self.storage = storage
self.browser = browser
self.visited_profiles = set(storage.get('profiles_walker', 'viewed'))
self.profiles_queue = set()
self.walk_cron = sched.repeat(60, self.walk)
self.view_cron = sched.schedule(randint(10,40), self.view_profile)
def save(self):
self.storage.set('profiles_walker', 'viewed', self.visited_profiles)
self.storage.save()
def stop(self):
self.event.cancel(self.event)
self.event = None
def walk(self):
self.profiles_queue = self.profiles_queue.union(self.browser.search_profiles()).difference(self.visited_profiles)
self.save()
def view_profile(self):
try:
try:
id = self.profiles_queue.pop()
except KeyError:
return # empty queue
try:
profile = self.browser.get_profile(id)
debug(u'Visited %s (%s)' % (profile.get_name(), id))
# Get score from the aum_score module
# d = self.nucentral_core.callService(context.Context.fromComponent(self), 'aum_score', 'score', profile)
# d.addCallback(self.score_cb, profile.getID())
# deferredlist.append(d)
# do not forget that we visited this profile, to avoid re-visiting it.
self.visited_profiles.add(id)
self.save()
except BrowserUnavailable:
# We consider this profil hasn't been [correctly] analysed
self.profiles_queue.add(id)
return
except Exception, e:
print e
finally:
self.sched.schedule(randint(10,40), self.view_profile)

5
weboob/backends/aum/pages/login.py
View file

@ -50,9 +50,10 @@ class RegisterPage(PageBase):
if isinstance(nickname, unicode):
nickname = nickname.encode('iso-8859-15', 'ignore')
self.browser['pseudo'] = nickname
self.browser['email'] = self.browser.login
self.browser['email'] = self.browser.username
self.browser['pass'] = password
self.browser['sex'] = [str(sex)]
self.browser['sex0'] = [str(sex)]
self.browser['sex'] = str(sex)
self.browser['birthday0'] = [str(birthday_d)]
self.browser['birthday1'] = [str(birthday_m)]
self.browser['birthday2'] = [str(birthday_y)]

34
weboob/backends/aum/pages/profile.py
View file

@ -19,6 +19,8 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""
from weboob.backends.aum.pages.base import PageBase
from weboob.capabilities.dating import Profile
from copy import deepcopy
from logging import warning
import re
@ -124,8 +126,7 @@ class FieldParticularSignes(FieldBase):
elif s.find('rousseur') >= 0:
d['freckle'] = True
class ProfilePage(PageBase):
class ProfilePage(PageBase, Profile):
empty_table = {'details': {'old': 0,
'birthday': (0,0,0),
'zipcode': 0,
@ -229,7 +230,7 @@ class ProfilePage(PageBase):
def __repr__(self):
if isinstance(self.name, unicode):
name = self.name.encode('ascii', 'backslashreplace')
name = self.name.encode('utf-8', 'backslashreplace')
else:
name = self.name
return '<Profile name="%s">' % name
@ -312,7 +313,6 @@ class ProfilePage(PageBase):
self.description = description
def parse_table(self, div):
d = self.table[self.tables[div.getAttribute('id')]]
fields = self.fields[self.tables[div.getAttribute('id')]]
table = div.getElementsByTagName('table')[1]
@ -391,29 +391,3 @@ class ProfilePage(PageBase):
def get_stats(self):
return self.stats
def get_profile_text(self):
body = u'Status: %s' % unicode(self.status)
if self.photos:
body += u'\nPhotos:'
for photo in self.photos:
body += u'\n\t\t%s' % unicode(photo)
body += u'\nStats:'
for label, value in self.get_stats().iteritems():
body += u'\n\t\t%-15s %s' % (label + ':', value)
body += u'\n\nInformations:'
for section, d in self.get_table().iteritems():
body += u'\n\t%s\n' % section
for key, value in d.items():
key = '%s:' % key
if isinstance(value, list):
body += u'\t\t%-15s %s\n' % (key, u', '.join([unicode(s) for s in value]))
elif isinstance(value, float):
body += u'\t\t%-15s %.2f\n' % (key, value)
else:
body += u'\t\t%-15s %s\n' % (key, unicode(value))
body += u'\n\nDescription:\n%s' % unicode(self.get_description())
return body

16
weboob/backends/dlfp/backend.py
View file

@ -63,22 +63,22 @@ class DLFPBackend(Backend, ICapMessages, ICapMessagesReply):
for message in self._iter_messages_of('telegram', thread, only_new):
yield message
def _iter_messages_of(self, what, thread, only_new):
if not what in self.storage.get(self.name, 'seen'):
self.storage.set(self.name, 'seen', what, {})
def _iter_messages_of(self, what, thread_wanted, only_new):
if not what in self.storage.get('seen'):
self.storage.set('seen', what, {})
seen = {}
for article in ArticlesList(what).iter_articles():
if thread and thread != article.id:
if thread_wanted and thread_wanted != article.id:
continue
thread = self.browser.get_content(article.id)
if not article.id in self.storage.get(self.name, 'seen', what):
if not article.id in self.storage.get('seen', what):
seen[article.id] = {'comments': []}
new = True
else:
seen[article.id] = self.storage.get(self.name, 'seen', what, article.id)
seen[article.id] = self.storage.get('seen', what, article.id)
new = False
if not only_new or new:
yield Message(thread.id,
@ -104,8 +104,8 @@ class DLFPBackend(Backend, ICapMessages, ICapMessagesReply):
comment.reply_id,
comment.body,
'Score: %d' % comment.score)
self.storage.set(self.name, 'seen', what, seen)
self.storage.save(self.name)
self.storage.set('seen', what, seen)
self.storage.save()
def post_reply(self, thread_id, reply_id, title, message):
return self.browser.post(thread_id, reply_id, title, message)

53
weboob/capabilities/dating.py Normal file
View file

@ -0,0 +1,53 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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.
"""
from .cap import ICap
class Profile(object):
def get_profile_text(self):
body = u'Status: %s' % unicode(self.status)
if self.photos:
body += u'\nPhotos:'
for photo in self.photos:
body += u'\n\t\t%s' % unicode(photo)
body += u'\nStats:'
for label, value in self.get_stats().iteritems():
body += u'\n\t\t%-15s %s' % (label + ':', value)
body += u'\n\nInformations:'
for section, d in self.get_table().iteritems():
body += u'\n\t%s\n' % section
for key, value in d.items():
key = '%s:' % key
if isinstance(value, list):
body += u'\t\t%-15s %s\n' % (key, u', '.join([unicode(s) for s in value]))
elif isinstance(value, float):
body += u'\t\t%-15s %.2f\n' % (key, value)
else:
body += u'\t\t%-15s %s\n' % (key, unicode(value))
body += u'\n\nDescription:\n%s' % unicode(self.get_description())
return body
class ICapDating(ICap):
def get_profile(self, _id):
raise NotImplementedError()
def start_profile_walker(self):
raise NotImplementedError()

21
weboob/frontends/havesex/__init__.py Normal file
View file

@ -0,0 +1,21 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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.
"""
from .application import HaveSex

63
weboob/frontends/havesex/application.py Normal file
View file

@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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.
"""
from weboob.tools.application import PromptApplication
from weboob.capabilities.dating import ICapDating
class HaveSex(PromptApplication):
APPNAME = 'havesex'
STORAGE_FILENAME = 'dating.storage'
def main(self, argv):
self.load_config()
self.weboob.load_backends(ICapDating, storage=self.create_storage(self.STORAGE_FILENAME))
return self.loop()
def split_id(self, id):
try:
bname, id = id.split('.', 1)
except ValueError:
return None, None
return self.weboob.backends.get(bname, None), id
@PromptApplication.command("exit program")
def command_exit(self):
print 'Returning in real-life...'
self.weboob.want_stop()
@PromptApplication.command("show a profile")
def command_profile(self, id):
backend, _id = self.split_id(id)
if not backend:
print 'Invalid ID: %s' % id
return False
profile = backend.get_profile(_id)
if not profile:
print 'Profile not found'
print profile.get_profile_text()
return True
@PromptApplication.command("start profiles walker")
def command_walker(self):
for name, backend in self.weboob.iter_backends():
backend.start_profiles_walker()

3
weboob/ouiboube.py
View file

@ -96,5 +96,8 @@ class Weboob:
def repeat(self, interval, function, *args):
return self.scheduler.repeat(interval, function, *args)
def want_stop(self):
return self.scheduler.want_stop()
def loop(self):
return self.scheduler.run()

2
weboob/scheduler.py
View file

@ -36,6 +36,8 @@ class Scheduler(object):
self.running = True
while self.running:
self.scheduler.run()
if not self.scheduler.queue:
self.scheduler.delayfunc(0.001)
return True
def want_stop(self):

23
weboob/tools/application/__init__.py Normal file
View file

@ -0,0 +1,23 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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.
"""
from .base import BaseApplication
from .console import ConsoleApplication
from .prompt import PromptApplication

93
weboob/tools/application/base.py Normal file
View file

@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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
from weboob import Weboob
class BaseApplication(object):
# Application name
APPNAME = ''
# Default configuration
CONFIG = {}
# Configuration directory
CONFDIR = os.path.join(os.path.expanduser('~'), '.weboob')
def __init__(self):
log_format = '%(asctime)s:%(levelname)s:%(filename)s:%(lineno)d %(message)s'
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG, format = log_format)
self.weboob = self.create_weboob()
self.config = None
def create_weboob(self):
return Weboob(self.APPNAME)
def create_storage(self, path=None, klass=None):
"""
Create a storage object.
@param path [str] an optional specific path.
@param klass [IStorage] what klass to instance.
@return a IStorage object
"""
if klass is None:
# load StandardStorage only here because some applications don't
# want # to depend on yaml and do not use this function
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)
return klass(path)
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:
# load Config only here because some applications don't want
# to depend on yaml and do not use this function
from weboob.tools.config.yamlconfig import YamlConfig
klass = YamlConfig
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 function """
raise NotImplementedError()
@classmethod
def run(klass):
app = klass()
sys.exit(app.main(sys.argv))

132
weboob/tools/application/console.py Normal file
View file

@ -0,0 +1,132 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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, tty, termios
import re
from inspect import getargspec
from functools import partial
from weboob.modules import BackendsConfig
from .base import BaseApplication
class ConsoleApplication(BaseApplication):
def __init__(self):
try:
BaseApplication.__init__(self)
except BackendsConfig.WrongPermissions, e:
print >>sys.stderr, 'Error: %s' % e.message
sys.exit(1)
def ask(self, question, default=None, masked=False, regexp=None):
"""
Ask a question to user.
@param question text displayed (str)
@param default optional default value (str)
@param masked if True, do not show typed text (bool)
@param regexp text must match this regexp (str)
@return entered text by user (str)
"""
correct = False
if not default is None:
question = '%s [%s]' % (question, default)
while not correct:
sys.stdout.write('%s: ' % (question))
if masked:
attr = termios.tcgetattr(sys.stdin)
tty.setcbreak(sys.stdin)
line = sys.stdin.readline().split('\n')[0]
if not line and not default is None:
line = default
if masked:
termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, attr)
sys.stdout.write('\n')
correct = not regexp or re.match(regexp, str(line))
return line
def process_command(self, command='help', *args):
def f(x):
return x.startswith('command_' + command)
matching_commands = filter(f, dir(self))
if len(matching_commands) == 0:
sys.stderr.write("No such command: %s.\n" % command)
elif len(matching_commands) == 1:
try:
return getattr(self, matching_commands[0])(*args)
except TypeError, e:
try:
sys.stderr.write("Command '%s' takes %s arguments.\n" % \
(command, int(str(e).split(' ')[3]) - 1))
except:
sys.stderr.write('%s\n' % e)
else:
sys.stderr.write("Ambiguious command %s: %s.\n" %
(command,
', '.join([s.replace('command_', '', 1)
for s in matching_commands])))
_command_help = []
def register_command(f, doc_string, register_to=_command_help):
def getArguments(func, skip=0):
"""
Get arguments of a function as a string.
skip is the number of skipped arguments.
"""
skip += 1
args, varargs, varkw, defaults = getargspec(func)
cut = len(args)
if defaults:
cut -= len(defaults)
args = ["<%s>" % a for a in args[skip:cut]] + \
["[%s]" % a for a in args[cut:]]
if varargs:
args.append("[%s..]" % varargs)
if varkw:
args.append("{WTF}" % varkw)
return " ".join(args)
command = '%s %s' % (f.func_name.replace('command_', ''),
getArguments(f))
register_to.append('%-30s %s' % (command, doc_string))
return f
def command(doc_string, f=register_command):
return partial(f, doc_string=doc_string)
@command("display this notice")
def command_help(self):
sys.stdout.write("Available commands:\n")
for f in self._command_help:
sys.stdout.write(' %s\n' % f)
register_command = staticmethod(register_command)
command = staticmethod(command)

63
weboob/tools/application/prompt.py Normal file
View file

@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
"""
Copyright(C) 2010 Romain Bignon
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 sched
import time
import select
import sys
from weboob import Weboob
from weboob.scheduler import Scheduler
from .console import ConsoleApplication
class PromptScheduler(Scheduler):
def __init__(self, prompt_cb, read_cb):
self.scheduler = sched.scheduler(time.time, self.sleep)
self.read_cb = read_cb
self.prompt_cb = prompt_cb
def sleep(self, d):
self.prompt_cb()
try:
read, write, excepts = select.select([sys.stdin], [], [], d or None)
if read:
line = sys.stdin.readline()
if not line:
self.want_stop()
else:
self.read_cb(line.strip())
except KeyboardInterrupt:
sys.stdout.write('\n')
class PromptApplication(ConsoleApplication):
def create_weboob(self):
return Weboob(self.APPNAME, scheduler=PromptScheduler(self.prompt, self.read_cb))
def prompt(self):
sys.stdout.write('> ')
sys.stdout.flush()
def loop(self):
self.weboob.loop()
def read_cb(self, line):
line = line.split()
self.process_command(*line)