nojhan/weboob-devel
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

port every console applications to ReplApplication

Browse source
This commit is contained in:
Romain Bignon 2010-09-24 12:58:27 -04:00
commit 37a53b783a
17 changed files with 460 additions and 369 deletions
Download patch file Download diff file Expand all files Collapse all files

29
weboob/applications/boobank/boobank.py
View file

@ -1,6 +1,4 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# vim: ft=python et softtabstop=4 cinoptions=4 shiftwidth=4 ts=4 ai
# Copyright(C) 2009-2010 Romain Bignon, Christophe Benz
#
@ -39,6 +37,8 @@ class Boobank(ReplApplication):
def do_list(self, line):
"""
list
List every available accounts.
"""
tot_balance = 0.0
@ -61,6 +61,8 @@ class Boobank(ReplApplication):
def do_history(self, id):
"""
history ID
Display old operations.
"""
id, backend_name = self.parse_id(id)
@ -76,6 +78,8 @@ class Boobank(ReplApplication):
def do_coming(self, id):
"""
coming ID
Display all future operations.
"""
id, backend_name = self.parse_id(id)
@ -88,25 +92,22 @@ class Boobank(ReplApplication):
for backend, operation in self.do(do):
self.format(operation)
def do_transfer(self, arg):
def do_transfer(self, line):
"""
transfer FROM TO AMOUNT
Make a transfer beetwen two account
"""
id_from , id_to, amount = arg.split()
id_from, id_to, amount = self.parseline(line, 3, 3)
id_from, backend_name = self.parse_id(id_from)
id_to, backend_name = self.parse_id(id_to)
names = (backend_name,) if backend_name is not None else None
self.load_backends(ICapBank, names=names)
def do(backend):
return backend.transfer(id_from, id_to, amount)
return backend.transfer(id_from, id_to, float(amount))
for backend, operation in self.do(do):
pass

36
weboob/applications/chatoob/chatoob.py
View file

@ -18,7 +18,7 @@
import logging
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
from weboob.capabilities.chat import ICapChat
from weboob.capabilities.contact import ICapContact, Contact
@ -26,36 +26,42 @@ from weboob.capabilities.contact import ICapContact, Contact
__all__ = ['Chatoob']
class Chatoob(ConsoleApplication):
class Chatoob(ReplApplication):
APPNAME = 'chatoob'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Christophe Benz'
def main(self, argv):
def load_default_backends(self):
self.load_backends(ICapChat)
#for backend, result in self.do('start_chat_polling', self.on_new_chat_message):
#logging.info(u'Polling chat messages for backend %s' % backend)
return self.process_command(*argv[1:])
def on_new_chat_message(self, message):
print 'on_new_chat_message: %s' % message
@ConsoleApplication.command('exit program')
def command_exit(self):
self.weboob.want_stop()
def do_list(self, line):
"""
list
@ConsoleApplication.command('list online contacts')
def command_list(self):
List all contacts.
"""
for backend, contact in self.do('iter_contacts', status=Contact.STATUS_ONLINE, caps=ICapContact):
self.format(contact)
@ConsoleApplication.command('get messages')
def command_messages(self):
def do_messages(self, line):
"""
messages
Get messages.
"""
for backend, message in self.do('iter_chat_messages'):
self.format(message)
@ConsoleApplication.command('send message to contact')
def command_send(self, _id, message):
def do_send(self, line):
"""
send CONTACT MESSAGE
Send a message to the specified contact.
"""
_id, message = self.parseline(line, 2, 2)
for backend, result in self.do('send_chat_message', _id, message):
if not result:
logging.error(u'Failed to send message to contact id="%s" on backend "%s"' % (_id, backend.name))

8
weboob/applications/geolooc/geolooc.py
View file

@ -19,23 +19,25 @@
import sys
from weboob.capabilities.geolocip import ICapGeolocIp
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
__all__ = ['Geolooc']
class Geolooc(ConsoleApplication):
class Geolooc(ReplApplication):
APPNAME = 'geolooc'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
def load_default_backends(self):
self.load_backends(ICapGeolocIp)
def main(self, argv):
if len(argv) < 2:
print >>sys.stderr, 'Syntax: %s ipaddr' % argv[0]
return 1
self.load_backends(ICapGeolocIp)
for backend, location in self.do('get_location', argv[1]):
self.format(location)

35
weboob/applications/havesex/havesex.py
View file

