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

rewrite module to browser2 (based on creditmutuel)

Browse source
This commit is contained in:
Romain Bignon 2015-07-05 17:17:55 +02:00
commit 3ad6c0add7
3 changed files with 298 additions and 302 deletions
Download patch file Download diff file Expand all files Collapse all files

174
modules/cic/browser.py
View file

@ -18,105 +18,104 @@
# along with weboob. If not, see <http://www.gnu.org/licenses/>. # along with weboob. If not, see <http://www.gnu.org/licenses/>.
try:
from urlparse import urlsplit, parse_qsl, urlparse from urlparse import urlsplit, parse_qsl, urlparse
from datetime import datetime, timedelta except ImportError:
from urllib.parse import urlsplit, parse_qsl, urlparse
from weboob.deprecated.browser import Browser, BrowserIncorrectPassword from datetime import datetime, timedelta
from random import randint
from weboob.tools.compat import basestring
from weboob.browser.browsers import LoginBrowser, need_login
from weboob.browser.profiles import Wget
from weboob.browser.url import URL
from weboob.exceptions import BrowserIncorrectPassword
from weboob.capabilities.bank import Transfer, TransferError from weboob.capabilities.bank import Transfer, TransferError
from .pages import LoginPage, LoginErrorPage, AccountsPage, UserSpacePage, EmptyPage, \ from .pages import LoginPage, LoginErrorPage, AccountsPage, UserSpacePage, \
OperationsPage, CardPage, ComingPage, NoOperationsPage, InfoPage, \ OperationsPage, CardPage, ComingPage, NoOperationsPage, \
TransfertPage, ChangePasswordPage, VerifCodePage TransfertPage, ChangePasswordPage, VerifCodePage, EmptyPage
__all__ = ['CICBrowser'] __all__ = ['CICBrowser']
# Browser class CICBrowser(LoginBrowser):
class CICBrowser(Browser): PROFILE = Wget()
PROTOCOL = 'https' BASEURL = 'https://www.cic.fr'
DOMAIN = 'www.cic.fr'
CERTHASH = '9f41522275058310a6fb348504daeadd16ae852a686a91383b10ad045da76d29' login = URL('/sb/fr/banques/particuliers/index.html',
ENCODING = 'iso-8859-1' '/(?P<subbank>.*)/fr/$',
USER_AGENT = Browser.USER_AGENTS['wget'] '/(?P<subbank>.*)/fr/banques/accueil.html',
PAGES = {'https://www.cic.fr/.*/fr/banques/particuliers/index.html': LoginPage, '/(?P<subbank>.*)/fr/banques/particuliers/index.html',
'https://www.cic.fr/.*/fr/identification/default.cgi': LoginErrorPage, LoginPage)
'https://www.cic.fr/.*/fr/banque/situation_financiere.cgi': AccountsPage, login_error = URL('/(?P<subbank>.*)/fr/identification/default.cgi', LoginErrorPage)
'https://www.cic.fr/.*/fr/banque/situation_financiere.html': AccountsPage, accounts = URL('/(?P<subbank>.*)/fr/banque/situation_financiere.cgi',
'https://www.cic.fr/.*/fr/banque/espace_personnel.aspx': UserSpacePage, '/(?P<subbank>.*)/fr/banque/situation_financiere.html',
'https://www.cic.fr/.*/fr/banque/mouvements.cgi.*': OperationsPage, AccountsPage)
'https://www.cic.fr/.*/fr/banque/mouvements.html.*': OperationsPage, user_space = URL('/(?P<subbank>.*)/fr/banque/espace_personnel.aspx', UserSpacePage)
'https://www.cic.fr/.*/fr/banque/mvts_instance.cgi.*': ComingPage, operations = URL('/(?P<subbank>.*)/fr/banque/mouvements.cgi.*',
'https://www.cic.fr/.*/fr/banque/nr/nr_devbooster.aspx.*': OperationsPage, '/(?P<subbank>.*)/fr/banque/mouvements.html.*',
'https://www.cic.fr/.*/fr/banque/operations_carte\.cgi.*': CardPage, '/(?P<subbank>.*)/fr/banque/nr/nr_devbooster.aspx.*',
'https://www.cic.fr/.*/fr/banque/CR/arrivee\.asp.*': NoOperationsPage, OperationsPage)
'https://www.cic.fr/.*/fr/banque/BAD.*': InfoPage, coming = URL('/(?P<subbank>.*)/fr/banque/mvts_instance.cgi.*', ComingPage)
'https://www.cic.fr/.*/fr/banque/.*Vir.*': TransfertPage, card = URL('/(?P<subbank>.*)/fr/banque/operations_carte.cgi.*', CardPage)
'https://www.cic.fr/.*/fr/validation/change_password.cgi': ChangePasswordPage, noop = URL('/(?P<subbank>.*)/fr/banque/CR/arrivee.asp.*', NoOperationsPage)
'https://www.cic.fr/.*/fr/validation/verif_code.cgi.*': VerifCodePage, info = URL('/(?P<subbank>.*)/fr/banque/BAD.*', EmptyPage)
'https://www.cic.fr/.*/fr/': EmptyPage, transfert = URL('/(?P<subbank>.*)/fr/banque/virements/vplw_vi.html', EmptyPage)
'https://www.cic.fr/.*/fr/banques/index.html': EmptyPage, transfert_2 = URL('/(?P<subbank>.*)/fr/banque/virements/vplw_cmweb.aspx.*', TransfertPage)
'https://www.cic.fr/.*/fr/banque/paci_beware_of_phishing.html.*': EmptyPage, change_pass = URL('/(?P<subbank>.*)/fr/validation/change_password.cgi', ChangePasswordPage)
'https://www.cic.fr/.*/fr/validation/(?!change_password|verif_code).*': EmptyPage, verify_pass = URL('/(?P<subbank>.*)/fr/validation/verif_code.cgi.*', VerifCodePage)
} empty = URL('/(?P<subbank>.*)/fr/banques/index.html',
'/(?P<subbank>.*)/fr/banque/paci_beware_of_phishing.*',
'/(?P<subbank>.*)/fr/validation/(?!change_password|verif_code).*',
'/(?P<subbank>.*)/fr/banque/paci_engine/static_content_manager.aspx',
'/(?P<subbank>.*)/fr/banque/DELG_Gestion.*',
EmptyPage)
currentSubBank = None currentSubBank = None
def is_logged(self): __states__ = ['currentSubBank']
return not self.is_on_page(LoginPage) and not self.is_on_page(LoginErrorPage)
def home(self): def do_login(self):
return self.location('https://www.cic.fr/sb/fr/banques/particuliers/index.html') self.login.go()
def login(self):
assert isinstance(self.username, basestring)
assert isinstance(self.password, basestring)
if not self.is_on_page(LoginPage):
self.location('https://www.cic.fr/', no_login=True)
if not self.page.logged:
self.page.login(self.username, self.password) self.page.login(self.username, self.password)
if not self.is_logged() or self.is_on_page(LoginErrorPage): if not self.page.logged or self.login_error.is_here():
raise BrowserIncorrectPassword() raise BrowserIncorrectPassword()
self.getCurrentSubBank() self.getCurrentSubBank()
@need_login
def get_accounts_list(self): def get_accounts_list(self):
if not self.is_on_page(AccountsPage): return self.accounts.stay_or_go(subbank=self.currentSubBank).iter_accounts()
self.location('https://www.cic.fr/%s/fr/banque/situation_financiere.cgi' % self.currentSubBank)
return self.page.get_list()
def get_account(self, id): def get_account(self, id):
assert isinstance(id, basestring) assert isinstance(id, basestring)
l = self.get_accounts_list() for a in self.get_accounts_list():
for a in l:
if a.id == id: if a.id == id:
return a return a
return None
def getCurrentSubBank(self): def getCurrentSubBank(self):
# the account list and history urls depend on the sub bank of the user # the account list and history urls depend on the sub bank of the user
url = urlparse(self.geturl()) url = urlparse(self.url)
self.currentSubBank = url.path.lstrip('/').split('/')[0] self.currentSubBank = url.path.lstrip('/').split('/')[0]
def list_operations(self, page_url): def list_operations(self, page_url):
if page_url.startswith('/') or page_url.startswith('https'): if page_url.startswith('/') or page_url.startswith('https'):
self.location(page_url) self.location(page_url)
else: else:
self.location('https://%s/%s/fr/banque/%s' % (self.DOMAIN, self.currentSubBank, page_url)) self.location('%s/%s/fr/banque/%s' % (self.BASEURL, self.currentSubBank, page_url))
go_next = True if not self.operations.is_here():
while go_next: return iter([])
if not self.is_on_page(OperationsPage):
return
for op in self.page.get_history(): return self.pagination(lambda: self.page.get_history())
yield op
go_next = self.page.go_next()
def get_history(self, account): def get_history(self, account):
transactions = [] transactions = []
@ -129,7 +128,7 @@ class CICBrowser(Browser):
elif last_debit is None: elif last_debit is None:
last_debit = (tr.date - timedelta(days=10)).month last_debit = (tr.date - timedelta(days=10)).month
coming_link = self.page.get_coming_link() if self.is_on_page(OperationsPage) else None coming_link = self.page.get_coming_link() if self.operations.is_here() else None
if coming_link is not None: if coming_link is not None:
for tr in self.list_operations(coming_link): for tr in self.list_operations(coming_link):
transactions.append(tr) transactions.append(tr)
@ -154,44 +153,53 @@ class CICBrowser(Browser):
def transfer(self, account, to, amount, reason=None): def transfer(self, account, to, amount, reason=None):
# access the transfer page # access the transfer page
transfert_url = 'WI_VPLV_VirUniSaiCpt.asp?RAZ=ALL&Cat=6&PERM=N&CHX=A' self.transfert.go(subbank=self.currentSubBank)
self.location('https://%s/%s/fr/banque/%s' % (self.DOMAIN, self.currentSubBank, transfert_url))
# fill the form # fill the form
self.select_form(name='FormVirUniSaiCpt') form = self.page.get_form(xpath="//form[@id='P:F']")
self['IDB'] = [account[-1]] try:
self['ICR'] = [to[-1]] form['data_input_indiceCompteADebiter'] = self.page.get_from_account_index(account)
self['MTTVIR'] = '%s' % str(amount).replace('.', ',') form['data_input_indiceCompteACrediter'] = self.page.get_to_account_index(to)
except ValueError as e:
raise TransferError(e.message)
form['[t:dbt%3adouble;]data_input_montant_value_0_'] = '%s' % str(amount).replace('.', ',')
if reason is not None: if reason is not None:
self['LIBDBT'] = reason form['[t:dbt%3astring;x(27)]data_input_libelleCompteDebite'] = reason
self['LIBCRT'] = reason form['[t:dbt%3astring;x(31)]data_input_motifCompteCredite'] = reason
self.submit() del form['_FID_GoCancel']
del form['_FID_DoValidate']
form['_FID_DoValidate.x'] = str(randint(3, 125))
form['_FID_DoValidate.y'] = str(randint(3, 22))
form.submit()
# look for known errors # look for known errors
content = unicode(self.response().get_data(), self.ENCODING) content = self.page.get_unicode_content()
insufficient_amount_message = u'Montant insuffisant.' insufficient_amount_message = u'Le montant du virement doit être positif, veuillez le modifier'
maximum_allowed_balance_message = u'Solde maximum autorisé dépassé.' maximum_allowed_balance_message = u'Montant maximum autorisé au débit pour ce compte'
if content.find(insufficient_amount_message) != -1: if insufficient_amount_message in content:
raise TransferError('The amount you tried to transfer is too low.') raise TransferError('The amount you tried to transfer is too low.')
if content.find(maximum_allowed_balance_message) != -1: if maximum_allowed_balance_message in content:
raise TransferError('The maximum allowed balance for the target account has been / would be reached.') raise TransferError('The maximum allowed balance for the target account has been / would be reached.')
# look for the known "all right" message # look for the known "all right" message
ready_for_transfer_message = u'Confirmez un virement entre vos comptes' ready_for_transfer_message = u'Confirmer un virement entre vos comptes'
if not content.find(ready_for_transfer_message): if ready_for_transfer_message not in content:
raise TransferError('The expected message "%s" was not found.' % ready_for_transfer_message) raise TransferError('The expected message "%s" was not found.' % ready_for_transfer_message)
# submit the confirmation form # submit the confirmation form
self.select_form(name='FormVirUniCnf') form = self.page.get_form(xpath="//form[@id='P:F']")
del form['_FID_DoConfirm']
form['_FID_DoConfirm.x'] = str(randint(3, 125))
form['_FID_DoConfirm.y'] = str(randint(3, 22))
submit_date = datetime.now() submit_date = datetime.now()
self.submit() form.submit()
# look for the known "everything went well" message # look for the known "everything went well" message
content = unicode(self.response().get_data(), self.ENCODING) content = self.page.get_unicode_content()
transfer_ok_message = u'Votre virement a été exécuté ce jour' transfer_ok_message = u'Votre virement a &#233;t&#233; ex&#233;cut&#233;'
if not content.find(transfer_ok_message): if transfer_ok_message not in content:
raise TransferError('The expected message "%s" was not found.' % transfer_ok_message) raise TransferError('The expected message "%s" was not found.' % transfer_ok_message)
# We now have to return a Transfer object # We now have to return a Transfer object

