Add module for HelloBank support (closes #1276)
Signed-off-by: Romain Bignon <romain@symlink.me>
This commit is contained in:
parent
836dd27bfd
commit
14b6d75533
10 changed files with 942 additions and 0 deletions
23
modules/hellobank/__init__.py
Normal file
23
modules/hellobank/__init__.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# 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 .backend import HelloBankBackend
|
||||
|
||||
__all__ = ['HelloBankBackend']
|
||||
151
modules/hellobank/backend.py
Executable file
151
modules/hellobank/backend.py
Executable file
|
|
@ -0,0 +1,151 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 Christophe Lampin
|
||||
# Copyright(C) 2010-2012 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/>.
|
||||
|
||||
|
||||
# python2.5 compatibility
|
||||
from __future__ import with_statement
|
||||
|
||||
from decimal import Decimal
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from weboob.capabilities.bank import ICapBank, AccountNotFound, Account, Recipient
|
||||
from weboob.capabilities.messages import ICapMessages, Thread
|
||||
from weboob.tools.backend import BaseBackend, BackendConfig
|
||||
from weboob.tools.value import ValueBackendPassword
|
||||
|
||||
from .browser import HelloBank
|
||||
|
||||
|
||||
__all__ = ['HelloBankBackend']
|
||||
|
||||
|
||||
class HelloBankBackend(BaseBackend, ICapBank, ICapMessages):
|
||||
NAME = 'hellobank'
|
||||
MAINTAINER = u'Christophe Lampin'
|
||||
EMAIL = 'weboob@lampin.net'
|
||||
VERSION = '0.g'
|
||||
LICENSE = 'AGPLv3+'
|
||||
DESCRIPTION = 'Hello Bank ! website'
|
||||
CONFIG = BackendConfig(ValueBackendPassword('login', label='Account ID', masked=False),
|
||||
ValueBackendPassword('password', label='Password', regexp='^(\d{6}|)$'))
|
||||
BROWSER = HelloBank
|
||||
STORAGE = {'seen': []}
|
||||
|
||||
# Store the messages *list* for this duration
|
||||
CACHE_THREADS = timedelta(seconds=3 * 60 * 60)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
BaseBackend.__init__(self, *args, **kwargs)
|
||||
self._threads = None
|
||||
self._threads_age = datetime.utcnow()
|
||||
|
||||
def create_default_browser(self):
|
||||
return self.create_browser(self.config['login'].get(),
|
||||
self.config['password'].get())
|
||||
|
||||
def iter_accounts(self):
|
||||
for account in self.browser.get_accounts_list():
|
||||
yield account
|
||||
|
||||
def get_account(self, _id):
|
||||
if not _id.isdigit():
|
||||
raise AccountNotFound()
|
||||
with self.browser:
|
||||
account = self.browser.get_account(_id)
|
||||
if account:
|
||||
return account
|
||||
else:
|
||||
raise AccountNotFound()
|
||||
|
||||
def iter_history(self, account):
|
||||
with self.browser:
|
||||
return self.browser.iter_history(account)
|
||||
|
||||
def iter_coming(self, account):
|
||||
with self.browser:
|
||||
return self.browser.iter_coming_operations(account)
|
||||
|
||||
def iter_transfer_recipients(self, ignored):
|
||||
for account in self.browser.get_transfer_accounts().itervalues():
|
||||
recipient = Recipient()
|
||||
recipient.id = account.id
|
||||
recipient.label = account.label
|
||||
yield recipient
|
||||
|
||||
def transfer(self, account, to, amount, reason=None):
|
||||
if isinstance(account, Account):
|
||||
account = account.id
|
||||
|
||||
try:
|
||||
assert account.isdigit()
|
||||
assert to.isdigit()
|
||||
amount = Decimal(amount)
|
||||
except (AssertionError, ValueError):
|
||||
raise AccountNotFound()
|
||||
|
||||
with self.browser:
|
||||
return self.browser.transfer(account, to, amount, reason)
|
||||
|
||||
def iter_threads(self, cache=False):
|
||||
"""
|
||||
If cache is False, always fetch the threads from the website.
|
||||
"""
|
||||
old = self._threads_age < datetime.utcnow() - self.CACHE_THREADS
|
||||
threads = self._threads
|
||||
if not cache or threads is None or old:
|
||||
with self.browser:
|
||||
threads = list(self.browser.iter_threads())
|
||||
# the website is stupid and does not have the messages in the proper order
|
||||
threads = sorted(threads, key=lambda t: t.date, reverse=True)
|
||||
self._threads = threads
|
||||
seen = self.storage.get('seen', default=[])
|
||||
for thread in threads:
|
||||
if thread.id not in seen:
|
||||
thread.root.flags |= thread.root.IS_UNREAD
|
||||
else:
|
||||
thread.root.flags &= ~thread.root.IS_UNREAD
|
||||
yield thread
|
||||
|
||||
def fill_thread(self, thread, fields=None):
|
||||
if fields is None or 'root' in fields:
|
||||
return self.get_thread(thread)
|
||||
|
||||
def get_thread(self, _id):
|
||||
if isinstance(_id, Thread):
|
||||
thread = _id
|
||||
_id = thread.id
|
||||
else:
|
||||
thread = Thread(_id)
|
||||
with self.browser:
|
||||
thread = self.browser.get_thread(thread)
|
||||
return thread
|
||||
|
||||
def iter_unread_messages(self):
|
||||
threads = list(self.iter_threads(cache=True))
|
||||
for thread in threads:
|
||||
if thread.root.flags & thread.root.IS_UNREAD:
|
||||
thread = self.fillobj(thread) or thread
|
||||
yield thread.root
|
||||
|
||||
def set_message_read(self, message):
|
||||
self.storage.get('seen', default=[]).append(message.thread.id)
|
||||
self.storage.save()
|
||||
|
||||
OBJECTS = {Thread: fill_thread}
|
||||
228
modules/hellobank/browser.py
Executable file
228
modules/hellobank/browser.py
Executable file
|
|
@ -0,0 +1,228 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 Christophe Lampin
|
||||
# Copyright(C) 2009-2013 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/>.
|
||||
|
||||
|
||||
import urllib
|
||||
import mechanize
|
||||
from datetime import datetime
|
||||
from logging import warning
|
||||
|
||||
from weboob.tools.browser import BaseBrowser, BrowserIncorrectPassword, BrowserPasswordExpired
|
||||
from weboob.capabilities.bank import TransferError, Transfer
|
||||
|
||||
from .perso.accounts_list import AccountsList, AccountPrelevement
|
||||
from .perso.transactions import AccountHistory, AccountComing
|
||||
from .perso.transfer import TransferPage, TransferConfirmPage, TransferCompletePage
|
||||
from .perso.login import LoginPage, ConfirmPage, InfoMessagePage
|
||||
from .perso.messages import MessagePage, MessagesPage
|
||||
|
||||
__all__ = ['HelloBank']
|
||||
|
||||
|
||||
class HelloBank(BaseBrowser):
|
||||
DOMAIN = 'client.hellobank.fr'
|
||||
PROTOCOL = 'https'
|
||||
ENCODING = None # refer to the HTML encoding
|
||||
PAGES = {'.*TableauBord.*': AccountsList,
|
||||
'.*type=folder.*': AccountHistory,
|
||||
'.*pageId=mouvementsavenir.*': AccountComing,
|
||||
'.*NS_AVEDP.*': AccountPrelevement,
|
||||
'.*NS_VIRDF.*': TransferPage,
|
||||
'.*NS_VIRDC.*': TransferConfirmPage,
|
||||
'.*/NS_VIRDA\?stp=(?P<id>\d+).*': TransferCompletePage,
|
||||
'.*type=homeconnex.*': LoginPage,
|
||||
'.*layout=HomeConnexion.*': ConfirmPage,
|
||||
'.*SAF_CHM_VALID.*': ConfirmPage,
|
||||
'.*Action=DSP_MSG.*': InfoMessagePage,
|
||||
'.*Messages_recus.*': MessagesPage,
|
||||
'.*Lire_Message.*': MessagePage,
|
||||
}
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
BaseBrowser.__init__(self, *args, **kwargs)
|
||||
|
||||
def home(self):
|
||||
self.location('https://client.hellobank.fr/banque/portail/digitale/HomeConnexion?type=homeconnex')
|
||||
|
||||
def is_logged(self):
|
||||
return not self.is_on_page(LoginPage)
|
||||
|
||||
def login(self):
|
||||
assert isinstance(self.username, basestring)
|
||||
assert isinstance(self.password, basestring)
|
||||
assert self.password.isdigit()
|
||||
|
||||
if not self.is_on_page(LoginPage):
|
||||
self.home()
|
||||
|
||||
self.page.login(self.username, self.password)
|
||||
self.location('/NSFR?Action=DSP_VGLOBALE', no_login=True)
|
||||
|
||||
if self.is_on_page(LoginPage):
|
||||
raise BrowserIncorrectPassword()
|
||||
|
||||
def get_accounts_list(self):
|
||||
# We have to parse transfer page to get the IBAN numbers
|
||||
if not self.is_on_page(TransferPage):
|
||||
now = datetime.now()
|
||||
self.location('/NS_VIRDF?Origine=DSP_VIR&stp=%s' % now.strftime("%Y%m%d%H%M%S"))
|
||||
|
||||
accounts = self.page.get_accounts()
|
||||
if len(accounts) == 0:
|
||||
print 'no accounts'
|
||||
# oops, no accounts? check if we have not exhausted the allowed use
|
||||
# of this password
|
||||
for img in self.document.getroot().cssselect('img[align="middle"]'):
|
||||
if img.attrib.get('alt', '') == 'Changez votre code secret':
|
||||
raise BrowserPasswordExpired('Your password has expired')
|
||||
self.location('/NSFR?Action=DSP_VGLOBALE')
|
||||
return self.page.get_list(accounts)
|
||||
|
||||
def get_account(self, id):
|
||||
assert isinstance(id, basestring)
|
||||
|
||||
l = self.get_accounts_list()
|
||||
for a in l:
|
||||
if a.id == id:
|
||||
return a
|
||||
|
||||
return None
|
||||
|
||||
def get_IBAN_from_account(self, account):
|
||||
self.go_to_history_page(account)
|
||||
return self.page.get_IBAN()
|
||||
|
||||
def go_to_history_page(self,account):
|
||||
if account._link_id is None:
|
||||
return iter([])
|
||||
|
||||
if not self.is_on_page(AccountsList):
|
||||
self.location('/NSFR?Action=DSP_VGLOBALE')
|
||||
|
||||
data = {'gt': 'homepage:basic-theme',
|
||||
'externalIAId': 'IAStatements',
|
||||
'cboFlowName': 'flow/iastatement',
|
||||
'contractId': account._link_id,
|
||||
'groupId': '-2',
|
||||
'pastOrPendingOperations': 1,
|
||||
'groupSelected':'-2',
|
||||
'step': 'STAMENTS',
|
||||
'pageId': 'releveoperations',
|
||||
'sendEUD': 'true',
|
||||
}
|
||||
self.location('/udc', urllib.urlencode(data))
|
||||
|
||||
return None
|
||||
|
||||
def go_to_coming_operations_page(self,account):
|
||||
if account._link_id is None:
|
||||
return iter([])
|
||||
|
||||
if not self.is_on_page(AccountsList):
|
||||
self.location('/NSFR?Action=DSP_VGLOBALE')
|
||||
|
||||
data = {'gt': 'homepage:basic-theme',
|
||||
'externalIAId': 'IAStatements',
|
||||
'cboFlowName': 'flow/iastatement',
|
||||
'contractId': account._link_id,
|
||||
'groupId': '-2',
|
||||
'pastOrPendingOperations': 2,
|
||||
'groupSelected':'-2',
|
||||
'step': 'STAMENTS',
|
||||
'pageId': 'mouvementsavenir',
|
||||
'sendEUD': 'true',
|
||||
}
|
||||
self.location('/udc', urllib.urlencode(data))
|
||||
|
||||
return None
|
||||
|
||||
def iter_history(self, account):
|
||||
self.go_to_history_page(account)
|
||||
return self.page.iter_operations()
|
||||
|
||||
def iter_coming_operations(self, account):
|
||||
self.go_to_coming_operations_page(account)
|
||||
return self.page.iter_coming_operations()
|
||||
|
||||
def get_transfer_accounts(self):
|
||||
if not self.is_on_page(TransferPage):
|
||||
self.location('/NS_VIRDF')
|
||||
|
||||
assert self.is_on_page(TransferPage)
|
||||
return self.page.get_accounts()
|
||||
|
||||
def transfer(self, from_id, to_id, amount, reason=None):
|
||||
if not self.is_on_page(TransferPage):
|
||||
self.location('/NS_VIRDF')
|
||||
|
||||
# Need to clean HTML before parse it
|
||||
html = self.response().get_data().replace("<!input", "<input")
|
||||
response = mechanize.make_response(
|
||||
html, [("Content-Type", "text/html")],
|
||||
"https://client.hellobank.fr/NS_VIRDF", 200, "OK")
|
||||
self.set_response(response)
|
||||
|
||||
accounts = self.page.get_accounts()
|
||||
self.page.transfer(from_id, to_id, amount, reason)
|
||||
|
||||
if not self.is_on_page(TransferCompletePage):
|
||||
raise TransferError('An error occured during transfer')
|
||||
|
||||
transfer = Transfer(self.page.get_id())
|
||||
transfer.amount = amount
|
||||
transfer.origin = accounts[from_id].label
|
||||
transfer.recipient = accounts[to_id].label
|
||||
transfer.date = datetime.now()
|
||||
return transfer
|
||||
|
||||
def messages_page(self):
|
||||
if not self.is_on_page(MessagesPage):
|
||||
if not self.is_on_page(AccountsList):
|
||||
self.location('/NSFR?Action=DSP_VGLOBALE')
|
||||
self.location(self.page.get_messages_link())
|
||||
assert self.is_on_page(MessagesPage)
|
||||
|
||||
def iter_threads(self):
|
||||
self.messages_page()
|
||||
for thread in self.page.iter_threads():
|
||||
yield thread
|
||||
|
||||
def get_thread(self, thread):
|
||||
self.messages_page()
|
||||
if not hasattr(thread, '_link_id') or not thread._link_id:
|
||||
for t in self.iter_threads():
|
||||
if t.id == thread.id:
|
||||
thread = t
|
||||
break
|
||||
# mimic validerFormulaire() javascript
|
||||
# yes, it makes no sense
|
||||
page_id, unread = thread._link_id
|
||||
self.select_form('listerMessages')
|
||||
self.form.set_all_readonly(False)
|
||||
self['identifiant'] = page_id
|
||||
if len(thread.id):
|
||||
self['idMessage'] = thread.id
|
||||
# the JS does this, but it makes us unable to read unread messages
|
||||
#if unread:
|
||||
# self['newMsg'] = thread.id
|
||||
self.submit()
|
||||
assert self.is_on_page(MessagePage)
|
||||
thread.root.content = self.page.get_content()
|
||||
return thread
|
||||
0
modules/hellobank/perso/__init__.py
Normal file
0
modules/hellobank/perso/__init__.py
Normal file
95
modules/hellobank/perso/accounts_list.py
Executable file
95
modules/hellobank/perso/accounts_list.py
Executable file
|
|
@ -0,0 +1,95 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 Christophe Lampin
|
||||
# Copyright(C) 2009-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/>.
|
||||
|
||||
|
||||
import re
|
||||
from decimal import Decimal
|
||||
|
||||
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
|
||||
from weboob.capabilities.bank import Account
|
||||
from weboob.capabilities.base import NotAvailable
|
||||
from weboob.tools.browser import BasePage, BrokenPageError, BrowserPasswordExpired
|
||||
from weboob.tools.json import json
|
||||
import unicodedata as ud
|
||||
|
||||
__all__ = ['AccountsList', 'AccountPrelevement']
|
||||
|
||||
|
||||
class AccountsList(BasePage):
|
||||
ACCOUNT_TYPES = {
|
||||
1: Account.TYPE_CHECKING,
|
||||
2: Account.TYPE_SAVINGS,
|
||||
3: Account.TYPE_DEPOSIT,
|
||||
5: Account.TYPE_MARKET, # FIX ME : I don't know the right code
|
||||
9: Account.TYPE_LOAN,
|
||||
}
|
||||
|
||||
def on_loaded(self):
|
||||
pass
|
||||
|
||||
def get_list(self, accounts_ids):
|
||||
l = []
|
||||
# Read the json data
|
||||
json_data = self.browser.readurl('/banque/PA_Autonomy-war/ProxyIAService?cleOutil=IA_SMC_UDC&service=getlstcpt&dashboard=true&refreshSession=true&cre=udc&poka=true')
|
||||
json_infos = json.loads(json_data)
|
||||
for famille in json_infos['smc']['data']['familleCompte']:
|
||||
id_famille = famille['idFamilleCompte']
|
||||
for compte in famille['compte']:
|
||||
account = Account()
|
||||
account.label = u''+compte['libellePersoProduit']
|
||||
account.currency = account.get_currency(compte['devise'])
|
||||
account.balance = Decimal(compte['soldeDispo'])
|
||||
account.coming = Decimal(compte['soldeAVenir'])
|
||||
account.type = self.ACCOUNT_TYPES.get(id_famille, Account.TYPE_UNKNOWN)
|
||||
account.id = 0
|
||||
account._link_id = 'KEY'+compte['key']
|
||||
|
||||
# IBAN aren't in JSON
|
||||
# Fast method, get it from transfer page.
|
||||
for i,a in accounts_ids.items():
|
||||
if a.label == account.label:
|
||||
account.id = i
|
||||
# But it's doesn't work with LOAN and MARKET, so use slow method : Get it from transaction page.
|
||||
if account.id == 0:
|
||||
account.id = self.browser.get_IBAN_from_account(account)
|
||||
l.append(account)
|
||||
|
||||
if len(l) == 0:
|
||||
print 'no accounts'
|
||||
# oops, no accounts? check if we have not exhausted the allowed use
|
||||
# of this password
|
||||
for img in self.document.getroot().cssselect('img[align="middle"]'):
|
||||
if img.attrib.get('alt', '') == 'Changez votre code secret':
|
||||
raise BrowserPasswordExpired('Your password has expired')
|
||||
return l
|
||||
|
||||
def get_execution_id(self):
|
||||
return self.document.xpath('//input[@name="_flowExecutionKey"]')[0].attrib['value']
|
||||
|
||||
def get_messages_link(self):
|
||||
"""
|
||||
Get the link to the messages page, which seems to have an identifier in it.
|
||||
"""
|
||||
return self.document.xpath('//a[@title="Messagerie"]')[0].attrib['href']
|
||||
|
||||
|
||||
|
||||
class AccountPrelevement(AccountsList):
|
||||
pass
|
||||
118
modules/hellobank/perso/login.py
Executable file
118
modules/hellobank/perso/login.py
Executable file
|
|
@ -0,0 +1,118 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 Christophe Lampin
|
||||
# Copyright(C) 2009-2011 Romain Bignon, Pierre Mazière
|
||||
#
|
||||
# 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/>.
|
||||
|
||||
|
||||
import time
|
||||
import re
|
||||
from weboob.tools.mech import ClientForm
|
||||
import urllib
|
||||
|
||||
from weboob.tools.browser import BasePage, BrowserUnavailable
|
||||
from weboob.tools.captcha.virtkeyboard import VirtKeyboard,VirtKeyboardError
|
||||
|
||||
__all__ = ['LoginPage', 'ConfirmPage', 'ChangePasswordPage']
|
||||
|
||||
|
||||
class HelloBankVirtKeyboard(VirtKeyboard):
|
||||
symbols={'0':'4d1e060efb694ee60e4bd062d800401c',
|
||||
'1':'509134b5c09980e282cdd5867815e9e3',
|
||||
'2':'4cd09c9c44405e00b12e0371e2f972ae',
|
||||
'3':'227d854efc5623292eda4ca2f9bfc4d7',
|
||||
'4':'be8d23e7f5fce646193b7b520ff80443',
|
||||
'5':'5fe450b35c946c3a983f1df6e5b41fd1',
|
||||
'6':'113a6f63714f5094c7f0b25caaa66f78',
|
||||
'7':'de0e93ba880a8a052aea79237f08f3f8',
|
||||
'8':'3d70474c05c240b606556c89baca0568',
|
||||
'9':'040954a5e5e93ec2fb03ac0cfe592ac2'
|
||||
}
|
||||
|
||||
|
||||
url="/NSImgBDGrille?timestamp=%d"
|
||||
|
||||
color=17
|
||||
|
||||
def __init__(self,basepage):
|
||||
coords = {}
|
||||
coords["01"] = (31,28,49,49)
|
||||
coords["02"] = (108,28,126,49)
|
||||
coords["03"] = (185,28,203,49)
|
||||
coords["04"] = (262,28,280,49)
|
||||
coords["05"] = (339,28,357,49)
|
||||
coords["06"] = (31,100,49,121)
|
||||
coords["07"] = (108,100,126,121)
|
||||
coords["08"] = (185,100,203,121)
|
||||
coords["09"] = (262,100,280,121)
|
||||
coords["10"] = (339,100,357,121)
|
||||
|
||||
VirtKeyboard.__init__(self,basepage.browser.openurl(self.url % time.time()),coords,self.color)
|
||||
self.check_symbols(self.symbols,basepage.browser.responses_dirname)
|
||||
|
||||
def get_symbol_code(self,md5sum):
|
||||
code=VirtKeyboard.get_symbol_code(self,md5sum)
|
||||
return code
|
||||
|
||||
def get_string_code(self,string):
|
||||
code=''
|
||||
for c in string:
|
||||
code+=self.get_symbol_code(self.symbols[c])
|
||||
return code
|
||||
|
||||
|
||||
class LoginPage(BasePage):
|
||||
def on_loaded(self):
|
||||
for td in self.document.getroot().cssselect('td.LibelleErreur'):
|
||||
if td.text is None:
|
||||
continue
|
||||
msg = td.text.strip()
|
||||
if 'indisponible' in msg:
|
||||
raise BrowserUnavailable(msg)
|
||||
|
||||
def login(self, login, password):
|
||||
try:
|
||||
vk=HelloBankVirtKeyboard(self)
|
||||
except VirtKeyboardError,err:
|
||||
self.logger.error("Error: %s"%err)
|
||||
return False
|
||||
|
||||
self.browser.select_form('logincanalnet')
|
||||
self.browser.set_all_readonly(False)
|
||||
self.browser['ch1'] = login.encode('utf-8')
|
||||
self.browser['ch5'] = vk.get_string_code(password)
|
||||
self.browser.submit()
|
||||
|
||||
|
||||
class ConfirmPage(BasePage):
|
||||
def get_error(self):
|
||||
for td in self.document.xpath('//td[@class="hdvon1"]'):
|
||||
if td.text:
|
||||
return td.text.strip()
|
||||
return None
|
||||
|
||||
def get_relocate_url(self):
|
||||
script = self.document.xpath('//script')[0]
|
||||
m = re.match('document.location.replace\("(.*)"\)', script.text[script.text.find('document.location.replace'):])
|
||||
if m:
|
||||
return m.group(1)
|
||||
|
||||
|
||||
class InfoMessagePage(BasePage):
|
||||
def on_loaded(self):
|
||||
pass
|
||||
|
||||
82
modules/hellobank/perso/messages.py
Executable file
82
modules/hellobank/perso/messages.py
Executable file
|
|
@ -0,0 +1,82 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2012 Laurent Bachelier
|
||||
#
|
||||
# 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 weboob.tools.browser import BasePage, BrokenPageError
|
||||
from weboob.capabilities.messages import Message, Thread
|
||||
from weboob.capabilities.base import NotLoaded
|
||||
from weboob.tools.capabilities.messages.genericArticle import try_drop_tree
|
||||
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
||||
from lxml.html import make_links_absolute
|
||||
|
||||
__all__ = ['MessagesPage', 'MessagePage']
|
||||
|
||||
|
||||
class MessagesPage(BasePage):
|
||||
def iter_threads(self):
|
||||
table = self.parser.select(self.document.getroot(), 'table#listeMessages', 1)
|
||||
for tr in table.xpath('./tr'):
|
||||
if tr.attrib.get('class', '') not in ('msgLu', 'msgNonLu'):
|
||||
continue
|
||||
author = unicode(self.parser.select(tr, 'td.colEmetteur', 1).text)
|
||||
link = self.parser.select(tr, 'td.colObjet a', 1)
|
||||
date_raw = self.parser.select(tr, 'td.colDate1', 1).attrib['data']
|
||||
jsparams = re.search('\((.+)\)', link.attrib['onclick']).groups()[0]
|
||||
jsparams = [i.strip('\'" ') for i in jsparams.split(',')]
|
||||
page_id, _id, unread = jsparams
|
||||
# this means unread on the website
|
||||
unread = False if unread == "false" else True
|
||||
# 2012/02/29:01h30min45sec
|
||||
dt_match = re.match('(\d+)/(\d+)/(\d+):(\d+)h(\d+)min(\d+)sec', date_raw).groups()
|
||||
dt_match = [int(d) for d in dt_match]
|
||||
thread = Thread(_id)
|
||||
thread._link_id = (page_id, unread)
|
||||
thread.date = datetime(*dt_match)
|
||||
thread.title = unicode(link.text)
|
||||
message = Message(thread, 0)
|
||||
message.set_empty_fields(None)
|
||||
message.flags = message.IS_HTML
|
||||
message.title = thread.title
|
||||
message.date = thread.date
|
||||
message.sender = author
|
||||
message.content = NotLoaded # This is the only thing we are missing
|
||||
thread.root = message
|
||||
yield thread
|
||||
|
||||
|
||||
class MessagePage(BasePage):
|
||||
def get_content(self):
|
||||
"""
|
||||
Get the message content.
|
||||
This page has a date, but it is less precise than the main list page,
|
||||
so we only use it for the message content.
|
||||
"""
|
||||
try:
|
||||
content = self.parser.select(self.document.getroot(),
|
||||
'div.txtMessage div.contenu', 1)
|
||||
except BrokenPageError:
|
||||
# This happens with some old messages (2007)
|
||||
content = self.parser.select(self.document.getroot(), 'div.txtMessage', 1)
|
||||
|
||||
content = make_links_absolute(content, self.url)
|
||||
try_drop_tree(self.parser, content, 'script')
|
||||
return self.parser.tostring(content)
|
||||
96
modules/hellobank/perso/transactions.py
Executable file
96
modules/hellobank/perso/transactions.py
Executable file
|
|
@ -0,0 +1,96 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 Christophe Lampin
|
||||
# Copyright(C) 2009-2012 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/>.
|
||||
|
||||
|
||||
import re
|
||||
|
||||
from weboob.tools.browser import BasePage
|
||||
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
|
||||
|
||||
|
||||
__all__ = ['AccountHistory', 'AccountComing']
|
||||
|
||||
|
||||
class Transaction(FrenchTransaction):
|
||||
PATTERNS = [(re.compile(u'^(?P<category>CHEQUE)(?P<text>.*)'), FrenchTransaction.TYPE_CHECK),
|
||||
(re.compile('^(?P<category>FACTURE CARTE) DU (?P<dd>\d{2})(?P<mm>\d{2})(?P<yy>\d{2}) (?P<text>.*?)( CA?R?T?E? ?\d*X*\d*)?$'),
|
||||
FrenchTransaction.TYPE_CARD),
|
||||
(re.compile('^(?P<category>(PRELEVEMENT|TELEREGLEMENT|TIP)) (?P<text>.*)'),
|
||||
FrenchTransaction.TYPE_ORDER),
|
||||
(re.compile('^(?P<category>ECHEANCEPRET)(?P<text>.*)'), FrenchTransaction.TYPE_LOAN_PAYMENT),
|
||||
(re.compile('^(?P<category>RETRAIT DAB) (?P<dd>\d{2})/(?P<mm>\d{2})/(?P<yy>\d{2})( (?P<HH>\d+)H(?P<MM>\d+))? (?P<text>.*)'),
|
||||
FrenchTransaction.TYPE_WITHDRAWAL),
|
||||
(re.compile('^(?P<category>VIR(EMEN)?T? ((RECU|FAVEUR) TIERS|SEPA RECU)?)( /FRM)?(?P<text>.*)'),
|
||||
FrenchTransaction.TYPE_TRANSFER),
|
||||
(re.compile('^(?P<category>REMBOURST) CB DU (?P<dd>\d{2})(?P<mm>\d{2})(?P<yy>\d{2}) (?P<text>.*)'),
|
||||
FrenchTransaction.TYPE_PAYBACK),
|
||||
(re.compile('^(?P<category>REMBOURST)(?P<text>.*)'), FrenchTransaction.TYPE_PAYBACK),
|
||||
(re.compile('^(?P<category>COMMISSIONS)(?P<text>.*)'), FrenchTransaction.TYPE_BANK),
|
||||
(re.compile('^(?P<text>(?P<category>REMUNERATION).*)'), FrenchTransaction.TYPE_BANK),
|
||||
(re.compile('^(?P<category>REMISE CHEQUES)(?P<text>.*)'), FrenchTransaction.TYPE_DEPOSIT),
|
||||
]
|
||||
|
||||
|
||||
class AccountHistory(BasePage):
|
||||
def iter_operations(self):
|
||||
for tr in self.document.xpath('//table[@id="tableCompte"]//tr'):
|
||||
if len(tr.xpath('td[@class="debit"]')) == 0:
|
||||
continue
|
||||
|
||||
id = tr.find('td').find('input').attrib['id'].lstrip('_')
|
||||
op = Transaction(id)
|
||||
op.parse(date=tr.findall('td')[1].text,
|
||||
raw=tr.findall('td')[2].text.replace(u'\xa0', u''))
|
||||
|
||||
debit = tr.xpath('.//td[@class="debit"]')[0].text
|
||||
credit = tr.xpath('.//td[@class="credit"]')[0].text
|
||||
|
||||
op.set_amount(credit, debit)
|
||||
|
||||
yield op
|
||||
|
||||
def iter_coming_operations(self):
|
||||
i = 0
|
||||
for tr in self.document.xpath('//table[@id="tableauOperations"]//tr'):
|
||||
if 'typeop' in tr.attrib:
|
||||
tds = tr.findall('td')
|
||||
if len(tds) != 3:
|
||||
continue
|
||||
|
||||
text = tds[1].text or u''
|
||||
text = text.replace(u'\xa0', u'')
|
||||
for child in tds[1].getchildren():
|
||||
if child.text:
|
||||
text += child.text
|
||||
if child.tail:
|
||||
text += child.tail
|
||||
|
||||
i += 1
|
||||
operation = Transaction(i)
|
||||
operation.parse(date=tr.attrib['dateop'],
|
||||
raw=text)
|
||||
operation.set_amount(tds[2].text)
|
||||
yield operation
|
||||
|
||||
def get_IBAN(self):
|
||||
return self.document.xpath('//a[@class="lien_perso_libelle"]')[0].attrib['id'][10:26]
|
||||
|
||||
class AccountComing(AccountHistory):
|
||||
pass
|
||||
111
modules/hellobank/perso/transfer.py
Executable file
111
modules/hellobank/perso/transfer.py
Executable file
|
|
@ -0,0 +1,111 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright(C) 2013 Christophe Lampin
|
||||
# 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/>.
|
||||
|
||||
|
||||
import re
|
||||
|
||||
from weboob.tools.browser import BasePage, BrowserPasswordExpired
|
||||
from weboob.tools.ordereddict import OrderedDict
|
||||
from weboob.capabilities.bank import TransferError
|
||||
|
||||
|
||||
__all__ = ['TransferPage', 'TransferConfirmPage', 'TransferCompletePage']
|
||||
|
||||
|
||||
class Account(object):
|
||||
def __init__(self, id, label, send_checkbox, receive_checkbox):
|
||||
self.id = id
|
||||
self.label = label
|
||||
self.send_checkbox = send_checkbox
|
||||
self.receive_checkbox = receive_checkbox
|
||||
|
||||
|
||||
class TransferPage(BasePage):
|
||||
def on_loaded(self):
|
||||
for td in self.document.xpath('//td[@class="hdvon1"]'):
|
||||
if td.text and 'Vous avez atteint le seuil de' in td.text:
|
||||
raise BrowserPasswordExpired(td.text.strip())
|
||||
|
||||
def get_accounts(self):
|
||||
accounts = OrderedDict()
|
||||
first = True
|
||||
for table in self.document.xpath('//table[@class="tableCompte"]'):
|
||||
if first:
|
||||
first = False
|
||||
for tr in table.cssselect('tr.hdoc1, tr.hdotc1'):
|
||||
tds = tr.findall('td')
|
||||
id = tds[1].text.replace(u'\xa0', u'')
|
||||
label = tds[0].text.replace(u'\xa0', u' ')
|
||||
if label is None and tds[0].find('nobr') is not None:
|
||||
label = tds[0].find('nobr').text
|
||||
send_checkbox = tds[4].find('input').attrib['value'] if tds[4].find('input') is not None else None
|
||||
receive_checkbox = tds[5].find('input').attrib['value'] if tds[5].find('input') is not None else None
|
||||
account = Account(id, label, send_checkbox, receive_checkbox)
|
||||
accounts[id] = account
|
||||
return accounts
|
||||
|
||||
def transfer(self, from_id, to_id, amount, reason):
|
||||
accounts = self.get_accounts()
|
||||
|
||||
# Transform RIBs to short IDs
|
||||
if len(str(from_id)) == 23:
|
||||
from_id = str(from_id)[5:21]
|
||||
if len(str(to_id)) == 23:
|
||||
to_id = str(to_id)[5:21]
|
||||
|
||||
try:
|
||||
sender = accounts[from_id]
|
||||
except KeyError:
|
||||
raise TransferError('Account %s not found' % from_id)
|
||||
|
||||
try:
|
||||
recipient = accounts[to_id]
|
||||
except KeyError:
|
||||
raise TransferError('Recipient %s not found' % to_id)
|
||||
|
||||
if sender.send_checkbox is None:
|
||||
raise TransferError('Unable to make a transfer from %s' % sender.label)
|
||||
if recipient.receive_checkbox is None:
|
||||
raise TransferError('Unable to make a transfer to %s' % recipient.label)
|
||||
|
||||
self.browser.select_form(nr=0)
|
||||
self.browser['C1'] = [sender.send_checkbox]
|
||||
self.browser['C2'] = [recipient.receive_checkbox]
|
||||
self.browser['T6'] = str(amount).replace('.', ',')
|
||||
if reason:
|
||||
self.browser['T5'] = reason.encode('utf-8')
|
||||
self.browser.submit()
|
||||
|
||||
|
||||
class TransferConfirmPage(BasePage):
|
||||
def on_loaded(self):
|
||||
for td in self.document.getroot().cssselect('td#size2'):
|
||||
raise TransferError(td.text.strip())
|
||||
|
||||
for a in self.document.getiterator('a'):
|
||||
m = re.match('/NSFR\?Action=VIRDA&stp=(\d+)', a.attrib['href'])
|
||||
if m:
|
||||
self.browser.location('/NS_VIRDA?stp=%s' % m.group(1))
|
||||
return
|
||||
|
||||
|
||||
class TransferCompletePage(BasePage):
|
||||
def get_id(self):
|
||||
return self.group_dict['id']
|
||||
38
modules/hellobank/test.py
Executable file
38
modules/hellobank/test.py
Executable file
|
|
@ -0,0 +1,38 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# 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 weboob.tools.test import BackendTest
|
||||
from random import choice
|
||||
|
||||
|
||||
class HelloBankTest(BackendTest):
|
||||
BACKEND = 'hellobank'
|
||||
|
||||
def test_bank(self):
|
||||
l = list(self.backend.iter_accounts())
|
||||
if len(l) > 0:
|
||||
a = l[0]
|
||||
list(self.backend.iter_coming(a))
|
||||
list(self.backend.iter_history(a))
|
||||
|
||||
def test_msgs(self):
|
||||
threads = list(self.backend.iter_threads())
|
||||
thread = self.backend.fillobj(choice(threads), ['root'])
|
||||
assert len(thread.root.content)
|
||||
Loading…
Add table
Add a link
Reference in a new issue