/usr/share/weboob/munin/boobank-munin is in weboob 1.2-1.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 | #! /usr/bin/python
# -*- coding: utf-8 -*-
# vim: ft=python et softtabstop=4 cinoptions=4 shiftwidth=4 ts=4 ai
# Copyright(C) 2010-2011 Romain Bignon
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from __future__ import print_function
import os
import sys
import locale
import time
import logging
from weboob.core import Weboob, CallErrors
from weboob.capabilities.bank import CapBank
from weboob.exceptions import BrowserIncorrectPassword
class BoobankMuninPlugin(object):
def __init__(self):
if 'weboob_path' in os.environ:
self.weboob = Weboob(os.environ['weboob_path'])
else:
self.weboob = Weboob()
self.monitored_accounts = None
if 'boobank_monitored' in os.environ:
self.monitored_accounts = os.environ['boobank_monitored'].split(' ')
self.cache_expire = long(os.environ.get('boobank_cache_expire', 3600))
self.add_coming = int(os.environ.get('boobank_add_coming', 1))
self.cumulate = int(os.environ.get('boobank_cumulate', 1))
self.cache = None
def display_help(self):
print('boobank-munin is a plugin for munin')
print('')
print('Copyright(C) 2010-2011 Romain Bignon')
print('')
print('To use it, create a symlink /etc/munin/plugins/boobank to this script')
print('and add this section in /etc/munin/plugin-conf.d/munin-node:')
print('')
print('[boobank]')
print('user romain')
print('group romain')
print('env.HOME /home/romain')
print('# The weboob directory path.')
print('env.weboob_path /home/romain/.config/weboob/')
print('# Monitored accounts. If this parameter is missing, all accounts')
print('# will be displayed.')
print('env.boobank_monitored 0125XXXXXXXXXXXX@bnporc 0125XXXXXXXXXXXX@bnporc')
print('# To prevent mass connections to bank websites, results are cached.')
print('# You can set here the expiration delay (in seconds).')
print('env.boobank_cache_expire 7200')
print('# If enabled, coming operations are added to the value of accounts\'')
print('# balance.')
print('env.boobank_add_coming 1')
print('# Cumulate accounts values')
print('env.boobank_cumulate 1')
print('')
print('When you change configuration, you can use this command to reset cache:')
print('$ boobank-munin --reset')
def clear_cache(self):
for name in ('boobank-munin', 'boobank-munin-config'):
try:
os.unlink(self.cachepath(name))
except IOError:
pass
def cachepath(self, name):
tmpdir = os.path.join(self.weboob.workdir, "munin")
if not os.path.isdir(tmpdir):
os.makedirs(tmpdir)
return os.path.join(tmpdir, name)
def check_cache(self, name):
return self.print_cache(name, check=True)
def print_cache(self, name, check=False):
try:
f = open(self.cachepath(name), 'r')
except IOError:
return False
try:
last = int(f.readline().strip())
except ValueError:
return False
if check and (last + self.cache_expire) < time.time():
return False
for line in f:
sys.stdout.write(line)
return True
def new_cache(self, name):
os.umask(0o077)
new_name = '%s.new' % name
filename = self.cachepath(new_name)
try:
f = open(filename, 'w')
except IOError as e:
print('Unable to create the cache file %s: %s' % (filename, e), file=sys.stderr)
return
self.cache = f
self.cache.write('%d\n' % time.time())
def flush_cache(self):
old_name = self.cache.name
new_name = self.cache.name[:-4]
self.cache.close()
os.rename(old_name, new_name)
def write_output(self, line):
sys.stdout.write('%s\n' % line)
if self.cache:
self.cache.write('%s\n' % line)
def config(self):
if self.check_cache('boobank-munin-config'):
return
self.new_cache('boobank-munin-config')
self.weboob.load_backends(CapBank)
self.write_output('graph_title Bank accounts')
self.write_output('graph_vlabel balance')
self.write_output('graph_category weboob')
self.write_output('graph_args -l 0')
try:
accounts = []
if self.monitored_accounts is not None:
d = {}
for account in self.weboob.do('iter_accounts'):
if self.monitored(account):
d['%s@%s' % (account.id, account.backend)] = account
for id in self.monitored_accounts:
try:
accounts.append(d[id])
except KeyError:
pass
else:
accounts = reversed([a for a in self.weboob.do('iter_accounts')])
first = True
for account in accounts:
id = self.account2id(account)
type = 'STACK'
if first:
type = 'AREA'
first = False
self.write_output('%s.label %s' % (id, account.label.encode('iso-8859-15')))
if self.cumulate:
self.write_output('%s.draw %s' % (id, type))
except CallErrors as errors:
self.print_errors(errors)
self.print_cache('boobank-munin-config')
else:
self.flush_cache()
def monitored(self, account):
return not self.monitored_accounts or ('%s@%s' % (account.id, account.backend)) in self.monitored_accounts
def account2id(self, account):
return '%s_%s' % (account.backend, account.id)
def print_errors(self, errors):
for backend, err, backtrace in errors:
print((u'%s(%s): %s' % (type(err).__name__, backend.name, err)).encode(sys.stdout.encoding or locale.getpreferredencoding(), 'replace'), file=sys.stderr)
if isinstance(err, BrowserIncorrectPassword):
self.weboob.backends_config.edit_backend(backend.name, backend.NAME, {'_enabled': 'false'})
def execute(self):
if self.check_cache('boobank-munin'):
return
self.new_cache('boobank-munin')
self.weboob.load_backends(CapBank)
try:
for account in self.weboob.do('iter_accounts'):
if self.monitored(account):
balance = account.balance
if account.coming and self.add_coming:
balance += account.coming
self.write_output('%s.value %d' % (self.account2id(account), balance))
except CallErrors as errors:
self.print_errors(errors)
self.print_cache('boobank-munin')
else:
self.flush_cache()
def run(self):
cmd = (len(sys.argv) > 1 and sys.argv[1]) or "execute"
if cmd == 'execute':
self.execute()
elif cmd == 'config':
self.config()
elif cmd == 'autoconf':
print('no')
sys.exit(1)
elif cmd == 'suggest':
sys.exit(1)
elif cmd == 'help' or cmd == '-h' or cmd == '--help':
self.display_help()
elif cmd == 'reload' or cmd == '--reload' or \
cmd == 'reset' or cmd == '--reset':
self.clear_cache()
if self.cache:
self.cache.close()
sys.exit(0)
if __name__ == '__main__':
logging.basicConfig()
BoobankMuninPlugin().run()
|