@ -22,23 +22,25 @@ import logging
import sys
import weboob
from weboob.tools.application.prompt import PromptApplication
from weboob.tools.application.repl import ReplApplication
from weboob.capabilities.dating import ICapDating, OptimizationNotFound
__all__ = ['HaveSex']
class HaveSex(PromptApplication):
class HaveSex(ReplApplication):
APPNAME = 'havesex'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
STORAGE_FILENAME = 'dating.storage'
CONFIG = {'optimizations': ''}
def load_default_backends(self):
self.load_backends(ICapDating, storage=self.create_storage(self.STORAGE_FILENAME))
def main(self, argv):
self.load_config()
self.load_backends(ICapDating, storage=self.create_storage(self.STORAGE_FILENAME))
self.do('init_optimizations').wait()
@ -48,15 +50,14 @@ class HaveSex(PromptApplication):
if optimizations_list:
self.optims('Starting', 'start_optimization', optimizations_list)
return self.loop()
return ReplApplication.main(self, argv)
@PromptApplication.command("exit program")
def command_exit(self):
print 'Returning in real-life...'
self.weboob.want_stop()
def do_profile(self, id):
"""
profile ID
@PromptApplication.command("show a profile")
def command_profile(self, id):
Display a profile
"""
_id, backend_name = self.parse_id(id)
def print_node(node, level=1):
@ -116,10 +117,18 @@ class HaveSex(PromptApplication):
if isinstance(error, OptimizationNotFound):
logging.error(u'Optimization "%s" not found' % optim)
@PromptApplication.command("start optimizations")
def command_start(self, *optims):
def do_start(self, *optims):
"""
start OPTIMIZATION [OPTIMIZATION [...]]
Start optimization services.
"""
self.optims('Starting', 'start_optimization', optims)
@PromptApplication.command("stop optimizations")
def command_stop(self, *optims):
"""
stop OPTIMIZATION [OPTIMIZATION [...]]
Stop optimization services.
"""
self.optims('Stopping', 'stop_optimization', optims)

28
weboob/applications/monboob/monboob.py
View file

@ -31,7 +31,7 @@ import asyncore
from weboob.core.ouiboube import Weboob
from weboob.core.scheduler import Scheduler
from weboob.capabilities.messages import ICapMessages, ICapMessagesPost, Thread, Message
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
from weboob.tools.misc import html2text, get_backtrace, utc2local
@ -75,7 +75,7 @@ class MonboobScheduler(Scheduler):
return True
class Monboob(ConsoleApplication):
class Monboob(ReplApplication):
APPNAME = 'monboob'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
@ -91,11 +91,13 @@ class Monboob(ConsoleApplication):
def create_weboob(self):
return Weboob(scheduler=MonboobScheduler(self))
def main(self, argv):
self.load_config()
def load_default_backends(self):
self.load_backends(ICapMessages, storage=self.create_storage())
return self.process_command(*argv[1:])
def main(self, argv):
self.load_config()
return self.onecmd(' '.join(argv[1:]))
def get_email_address_ident(self, msg, header):
s = msg.get(header)
@ -108,8 +110,12 @@ class Monboob(ConsoleApplication):
except IndexError:
return s
@ConsoleApplication.command("pipe with a mail to post message")
def command_post(self):
def do_post(self):
"""
post
Pipe with a mail to post message.
"""
msg = message_from_file(sys.stdin)
return self.process_incoming_mail(msg)
@ -185,8 +191,12 @@ class Monboob(ConsoleApplication):
parent=Message(thread, msg_id),
content=content))
@ConsoleApplication.command("run daemon")
def command_run(self):
def do_run(self):
"""
run
Run the fetching daemon.
"""
self.weboob.repeat(int(self.config.get('interval')), self.process)
self.weboob.loop()

30
weboob/applications/traveloob/traveloob.py
View file

@ -19,28 +19,37 @@
import logging
from weboob.capabilities.travel import ICapTravel
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
__all__ = ['Traveloob']
class Traveloob(ConsoleApplication):
class Traveloob(ReplApplication):
APPNAME = 'traveloob'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
def main(self, argv):
return self.process_command(*argv[1:])
@ConsoleApplication.command('Search stations')
def command_stations(self, pattern):
def load_default_backends(self):
self.load_backends(ICapTravel)
def do_stations(self, pattern):
"""
search PATTERN
Search stations.
"""
for backend, station in self.do('iter_station_search', pattern):
self.format(station)
@ConsoleApplication.command('List all departures for a given station')
def command_departures(self, station, arrival=None):
def do_departures(self, line):
"""
departures STATION [ARRIVAL]
List all departures for a given station.
"""
station, arrival = self.parseline(line, 2, 1)
station_id, backend_name = self.parse_id(station)
if arrival:
arrival_id, backend_name2 = self.parse_id(arrival)
@ -57,6 +66,5 @@ class Traveloob(ConsoleApplication):
else:
backends = None
self.load_backends(ICapTravel, names=backends)
for backend, departure in self.do('iter_station_departures', station_id, arrival_id):
for backend, departure in self.do('iter_station_departures', station_id, arrival_id, backends=backends):
self.format(departure)

