move old code for old site in a new folder

This commit is contained in:
smurail 2015-05-05 11:05:41 +02:00 committed by Romain Bignon
commit eb2fd26332
11 changed files with 1 additions and 1 deletions

View file

View file

@ -0,0 +1,324 @@
# -*- coding: utf-8 -*-
# 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
from datetime import datetime
from logging import warning
from random import randint
from weboob.deprecated.browser import Browser, BrowserIncorrectPassword, BrowserPasswordExpired
from weboob.capabilities.bank import TransferError, Transfer, Account
from .perso.accounts_list import AccountsList, AccountPrelevement
from .perso.transactions import AccountHistory, AccountComing
from .perso.investment import AccountMarketInvestment, AccountLifeInsuranceInvestment
from .perso.transfer import TransferPage, TransferConfirmPage, TransferCompletePage
from .perso.login import LoginPage, ConfirmPage, ChangePasswordPage, InfoMessagePage
from .perso.messages import MessagePage, MessagesPage
from .pro import ProAccountsList, ProAccountHistory
__all__ = ['BNPorc']
class BNPorc(Browser):
DOMAIN = 'www.secure.bnpparibas.net'
PROTOCOL = 'https'
CERTHASH = ['5511f0ff19c982b6351c17b901bfa7419f075edb13f2df41e446248beb7866bb', 'fa8cb72ef2e46054469af916f7ec222b1904901fecde8511a0f769ba0385410d', '86cd4ba8cfbc53937dfc402e8c2d0a2d5ffb630a73bbeafd09c39f8b54a6a6c3']
ENCODING = None # refer to the HTML encoding
PAGES = {'.*pageId=unedescomptes.*': AccountsList,
'.*pageId=releveoperations.*': AccountHistory,
'.*FicheA': AccountHistory,
'.*SAF_DPF.*': AccountMarketInvestment,
'.*identifiant=Assurance_Vie_Consultation.*': AccountLifeInsuranceInvestment,
'.*Action=SAF_CHM.*': ChangePasswordPage,
'.*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,
'.*MessagesRecus.*': MessagesPage,
'.*BmmFicheLireMessage.*': MessagePage,
# Pro
'https?://www.secure.bnpparibas.net/banque/portail/entrepros/Fiche\?.*identifiant=PRO_Une_Comptes.*': ProAccountsList,
'https?://www.secure.bnpparibas.net/SAF_ROP.*': ProAccountHistory,
'https?://www.secure.bnpparibas.net/NS_AVEDT.*': ProAccountHistory,
}
def __init__(self, *args, **kwargs):
self.rotating_password = kwargs.pop('rotating_password', None)
self.password_changed_cb = kwargs.pop('password_changed_cb', None)
Browser.__init__(self, *args, **kwargs)
def home(self):
self.location('https://www.secure.bnpparibas.net/banque/portail/particulier/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()
#self.readurl('/SAF_SOA?Action=6')
def change_password(self, new_password):
assert new_password.isdigit() and len(new_password) == 6
buf = self.readurl('https://www.secure.bnpparibas.net/NSFR?Action=SAF_CHM', if_fail='raise')
buf = buf[buf.find('/SAF_CHM?Action=SAF_CHM'):]
buf = buf[:buf.find('"')]
self.location(buf)
assert self.is_on_page(ChangePasswordPage)
#self.readurl('/banque/portail/particulier/bandeau')
#self.readurl('/common/vide.htm')
self.page.change_password(self.password, new_password)
if not self.is_on_page(ConfirmPage) or self.page.get_error() is not None:
self.logger.error('Oops, unable to change password (%s)'
% (self.page.get_error() if self.is_on_page(ConfirmPage) else 'unknown'))
return
self.password, self.rotating_password = (new_password, self.password)
if self.password_changed_cb:
self.password_changed_cb(self.rotating_password, self.password)
def check_expired_password(func):
def inner(self, *args, **kwargs):
try:
return func(self, *args, **kwargs)
except BrowserPasswordExpired:
if self.rotating_password is not None:
warning('[%s] Your password has expired. Switching...' % self.username)
self.change_password(self.rotating_password)
return func(self, *args, **kwargs)
else:
raise
return inner
@check_expired_password
def get_accounts_list(self):
if not self.is_on_page(AccountsList):
self.location('/NSFR?Action=DSP_VGLOBALE')
return self.page.get_list()
def get_account(self, id):
assert isinstance(id, basestring)
if not self.is_on_page(AccountsList):
self.location('/NSFR?Action=DSP_VGLOBALE')
l = self.page.get_list()
for a in l:
if a.id == id:
return a
return None
def iter_history(self, account):
if account._link_id is None:
return iter([])
if account._stp is not None:
# Pro
self.location(self.buildurl('/SAF_ROP', Origine='DSP_HISTOCPT', ch4=account._link_id, stp=account._stp))
else:
# Perso
if not self.is_on_page(AccountsList):
self.location('/NSFR?Action=DSP_VGLOBALE')
execution = self.page.document.xpath('//form[@name="goToApplication"]/input[@name="execution"]')[0].attrib['value']
data = {'gt': 'homepage:basic-theme',
'externalIAId': 'IAStatements',
'cboFlowName': 'flow/iastatement',
'contractId': account._link_id,
'groupId': '-2',
'pastOrPendingOperations': 1,
'groupSelected':'-2',
'step': 'STAMENTS',
'pageId': 'releveoperations',
#'operationsPerPage': 100,
#'_eventId': 'changeOperationsPerPage',
'sendEUD': 'true',
'execution': execution,
}
self.location('https://www.secure.bnpparibas.net/banque/portail/particulier/FicheA', urllib.urlencode(data))
execution = self.page.document.xpath('//form[@name="displayStatementForm"]/input[@name="_flowExecutionKey"]')[0].attrib['value']
data = {'_eventId': 'changeOperationsPerPage',
'newCategoryId': '',
'categorisationInProgress': '',
'contractId': account._link_id,
'_flowExecutionKey': execution,
'groupId': '-2',
'operations.objectsPerPage': 100,
'operations.pageNumber': 1,
'pageId': 'releveoperations',
}
# it's not a joke, BNP guys are really crappy.
for i in xrange(30):
data['_operations.list[%d].checkedOff' % i] = 'on'
data['_operations.list[%d].selectedForCategorization' % i] = 'on'
self.location('https://www.secure.bnpparibas.net/banque/portail/particulier/FicheA', urllib.urlencode(data))
return self.page.iter_operations()
def iter_coming_operations(self, account):
if account._link_id is None:
return iter([])
if account._stp is not None:
# Pro
self.location(self.buildurl('/NS_AVEDT', Origine='DSP_DT', ch4=account._link_id, stp=account._stp))
else:
# Perso
if not self.is_on_page(AccountsList):
self.location('/NSFR?Action=DSP_VGLOBALE')
execution = self.page.document.xpath('//form[@name="goToApplication"]/input[@name="execution"]')[0].attrib['value']
data = {'gt': 'homepage:basic-theme',
'externalIAId': 'IAStatements',
'cboFlowName': 'flow/iastatement',
'contractId': account._link_id,
'groupId': '-2',
'pastOrPendingOperations': 2,
'groupSelected':'-2',
'step': 'STAMENTS',
'pageId': 'mouvementsavenir',
#'operationsPerPage': 100,
#'_eventId': 'changeOperationsPerPage',
'sendEUD': 'true',
'execution': execution,
}
self.location('https://www.secure.bnpparibas.net/banque/portail/particulier/FicheA', urllib.urlencode(data))
return self.page.iter_coming_operations()
def iter_investment(self, account):
if account.type == Account.TYPE_MARKET:
if not account.iban:
return iter([])
stp = datetime.strftime(datetime.now(), '%Y%m%d%H%M%S')
self.location('https://www.secure.bnpparibas.net/SAF_DPF?Origine=DSP_DPF&ch4=' + account.iban + 'stp=' + stp)
data = {'ch4': account.iban,
'Origine': 'DSP_DPF',
'chD': 'oui',
'chT': 'sans',
'chU': 'alpha',
'x': randint(0, 99),
'y': randint(0, 18)
}
self.location('https://www.secure.bnpparibas.net/SAF_DPF', urllib.urlencode(data))
elif account.type == Account.TYPE_LIFE_INSURANCE:
if not account._link_id:
return iter([])
self.location('https://www.secure.bnpparibas.net/banque/portail/particulier/Fiche?type=folder&identifiant=Assurance_Vie_Consultation_20071002051044&contrat=' + account._link_id)
else:
raise NotImplementedError()
return self.page.iter_investment()
@check_expired_password
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()
@check_expired_password
def transfer(self, from_id, to_id, amount, reason=None):
if not self.is_on_page(TransferPage):
self.location('/NS_VIRDF')
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

View file

@ -0,0 +1,133 @@
# -*- coding: utf-8 -*-
# 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.deprecated.browser import Page, BrokenPageError, BrowserPasswordExpired
class AccountsList(Page):
ACCOUNT_TYPES = {
u'Liquidités': Account.TYPE_CHECKING,
u'Epargne disponible': Account.TYPE_SAVINGS,
u'Titres': Account.TYPE_MARKET,
u'Assurance vie': Account.TYPE_LIFE_INSURANCE,
u'Crédit immobilier': Account.TYPE_LOAN,
}
def on_loaded(self):
pass
def _parse_iban(self, account, url):
m = re.search('ch4=(\w+)', url)
if m:
account.iban = unicode(m.group(1))
def _parse_account_group(self, table):
typename = unicode(table.attrib.get('summary', '').replace('Liste des contrats/comptes ', ''))
typeid = self.ACCOUNT_TYPES.get(typename, Account.TYPE_UNKNOWN)
account = None
for tr in table.xpath('.//tr'):
if tr.find('td') is not None and tr.find('td').attrib.get('class', '') == 'typeTitulaire':
account = self._parse_account(tr)
account.type = typeid
yield account
elif tr.get('class', '') == 'listeActionBig' and account is not None:
try:
url = tr.xpath('.//a')[-1].get('href', '')
except IndexError:
pass
else:
self._parse_iban(account, url)
account = None
def _parse_account(self, tr):
account = Account()
# for pro usage
account._stp = None
account.id = tr.xpath('.//td[@class="libelleCompte"]/input')[0].attrib['id'][len('libelleCompte'):]
if len(str(account.id)) == 23:
account.id = str(account.id)[5:21]
a = tr.xpath('.//td[@class="libelleCompte"]/a')[0]
m = re.match(r'javascript:goToStatements\(\'(\d+)\'', a.get('onclick', ''))
if m:
account._link_id = m.group(1)
else:
# Find _link_id of life insurances
m = re.match(r'javascript:overviewRedirectionOperation.*contrat=(\d+)', a.get('onclick', ''))
if m:
account._link_id = m.group(1)
else:
# Can't get history for this account.
account._link_id = None
# To prevent multiple-IDs for CIF (for example), add an arbitrary char in ID.
account.id += 'C'
account.label = u''+a.text.strip()
tds = tr.findall('td')
account.currency = account.get_currency(tds[3].find('a').text)
account.balance = self._parse_amount(tds[3].find('a'))
if tds[4].find('a') is not None:
account.coming = self._parse_amount(tds[4].find('a'))
else:
account.coming = NotAvailable
return account
def _parse_amount(self, elem):
return Decimal(FrenchTransaction.clean_amount(elem.text))
def get_list(self):
accounts = []
for table in self.document.xpath('//table[@class="tableCompte"]'):
for account in self._parse_account_group(table):
accounts.append(account)
if len(accounts) == 0:
# 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 accounts
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.
"""
for link in self.parser.select(self.document.getroot(), 'div#pantalon div.interieur a'):
if 'MessagesRecus' in link.attrib.get('href', ''):
return link.attrib['href']
raise BrokenPageError('Unable to find the link to the messages page')
class AccountPrelevement(AccountsList):
pass

View file

@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
# 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 decimal import Decimal
from xml.etree import ElementTree
from weboob.deprecated.browser import Page
from weboob.capabilities.bank import Investment
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.capabilities.base import NotAvailable
def clean_text(el):
text = ElementTree.tostring(el, 'utf-8', 'text').decode('utf-8')
return re.sub(ur'[\s\xa0]+', u' ', text).strip()
def clean_cells(cells):
return list(map(clean_text, cells))
def clean_amount(amount):
return Decimal(FrenchTransaction.clean_amount(amount)) if amount else NotAvailable
def clean_amounts(amounts):
return list(map(clean_amount, amounts))
class AccountMarketInvestment(Page):
def iter_investment(self):
table = self.document.xpath('//table[@align="center"]')[4]
rows = table.xpath('.//tr[@class="hdoc1"]')
for tr in rows:
cells = clean_cells(tr.findall('td'))
cells[2:] = clean_amounts(cells[2:])
inv = Investment()
inv.label, _, inv.quantity, inv.unitvalue, inv.valuation = cells
tr2 = tr.xpath('./following-sibling::tr')[0]
tr2td = tr2.findall('td')[1]
inv.id = inv.code = clean_text(tr2.xpath('.//a')[0])
inv.unitprice = clean_amount(tr2td.xpath('.//td[@class="hdotc1nb"]')[0].text)
inv.description = u''
inv.diff = inv.quantity * inv.unitprice - inv.valuation
yield inv
class AccountLifeInsuranceInvestment(Page):
def iter_investment(self):
rows = self.document.xpath('//table[@id="mefav_repartition_supports_BPF"]//tr') or \
self.document.xpath('//tbody[@id="mefav_repartition_supports"]//tr')
for tr in rows:
cells = clean_cells(tr.findall('td'))
cells[3:] = clean_amounts(cells[3:])
inv = Investment()
inv.label, _, inv.code, inv.quantity, inv.unitvalue, inv.valuation, _ = cells
if inv.code:
inv.id = inv.code
if not inv.unitvalue:
# XXX Fonds eu Euros
inv.code = u'XX' + re.sub(ur'[^A-Za-z0-9]', u'', inv.label).upper()
inv.description = u''
yield inv

View file

@ -0,0 +1,162 @@
# -*- coding: utf-8 -*-
# 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
import urllib
from weboob.deprecated.browser import Page, BrowserUnavailable
from weboob.tools.captcha.virtkeyboard import VirtKeyboard, VirtKeyboardError
class BNPVirtKeyboard(VirtKeyboard):
symbols={'0': '9cc4789a2cb223e8f2d5e676e90264b5',
'1': 'e10b58fc085f9683052d5a63c96fc912',
'2': '04ec647e7b3414bcc069f0c54eb55a4c',
'3': 'fde84fd9bac725db8463554448f1e469',
'4': '2359eea8671bf112b58264bec0294f71',
'5': '82b55b63480114f04fad8c5c4fa5673a',
'6': 'e074864faeaeabb3be3d118192cd8879',
'7': 'af5740e4ca71fadc6f4ae1412d864a1c',
'8': 'cab759c574038ad89a0e35cc76ab7214',
'9': '828cf0faf86ac78e7f43208907620527'
}
url="/NSImgGrille?timestamp=%d"
color=27
def __init__(self, page):
coords = {}
size = 136
x, y, width, height = (0, 0, size/5, size/5)
for a in page.document.xpath('//div[@id="secret-nbr-keyboard"]/a'):
code = a.attrib['ondblclick']
coords[code] = (x+1, y+1, x+height-2, y+height-2)
if (x + width + 1) >= size:
y += height
x = 0
else:
x += width
VirtKeyboard.__init__(self, page.browser.openurl(self.url % time.time()), coords, self.color)
self.check_symbols(self.symbols, page.browser.responses_dirname)
def get_symbol_code(self, md5sum):
code = VirtKeyboard.get_symbol_code(self, md5sum)
return re.sub(u'[^\d]', '', code)
def get_string_code(self, string):
code = ''
for c in string:
code += self.get_symbol_code(self.symbols[c])
return code
class LoginPage(Page):
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=BNPVirtKeyboard(self)
except VirtKeyboardError as err:
self.logger.error("Error: %s"%err)
return False
# Mechanize does not recognize the form..
form = self.document.xpath('//form[@name="logincanalnet"]')[0]
url = form.attrib['action']
params = {}
for ctrl in form.findall('input'):
params[ctrl.attrib['name']] = ctrl.attrib['value']
params['ch1'] = login.encode('iso-8859-1')
params['ch5'] = vk.get_string_code(password)
self.browser.location(url, urllib.urlencode(params))
class ConfirmPage(Page):
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(Page):
def on_loaded(self):
pass
class ChangePasswordPage(Page):
def change_password(self, current, new):
try:
vk=BNPVirtKeyboard(self)
except VirtKeyboardError as err:
self.logger.error("Error: %s"%err)
return False
from mechanize import Cookie
c = Cookie(0, 'wbo_segment_369721', 'AA%7CAB%7C%7C%7C',
None, False,
'.secure.bnpparibas.net', True, True,
'/', False,
False,
None,
False,
None,
None,
{})
cookiejar = self.browser._ua_handlers["_cookies"].cookiejar
cookiejar.set_cookie(c)
code_current=vk.get_string_code(current)
code_new=vk.get_string_code(new)
data = (('ch1', code_current),
('ch2', code_new),
('radiobutton3', 'radiobutton'),
('ch3', code_new),
('x', 23),
('y', 13),
)
headers = {'Referer': self.url}
#headers = {'Referer': "https://www.secure.bnpparibas.net/SAF_CHM?Action=SAF_CHM&Origine=SAF_CHM&stp=%s" % (int(datetime.now().strftime('%Y%m%d%H%M%S')))}
#import time
#time.sleep(10)
request = self.browser.request_class('https://www.secure.bnpparibas.net/SAF_CHM_VALID', urllib.urlencode(data), headers)
self.browser.location(request)

View file

@ -0,0 +1,80 @@
# -*- 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.deprecated.browser import Page, 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
class MessagesPage(Page):
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(Page):
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)

View file

@ -0,0 +1,94 @@
# -*- coding: utf-8 -*-
# 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.deprecated.browser import Page
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
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>PRLV( EUROPEEN)? SEPA) (?P<text>.*?)( MDT/.*?)?( ECH/\d+)?( ID .*)?$'),
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+))?( \d+)? (?P<text>.*)'),
FrenchTransaction.TYPE_WITHDRAWAL),
(re.compile('^(?P<category>VIR(EMEN)?T? (RECU |FAVEUR )?(TIERS )?)\w+ \d+/\d+ \d+H\d+ \w+ (?P<text>.*)$'),
FrenchTransaction.TYPE_TRANSFER),
(re.compile('^(?P<category>VIR(EMEN)?T? (EUROPEEN )?(SEPA )?(RECU |FAVEUR |EMIS )?(TIERS )?)(/FRM |/DE |/MOTIF |/BEN )?(?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(Page):
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
class AccountComing(AccountHistory):
pass

View file

@ -0,0 +1,105 @@
# -*- 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/>.
import re
from weboob.deprecated.browser import Page, BrowserPasswordExpired
from weboob.tools.ordereddict import OrderedDict
from weboob.capabilities.bank import TransferError
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(Page):
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()
for table in self.document.getiterator('table'):
if table.attrib.get('cellspacing') == '2':
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
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(Page):
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(Page):
def get_id(self):
return self.group_dict['id']

View file

@ -0,0 +1,142 @@
# -*- coding: utf-8 -*-
# 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 re
from urlparse import urlparse, parse_qsl
from decimal import Decimal, InvalidOperation
from weboob.capabilities import NotAvailable
from weboob.capabilities.bank import Account
from weboob.deprecated.browser import Page
from .perso.transactions import Transaction
class ProAccountsList(Page):
COL_LABEL = 1
COL_ID = 2
COL_BALANCE = 3
COL_COMING = 5
def get_list(self, pro=True):
accounts = []
for tr in self.document.xpath('//tr[@class="comptes"]'):
cols = tr.findall('td')
if len(cols) < 5:
continue
account = Account()
account.id = self.parser.tocleanstring(cols[self.COL_ID]).replace(" ", "")
account.label = self.parser.tocleanstring(cols[self.COL_LABEL])
account.balance = Decimal(self.parser.tocleanstring(cols[self.COL_BALANCE]))
try:
account.coming = Decimal(self.parser.tocleanstring(cols[self.COL_COMING]))
except InvalidOperation:
if self.parser.tocleanstring(cols[self.COL_COMING]) != '-':
self.logger.warning('Unable to parse coming value', exc_info=True)
account.coming = NotAvailable
account._link_id = None
account._stp = None
a = cols[self.COL_LABEL].find('a')
if a is not None:
url = urlparse(a.attrib['href'])
p = dict(parse_qsl(url.query))
account._link_id = p.get('ch4', None)
account._stp = p.get('stp', None)
for input_tag in tr.xpath('.//input[starts-with(@id, "urlRib")]'):
m = re.search('ch4=(\w+)', input_tag.get('value', ''))
if m:
account.iban = unicode(m.group(1))
accounts.append(account)
# If there are also personnal accounts linked, display the page and iter on them.
if pro and len(self.document.xpath('//div[@class="onglets"]//a[contains(@href, "afficherComptesPrives")]')) > 0:
self.browser.select_form(name='myForm')
self.browser.set_all_readonly(False)
self.browser['udcAction'] = '/afficherComptesPrives'
self.browser.submit()
for a in self.browser.page.get_list(False):
accounts.append(a)
return accounts
class ProAccountHistory(Page):
COL_DATE = 0
COL_LABEL = 1
COL_DEBIT = -2
COL_CREDIT = -1
def on_loaded(self):
# If transactions are ordered by type, force order by date.
try:
checkbox = self.document.xpath('//input[@name="szTriDate"]')[0]
except IndexError:
return
if not 'checked' in checkbox.attrib:
self.browser.select_form(name='formtri')
self.browser['szTriDate'] = ['date']
self.browser['szTriRub'] = []
self.browser.submit()
def iter_operations(self):
for i, tr in enumerate(self.document.xpath('//tr[@class="hdoc1" or @class="hdotc1"]')):
cols = tr.findall('td')
if len(cols) < 4:
continue
op = Transaction(i)
date = self.parser.tocleanstring(cols[self.COL_DATE])
raw = self.parser.tocleanstring(cols[self.COL_LABEL])
raw = re.sub(r'[ \xa0]+', ' ', raw).strip()
op.parse(date=date, raw=raw)
debit = self.parser.tocleanstring(cols[self.COL_DEBIT])
credit = self.parser.tocleanstring(cols[self.COL_CREDIT])
op.set_amount(credit, debit)
yield op
def iter_coming_operations(self):
for i, tr in enumerate(self.document.xpath('//tr[@class="hdoc1" or @class="hdotc1"]')):
cols = tr.findall('td')
if len(cols) < 4:
continue
op = Transaction(i)
date = self.parser.tocleanstring(cols[self.COL_DATE])
raw = self.parser.tocleanstring(cols[self.COL_LABEL])
raw = re.sub(r'[ \xa0]+', ' ', raw).strip()
op.parse(date=date, raw=raw)
credit = self.parser.tocleanstring(cols[self.COL_CREDIT])
op.set_amount(credit)
yield op