8
modules/cic/module.py
View file

@ -1,6 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright(C) 2010-2011 Julien Veyssier # Copyright(C) 2010-2011 Julien Veyssier
# Copyright(C) 2012-2013 Romain Bignon
# #
# This file is part of weboob. # This file is part of weboob.
# #
@ -33,8 +34,8 @@ __all__ = ['CICModule']
class CICModule(Module, CapBank): class CICModule(Module, CapBank):
NAME = 'cic' NAME = 'cic'
MAINTAINER = u'Romain Bignon' MAINTAINER = u'Julien Veyssier'
EMAIL = 'romain@weboob.org' EMAIL = 'julien.veyssier@aiur.fr'
VERSION = '1.1' VERSION = '1.1'
DESCRIPTION = u'CIC' DESCRIPTION = u'CIC'
LICENSE = 'AGPLv3+' LICENSE = 'AGPLv3+'
@ -57,13 +58,11 @@ class CICModule(Module, CapBank):
raise AccountNotFound() raise AccountNotFound()
def iter_coming(self, account): def iter_coming(self, account):
with self.browser:
for tr in self.browser.get_history(account): for tr in self.browser.get_history(account):
if tr._is_coming: if tr._is_coming:
yield tr yield tr
def iter_history(self, account): def iter_history(self, account):
with self.browser:
for tr in self.browser.get_history(account): for tr in self.browser.get_history(account):
if not tr._is_coming: if not tr._is_coming:
yield tr yield tr
@ -88,5 +87,4 @@ class CICModule(Module, CapBank):
except (AssertionError, ValueError): except (AssertionError, ValueError):
raise AccountNotFound() raise AccountNotFound()
with self.browser:
return self.browser.transfer(account, to, amount, reason) return self.browser.transfer(account, to, amount, reason)