3
weboob/applications/videoob/videoob.py
View file

@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2010 Christophe Benz, Romain Bignon
#
# This program is free software; you can redistribute it and/or modify
@ -55,7 +54,7 @@ class Videoob(ReplApplication):
for backend, video in self.do('get_video', _id, backends=backend_names):
self.format(video)
self.flush()
def do_nsfw(self, line):
"""
nsfw [on | off]

70
weboob/applications/weboobcfg/weboobcfg.py
View file

@ -22,21 +22,18 @@ import os
import subprocess
import re
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
from weboob.tools.ordereddict import OrderedDict
__all__ = ['WeboobCfg']
class WeboobCfg(ConsoleApplication):
class WeboobCfg(ReplApplication):
APPNAME = 'weboob-config'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Christophe Benz, Romain Bignon'
def main(self, argv):
return self.process_command(*argv[1:])
def caps_included(self, modcaps, caps):
modcaps = [x.__name__ for x in modcaps]
for cap in caps:
@ -44,8 +41,18 @@ class WeboobCfg(ConsoleApplication):
return False
return True
@ConsoleApplication.command('Add a configured backend')
def command_add(self, name, *options):
def do_add(self, line):
"""
add NAME [OPTIONS ...]
Add a configured backend.
"""
name, options = self.parseline(line, 2, 1)
if options:
options = options.split(' ')
else:
options = ()
self.weboob.modules_loader.load_all()
if name not in [_name for _name, backend in self.weboob.modules_loader.loaded.iteritems()]:
logging.error(u'Backend "%s" does not exist.' % name)
@ -101,8 +108,12 @@ class WeboobCfg(ConsoleApplication):
except ConfigParser.DuplicateSectionError:
print 'Instance "%s" already exists for backend "%s".' % (new_name, name)
@ConsoleApplication.command('Show configured backends')
def command_listconfigured(self):
def do_listconfigured(self):
"""
list
Show configured backends.
"""
self.set_default_formatter('table')
for instance_name, name, params in sorted(self.weboob.backends_config.iter_backends()):
backend = self.weboob.modules_loader.get_or_load_module(name)
@ -112,20 +123,33 @@ class WeboobCfg(ConsoleApplication):
])
self.format(row)
@ConsoleApplication.command('Remove a configured backend')
def command_remove(self, instance_name):
def do_remove(self, instance_name):
"""
remove NAME
Remove a configured backend.
"""
try:
self.weboob.backends_config.remove_backend(instance_name)
except ConfigParser.NoSectionError:
logging.error('Backend instance "%s" does not exist' % instance_name)
return 1
@ConsoleApplication.command('Edit configuration file')
def command_edit(self):
def do_edit(self):
"""
edit
Edit configuration file.
"""
subprocess.call([os.environ.get('EDITOR', 'vi'), self.weboob.backends_config.confpath])
@ConsoleApplication.command('Show available backends')
def command_backends(self, *caps):
def do_backends(self, line):
"""
backends [CAPS ...]
Show available backends
"""
caps = line.split(' ')
self.set_default_formatter('table')
self.weboob.modules_loader.load_all()
for name, backend in sorted(self.weboob.modules_loader.loaded.iteritems()):
@ -137,8 +161,12 @@ class WeboobCfg(ConsoleApplication):
])
self.format(row)
@ConsoleApplication.command('Display information about a backend')
def command_info(self, name):
def do_info(self, name):
"""
info NAME
Display information about a backend.
"""
try:
backend = self.weboob.modules_loader.get_or_load_module(name)
except KeyError:
@ -166,8 +194,12 @@ class WeboobCfg(ConsoleApplication):
print '| | %s: %s' % (key, value)
print "'-----------------'"
@ConsoleApplication.command('Show applications')
def command_applications(self, *caps):
def do_applications(self, line):
"""
applications
Show applications.
"""
applications = set()
import weboob.applications
for path in weboob.applications.__path__:

7
weboob/applications/weboobcli/weboobcli.py
View file

@ -18,17 +18,20 @@
import sys
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
__all__ = ['WeboobCli']
class WeboobCli(ConsoleApplication):
class WeboobCli(ReplApplication):
APPNAME = 'weboob-cli'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
def load_default_backends(self):
pass
def main(self, argv):
if len(argv) < 3:
print >>sys.stderr, "Syntax: %s capability command [args ..]" % argv[0]

16
weboob/applications/weboobdebug/weboobdebug.py
View file

@ -18,19 +18,23 @@
import logging
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
class WeboobDebug(ConsoleApplication):
class WeboobDebug(ReplApplication):
APPNAME = 'weboobdebug'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Christophe Benz'
def main(self, argv):
return self.process_command(*argv[1:])
def load_default_backends(self):
pass
@ConsoleApplication.command('Debug backend')
def command_shell(self, backend_name):
def do_shell(self, backend_name):
"""
shell BACKEND
Debug a backend.
"""
try:
backend = self.weboob.load_backends(names=[backend_name])[backend_name]
except KeyError:

37
weboob/applications/weboorrents/weboorrents.py
View file

@ -20,23 +20,26 @@ from __future__ import with_statement
import sys
from weboob.capabilities.torrent import ICapTorrent
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
__all__ = ['Weboorrents']
class Weboorrents(ConsoleApplication):
class Weboorrents(ReplApplication):
APPNAME = 'weboorrents'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
def main(self, argv):
def load_default_backends(self):
self.load_backends(ICapTorrent)
return self.process_command(*argv[1:])
@ConsoleApplication.command('Get information about a torrent')
def command_info(self, id):
def do_info(self, id):
"""
info ID
Get information about a torrent.
"""
_id, backend_name = self.parse_id(id)
found = 0
@ -48,8 +51,16 @@ class Weboorrents(ConsoleApplication):
if not found:
print >>sys.stderr, 'Torrent "%s" not found' % id
@ConsoleApplication.command('Get the torrent file')
def command_getfile(self, id, dest):
def do_getfile(self, line):
"""
getfile ID FILENAME
Get the .torrent file.
FILENAME is where to write the file. If FILENAME is '-',
the file is written to stdout.
"""
id, dest = self.parseline(line, 2, 2)
_id, backend_name = self.parse_id(id)
for backend, buf in self.do('get_torrent_file', _id, backends=backend_name):
@ -63,8 +74,14 @@ class Weboorrents(ConsoleApplication):
print >>sys.stderr, 'Torrent "%s" not found' % id
@ConsoleApplication.command('Search torrents')
def command_search(self, pattern=None):
def do_search(self, pattern):
"""
search [PATTERN]
Search torrents.
"""
if not pattern:
pattern = None
self.set_formatter_header(u'Search pattern: %s' % pattern if pattern else u'Latest torrents')
for backend, torrent in self.do('iter_torrents', pattern=pattern):
self.format(torrent)

36
weboob/applications/wetboobs/wetboobs.py
View file

@ -16,51 +16,61 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
import logging
import sys
from weboob.core import CallErrors
from weboob.capabilities.weather import ICapWeather, CityNotFound
from weboob.tools.application.console import ConsoleApplication
from weboob.tools.application.repl import ReplApplication
__all__ = ['WetBoobs']
class WetBoobs(ConsoleApplication):
class WetBoobs(ReplApplication):
APPNAME = 'wetboobs'
VERSION = '0.1'
COPYRIGHT = 'Copyright(C) 2010 Romain Bignon'
def main(self, argv):
def load_default_backends(self):
self.load_backends(ICapWeather)
return self.process_command(*argv[1:])
def do_search(self, pattern):
"""
search PATTERN
@ConsoleApplication.command('search cities')
def command_search(self, pattern):
Search cities.
"""
for backend, city in self.do('iter_city_search', pattern):
self.format(city)
@ConsoleApplication.command('get current weather')
def command_current(self, city):
def do_current(self, city):
"""
current CITY
Get current weather.
"""
try:
for backend, current in self.do('get_current', city):
self.format(current)
except CallErrors, e:
for error in e:
if isinstance(error, CityNotFound):
logging.error('City "%s" not found' % city)
print >>sys.stderr, 'City "%s" not found' % city
else:
raise error
@ConsoleApplication.command('get forecasts')
def command_forecasts(self, city):
def do_forecasts(self, city):
"""
forecasts CITY
Get forecasts.
"""
try:
for backend, forecast in self.do('iter_forecast', city):
self.format(forecast)
except CallErrors, e:
for error in e:
if isinstance(error, CityNotFound):
logging.error('City "%s" not found' % city)
print >>sys.stderr, 'City "%s" not found' % city
else:
raise error