398
modules/cic/pages.py
View file

@ -18,58 +18,85 @@
# along with weboob. If not, see <http://www.gnu.org/licenses/>. # along with weboob. If not, see <http://www.gnu.org/licenses/>.
import urllib try:
from urlparse import urlparse, parse_qs from urlparse import urlparse, parse_qs
from decimal import Decimal except ImportError:
from urllib.parse import urlparse, parse_qs
from decimal import Decimal, InvalidOperation
import re import re
from dateutil.relativedelta import relativedelta from dateutil.relativedelta import relativedelta
from weboob.deprecated.browser import Page, BrowserIncorrectPassword, BrokenPageError from weboob.browser.pages import HTMLPage, FormNotFound, LoggedPage
from weboob.tools.ordereddict import OrderedDict from weboob.browser.elements import ListElement, ItemElement, SkipItem, method
from weboob.browser.filters.standard import Filter, Env, CleanText, CleanDecimal, Field, TableCell
from weboob.browser.filters.html import Link
from weboob.exceptions import BrowserIncorrectPassword, ParseError
from weboob.capabilities import NotAvailable
from weboob.capabilities.bank import Account from weboob.capabilities.bank import Account
from weboob.tools.capabilities.bank.transactions import FrenchTransaction from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.tools.date import parse_french_date from weboob.tools.date import parse_french_date
class LoginPage(Page): class LoginPage(HTMLPage):
REFRESH_MAX = 10.0
def login(self, login, passwd): def login(self, login, passwd):
self.browser.select_form(name='ident') form = self.get_form(name='ident')
self.browser['_cm_user'] = login.encode(self.browser.ENCODING) form['_cm_user'] = login
self.browser['_cm_pwd'] = passwd.encode(self.browser.ENCODING) form['_cm_pwd'] = passwd
self.browser.submit(nologin=True) form.submit()
@property
def logged(self):
return self.doc.xpath('//div[@id="e_identification_ok"]')
class LoginErrorPage(Page): class LoginErrorPage(HTMLPage):
pass pass
class ChangePasswordPage(Page): class EmptyPage(LoggedPage, HTMLPage):
def on_loaded(self): REFRESH_MAX = 10.0
class UserSpacePage(LoggedPage, HTMLPage):
pass
class ChangePasswordPage(LoggedPage, HTMLPage):
def on_load(self):
raise BrowserIncorrectPassword('Please change your password') raise BrowserIncorrectPassword('Please change your password')
class VerifCodePage(Page): class VerifCodePage(LoggedPage, HTMLPage):
def on_loaded(self): def on_load(self):
raise BrowserIncorrectPassword('Unable to login: website asks a code from a card') raise BrowserIncorrectPassword('Unable to login: website asks a code from a card')
class InfoPage(Page): class TransfertPage(LoggedPage, HTMLPage):
pass def get_account_index(self, direction, account):
for div in self.doc.getroot().cssselect(".dw_dli_contents"):
inp = div.cssselect("input")[0]
if inp.name != direction:
continue
acct = div.cssselect("span.doux")[0].text.replace(" ", "")
if account.endswith(acct):
return inp.attrib['value']
else:
raise ValueError("account %s not found" % account)
def get_from_account_index(self, account):
return self.get_account_index('data_input_indiceCompteADebiter', account)
def get_to_account_index(self, account):
return self.get_account_index('data_input_indiceCompteACrediter', account)
def get_unicode_content(self):
return self.content.decode(self.detect_encoding())
class EmptyPage(Page): class AccountsPage(LoggedPage, HTMLPage):
pass
class TransfertPage(Page):
pass
class UserSpacePage(Page):
pass
class AccountsPage(Page):
TYPES = {'C/C': Account.TYPE_CHECKING, TYPES = {'C/C': Account.TYPE_CHECKING,
'Livret': Account.TYPE_SAVINGS, 'Livret': Account.TYPE_SAVINGS,
'Pret': Account.TYPE_LOAN, 'Pret': Account.TYPE_LOAN,
@ -78,94 +105,88 @@ class AccountsPage(Page):
'Compte Epargne': Account.TYPE_SAVINGS, 'Compte Epargne': Account.TYPE_SAVINGS,
} }
def get_list(self): @method
accounts = OrderedDict() class iter_accounts(ListElement):
item_xpath = '//tr'
flush_at_end = True
for tr in self.document.getiterator('tr'): class item(ItemElement):
if not tr.getchildren(): klass = Account
continue
first_td = tr.getchildren()[0]
if (first_td.attrib.get('class', '') == 'i g' \
or first_td.attrib.get('class', '') == 'p g' \
or 'i _c1' in first_td.attrib.get('class', '') \
or 'p _c1' in first_td.attrib.get('class', '')) \
and first_td.find('a') is not None:
a = first_td.find('a') def condition(self):
link = a.get('href', '') if len(self.el.xpath('./td')) < 2:
return False
first_td = self.el.xpath('./td')[0]
return (("i" in first_td.attrib.get('class', '') or "p" in first_td.attrib.get('class', ''))
and first_td.find('a') is not None)
class Label(Filter):
def filter(self, text):
return text.lstrip(' 0123456789').title()
class Type(Filter):
def filter(self, label):
for pattern, actype in AccountsPage.TYPES.iteritems():
if label.startswith(pattern):
return actype
return Account.TYPE_UNKNOWN
obj_id = Env('id')
obj_label = Label(CleanText('./td[1]/a/node()[not(contains(@class, "doux"))]'))
obj_coming = Env('coming')
obj_balance = Env('balance')
obj_currency = FrenchTransaction.Currency('./td[2] | ./td[3]')
obj__link_id = Link('./td[1]/a')
obj__card_links = []
obj_type = Type(Field('label'))
def parse(self, el):
link = el.xpath('./td[1]/a')[0].get('href', '')
if link.startswith('POR_SyntheseLst'): if link.startswith('POR_SyntheseLst'):
continue raise SkipItem()
url = urlparse(link) url = urlparse(link)
p = parse_qs(url.query) p = parse_qs(url.query)
if 'rib' not in p: if 'rib' not in p and 'webid' not in p:
continue raise SkipItem()
for i in (2,1): for td in el.xpath('./td[2] | ./td[3]'):
if tr.getchildren()[i].text is None: try:
if not tr.getchildren()[i].getchildren(): balance = CleanDecimal('.', replace_dots=True)(td)
except InvalidOperation:
continue continue
amout = tr.getchildren()[i].getchildren()[0].text
else: else:
amout = tr.getchildren()[i].text
balance = FrenchTransaction.clean_amount(amout)
currency = Account.get_currency(amout)
if len(balance) > 0:
break break
balance = Decimal(balance) else:
raise ParseError('Unable to find balance for account %s' % CleanText('./td[1]/a')(el))
id = p['rib'][0] id = p['rib'][0] if 'rib' in p else p['webid'][0]
if id in accounts:
account = accounts[id] # Handle cards
if id in self.parent.objects:
account = self.parent.objects[id]
if not account.coming: if not account.coming:
account.coming = Decimal('0.0') account.coming = Decimal('0.0')
account.coming += balance account.coming += balance
account._card_links.append(link) account._card_links.append(link)
continue raise SkipItem()
account = Account() self.env['id'] = id
account.id = id
if len(a.getchildren()) > 0: # Handle real balances
account.label = u' '.join([unicode(c.text) for c in a.getchildren()]) page = self.page.browser.open(link).page
else: coming = page.find_amount(u"Opérations à venir") if page else None
account.label = unicode(a.text) accounting = page.find_amount(u"Solde comptable") if page else None
account.label.strip().lstrip(' 0123456789').title()
for pattern, actype in self.TYPES.iteritems():
if account.label.startswith(pattern):
account.type = actype
account._link_id = link
account._card_links = []
# Find accounting amount
page = self.browser.get_document(self.browser.openurl(link))
coming = self.find_amount(page, u"Opérations à venir")
accounting = self.find_amount(page, u"Solde comptable")
if accounting is not None and accounting + (coming or Decimal('0')) != balance: if accounting is not None and accounting + (coming or Decimal('0')) != balance:
self.logger.warning('%s + %s != %s' % (accounting, coming, balance)) self.page.logger.warning('%s + %s != %s' % (accounting, coming, balance))
if accounting is not None: if accounting is not None:
balance = accounting balance = accounting
if coming is not None: self.env['balance'] = balance
account.coming = coming self.env['coming'] = coming or NotAvailable
account.balance = balance
account.currency = currency
accounts[account.id] = account
return accounts.itervalues()
def find_amount(self, page, title):
try:
td = page.xpath(u'//th[contains(text(), "%s")]/../td' % title)[0]
except IndexError:
return None
else:
return Decimal(FrenchTransaction.clean_amount(td.text))
class Transaction(FrenchTransaction): class Transaction(FrenchTransaction):
@ -183,146 +204,115 @@ class Transaction(FrenchTransaction):
_is_coming = False _is_coming = False
class OperationsPage(Page): class Pagination(object):
def get_history(self): def next_page(self):
seen = set() try:
for tr in self.document.getiterator('tr'): form = self.page.get_form('//form[@id="paginationForm"]')
# columns can be: except FormNotFound:
# - date | value | operation | debit | credit | contre-valeur return
# - date | value | operation | debit | credit
# - date | operation | debit | credit
# That's why we skip any extra columns, and take operation, debit
# and credit from last instead of first indexes.
tds = tr.getchildren()[:5]
if len(tds) < 4:
continue
if tds[0].attrib.get('class', '') == 'i g' or \ text = CleanText.clean(form.el)
tds[0].attrib.get('class', '') == 'p g' or \ m = re.search(u'(\d+) / (\d+)', text or '', flags=re.MULTILINE)
tds[0].attrib.get('class', '').endswith('_c1 c _c1') or \ if not m:
tds[0].attrib.get('class', '').endswith('_c1 i _c1'): return
operation = Transaction(0)
parts = [txt.strip() for txt in tds[-3].itertext() if len(txt.strip()) > 0] cur = int(m.group(1))
last = int(m.group(2))
if cur == last:
return
form['page'] = str(cur + 1)
return form.request
class OperationsPage(LoggedPage, HTMLPage):
@method
class get_history(Pagination, Transaction.TransactionsElement):
head_xpath = '//table[@class="liste"]//thead//tr/th'
item_xpath = '//table[@class="liste"]//tbody/tr'
class item(Transaction.TransactionElement):
condition = lambda self: len(self.el.xpath('./td')) >= 4 and len(self.el.xpath('./td[@class="i g" or @class="p g" or contains(@class, "_c1")]')) > 0
class OwnRaw(Filter):
def __call__(self, item):
parts = [txt.strip() for txt in TableCell('raw')(item)[0].itertext() if len(txt.strip()) > 0]
# To simplify categorization of CB, reverse order of parts to separate # To simplify categorization of CB, reverse order of parts to separate
# location and institution. # location and institution.
if parts[0].startswith('PAIEMENT CB'): if parts[0].startswith('PAIEMENT CB'):
parts.reverse() parts.reverse()
date = tds[0].text return u' '.join(parts)
vdate = tds[1].text if len(tds) >= 5 else None
raw = u' '.join(parts)
operation.parse(date=date, vdate=vdate, raw=raw) obj_raw = Transaction.Raw(OwnRaw())
credit = self.parser.tocleanstring(tds[-1]) def find_amount(self, title):
debit = self.parser.tocleanstring(tds[-2]) try:
operation.set_amount(credit, debit) td = self.doc.xpath(u'//th[contains(text(), "%s")]/../td' % title)[0]
operation.id = operation.unique_id(seen) except IndexError:
yield operation return None
else:
def go_next(self): return Decimal(FrenchTransaction.clean_amount(td.text))
form = self.document.xpath('//form[@id="paginationForm"]')
if len(form) == 0:
return False
form = form[0]
text = self.parser.tocleanstring(form)
m = re.search(u'(\d+) / (\d+)', text or '', flags=re.MULTILINE)
if not m:
return False
cur = int(m.group(1))
last = int(m.group(2))
if cur == last:
return False
inputs = {}
for elm in form.xpath('.//input[@type="input"]'):
key = elm.attrib['name']
value = elm.attrib['value']
inputs[key] = value
inputs['page'] = str(cur + 1)
self.browser.location(form.attrib['action'], urllib.urlencode(inputs))
return True
def get_coming_link(self): def get_coming_link(self):
try: try:
a = self.parser.select(self.document, u'//a[contains(text(), "Opérations à venir")]', 1, 'xpath') a = self.doc.xpath(u'//a[contains(text(), "Opérations à venir")]')[0]
except BrokenPageError: except IndexError:
return None return None
else: else:
return a.attrib['href'] return a.attrib['href']
class ComingPage(OperationsPage): class ComingPage(OperationsPage, LoggedPage):
def get_history(self): @method
index = 0 class get_history(Pagination, Transaction.TransactionsElement):
for tr in self.document.xpath('//table[@class="liste"]/tbody/tr'): head_xpath = '//table[@class="liste"]//thead//tr/th/text()'
tds = tr.findall('td') item_xpath = '//table[@class="liste"]//tbody/tr'
if len(tds) < 3:
continue
tr = Transaction(index) col_date = u"Date de l'annonce"
date = self.parser.tocleanstring(tds[0]) class item(Transaction.TransactionElement):
raw = self.parser.tocleanstring(tds[1]) obj__is_coming = True
amount = self.parser.tocleanstring(tds[-1])
tr.parse(date=date, raw=raw)
tr.set_amount(amount)
tr._is_coming = True
yield tr
class CardPage(OperationsPage): class CardPage(OperationsPage, LoggedPage):
def get_history(self): @method
index = 0 class get_history(Pagination, ListElement):
class list_cards(ListElement):
item_xpath = '//table[@class="liste"]/tbody/tr/td/a'
# Check if this is a multi-cards page class item(ItemElement):
pages = [] def __iter__(self):
for a in self.document.xpath('//table[@class="liste"]/tbody/tr/td/a'): card_link = self.el.get('href')
card_link = a.get('href') history_url = '%s/%s/fr/banque/%s' % (self.page.browser.BASEURL, self.page.browser.currentSubBank, card_link)
history_url = 'https://%s/%s/fr/banque/%s' % (self.browser.DOMAIN, self.browser.currentSubBank, card_link) page = self.page.browser.location(history_url).page
page = self.browser.get_document(self.browser.openurl(history_url))
pages.append(page)
if len(pages) == 0: for op in page.get_history():
# If not, add this page as transactions list yield op
pages.append(self.document)
for page in pages: class list_history(Transaction.TransactionsElement):
label = self.parser.tocleanstring(self.parser.select(page.getroot(), 'div.lister p.c', 1)) head_xpath = '//table[@class="liste"]//thead/tr/th'
item_xpath = '//table[@class="liste"]/tbody/tr'
def parse(self, el):
label = CleanText('//div[contains(@class, "lister")]//p[@class="c"]')(el)
if not label:
return
label = re.findall('(\d+ [^ ]+ \d+)', label)[-1] label = re.findall('(\d+ [^ ]+ \d+)', label)[-1]
# use the trick of relativedelta to get the last day of month. # use the trick of relativedelta to get the last day of month.
debit_date = parse_french_date(label) + relativedelta(day=31) self.env['debit_date'] = parse_french_date(label) + relativedelta(day=31)
for tr in page.xpath('//table[@class="liste"]/tbody/tr'): class item(Transaction.TransactionElement):
tds = tr.findall('td')[:4] condition = lambda self: len(self.el.xpath('./td')) >= 4
if len(tds) < 4:
continue
tr = Transaction(index) obj_raw = Transaction.Raw('./td[last()-2] | ./td[last()-1]')
obj_type = Transaction.TYPE_CARD
parts = [txt.strip() for txt in list(tds[-3].itertext()) + list(tds[-2].itertext()) if len(txt.strip()) > 0] obj_date = Env('debit_date')
obj_rdate = Transaction.Date(TableCell('date'))
tr.parse(date=tds[0].text.strip(' \xa0'),
raw=u' '.join(parts))
tr.date = debit_date
tr.type = tr.TYPE_CARD
# Don't take all of the content (with tocleanstring for example),
# because there is a span.aide.
tr.set_amount(tds[-1].text)
yield tr
class NoOperationsPage(OperationsPage): class NoOperationsPage(OperationsPage, LoggedPage):
def get_history(self): def get_history(self):
return iter([]) return iter([])