Skip to content
Commits on Source (27)
......@@ -13,3 +13,4 @@ modules/modules.list
*.idea/
*.DS_Store
*.coverage*
.vscode/
......@@ -368,7 +368,8 @@ def get_history_by_receipt(account, coming, sel_tbl1=None):
self.location('/cyber/internet/Page.do', params=next_params)
if coming and account._coming_count:
for i in range(account._coming_count):
for i in range(account._coming_start,
account._coming_start + account._coming_count):
for tr in get_history_by_receipt(account, coming, sel_tbl1=i):
yield tr
else:
......
......@@ -310,26 +310,27 @@ def login(self, login, passwd):
class MyVirtKeyboard(SplitKeyboard):
char_to_hash = {
'0': '6a2cb38bcfc27781faaec727ad304ce2',
'1': '296140f37a22b5e2b4871272aed22444',
'2': 'c1318fd381665a97e1052f85213867a7',
'3': 'fe19d2cc8f8d09b818b05c2a10218233',
'4': 'd5a03e69857bf01fc373cedbe2530ca9',
'5': '289ae90e4adfa58ef4767d9151c96348',
'6': '88938bbbb6b81ee2a32568f7081be488',
'7': '96499777fb95974ee651f19181de6c01',
'8': '6e2e052c9301d1f381155912ad4d3874',
'9': '5958d54d88bfaa172305575164b39a8d',
'0': 'cce0f72c47c74a3dde57c4fdbcda1db4',
'1': 'f5d22afb3ece4dec73bd8a2a4c2844da',
'2': '6d3e5db2ccac3f2c13c1f0ba22571857',
'3': ('c8e4f6addac4d322f0f9668d472a146c', '34d0566ea3f2330c675365da3178f6ab'),
'4': '8a8c769418ec829c208ed442fbf5fe77',
'5': '2c3ae480bc91f73b431b048b584026c7',
'6': 'a80d639443818e838b434c36dd518df5',
'7': '8e59048702e4c5f89bbbc1a598d06d1e',
'8': '46bc59a5b288c63477ff52811a3961c5',
'9': 'a7bf34568154ef91e990aa5bade3e946',
}
codesep = ' '
def convert(self, buffer):
im = Image.open(BytesIO(buffer))
im = im.resize((5, 8), Image.BILINEAR)
im = im.resize((5, 8), Image.BICUBIC)
im = im.filter(ImageFilter.UnsharpMask(radius=2,
percent=110,
threshold=3))
im = im.convert("P", dither=Image.NONE)
im = im.convert("L", dither=Image.NONE)
im = Image.eval(im, lambda x: 0 if x < 160 else 255)
s = BytesIO()
im.save(s, 'png')
......@@ -682,6 +683,11 @@ def iter_accounts(self, next_pages):
self.logger.debug('there are no cards on this page')
continue
# We are processing another card, so reset account
if CleanText('.')(cols[0]) and account is not None:
yield account
account = None
id = CleanText(None).filter(cols[self.COL_ID])
if len(id) > 0:
if account is not None:
......@@ -700,7 +706,15 @@ def iter_accounts(self, next_pages):
account._coming_params = params.copy()
account._coming_params['dialogActionPerformed'] = 'SELECTION_ENCOURS_CARTE'
account._coming_params['attribute($SEL_$%s)' % tr.attrib['id'].split('_')[0]] = tr.attrib['id'].split('_', 1)[1]
account._coming_count = len(self.doc.xpath('//table[@id="tbl1"]/tbody/tr/td[5]/span[not(contains(text(), "(1)"))]'))
# select current row and next rows till parent name is empty
account._coming_start = int(tr.attrib['id'].split('_', 1)[1])
account._coming_count = 1
for row in tr.xpath('./following-sibling::tr[./td[5]/span[not(contains(text(), "(1)"))]]'):
if CleanText('./td[2]')(row):
break
account._coming_count += 1
elif account is None:
raise BrokenPageError('Unable to find accounts on cards page')
else:
......
......@@ -30,6 +30,7 @@
from .pages import (
LoginPage, HomePage, AccountsPage, OldAccountsPage, HistoryPage, InvestmentPage, InvestDetailPage,
InvestmentListPage, QuestionPage, ChangePassPage, LogonFlowPage, ViewPage, SwitchPage,
HandlePasswordsPage, PostponePasswords,
)
......@@ -61,6 +62,8 @@ class BinckBrowser(LoginBrowser):
r'FsmaMandatoryQuestionnairesOverview', QuestionPage)
change_pass = URL(r'/ChangePassword/Index',
r'/EditSetting/GetSetting\?code=MutationPassword', ChangePassPage)
handle_passwords = URL(r'/PersonalCredentials/Index', HandlePasswordsPage)
postpone_passwords = URL(r'/PersonalCredentials/PostPone', PostponePasswords)
def deinit(self):
if self.page and self.page.logged:
......@@ -94,6 +97,13 @@ def switch_account(self, account_id):
@need_login
def iter_accounts(self):
# If we already know that it is an old website connection,
# we can call old_website_connection() right away.
if self.old_website_connection:
for account in self.iter_old_accounts():
yield account
return
if self.unique_account:
self.account.stay_or_go()
else:
......@@ -121,41 +131,46 @@ def iter_accounts(self):
# so we need to fetch them on the OldAccountsPage for now:
else:
self.old_website_connection = True
self.old_accounts.go()
for a in self.page.iter_accounts():
try:
self.old_accounts.stay_or_go().go_to_account(a.id)
except ServerError as exception:
# get html error to parse
parser = etree.HTMLParser()
html_error = etree.parse(StringIO(exception.response.text), parser)
account_error = html_error.xpath('//p[contains(text(), "Votre compte est")]/text()')
if account_error:
raise ActionNeeded(account_error[0])
else:
raise
a.iban = self.page.get_iban()
# Get token
token = self.page.get_token()
# Get investment page
data = {'grouping': "SecurityCategory"}
try:
a._invpage = self.investment.go(data=data, headers=token) \
if self.page.is_investment() else None
except HTTPNotFound:
# if it's not an invest account, the portfolio link may be present but hidden and return a 404
a._invpage = None
for account in self.iter_old_accounts():
yield account
@need_login
def iter_old_accounts(self):
self.old_accounts.go()
for a in self.page.iter_accounts():
try:
self.old_accounts.stay_or_go().go_to_account(a.id)
except ServerError as exception:
# get html error to parse
parser = etree.HTMLParser()
html_error = etree.parse(StringIO(exception.response.text), parser)
account_error = html_error.xpath('//p[contains(text(), "Votre compte est")]/text()')
if account_error:
raise ActionNeeded(account_error[0])
else:
raise
a.iban = self.page.get_iban()
# Get token
token = self.page.get_token()
# Get investment page
data = {'grouping': "SecurityCategory"}
try:
a._invpage = self.investment.go(data=data, headers=token) \
if self.page.is_investment() else None
except HTTPNotFound:
# if it's not an invest account, the portfolio link may be present but hidden and return a 404
a._invpage = None
if a._invpage:
a.valuation_diff = a._invpage.get_valuation_diff()
# Get history page
data = [('currencyCode', a.currency), ('startDate', ""), ('endDate', "")]
a._histpages = [self.history.go(data=data, headers=token)]
while self.page.doc['EndOfData'] is False:
a._histpages.append(self.history.go(data=self.page.get_nextpage_data(data[:]), headers=token))
if a._invpage:
a.valuation_diff = a._invpage.get_valuation_diff()
# Get history page
data = [('currencyCode', a.currency), ('startDate', ""), ('endDate', "")]
a._histpages = [self.history.go(data=data, headers=token)]
while self.page.doc['EndOfData'] is False:
a._histpages.append(self.history.go(data=self.page.get_nextpage_data(data[:]), headers=token))
yield a
yield a
@need_login
def iter_investment(self, account):
......
......@@ -47,6 +47,13 @@ def on_load(self):
form.submit()
class BinckPage(LoggedPage, HTMLPage):
# Used to factorize the get_token() method
def get_token(self):
return [{Attr('.', 'name')(input): Attr('.', 'value')(input)}
for input in self.doc.xpath('//input[contains(@name, "Token")]')][0]
class ViewPage(LoggedPage, HTMLPage):
# We automatically skip the new website tutorial
def on_load(self):
......@@ -75,6 +82,16 @@ def on_load(self):
raise BrowserPasswordExpired(message)
class HandlePasswordsPage(BinckPage):
def on_load(self):
token = self.get_token()
self.browser.postpone_passwords.go(headers=token, method='POST')
self.browser.home_page.go()
class PostponePasswords(LoggedPage, HTMLPage):
pass
class LogonFlowPage(HTMLPage):
def on_load(self):
raise ActionNeeded(CleanText('//article//h1 | //article//h3')(self.doc))
......@@ -91,7 +108,7 @@ def get_error(self):
return CleanText('//div[contains(@class, "errors")]')(self.doc)
class AccountsPage(LoggedPage, HTMLPage):
class AccountsPage(BinckPage):
TYPES = {'L': Account.TYPE_SAVINGS,
'CT': Account.TYPE_MARKET,
'PEA': Account.TYPE_PEA,
......@@ -103,10 +120,6 @@ class AccountsPage(LoggedPage, HTMLPage):
def has_accounts_table(self):
return self.doc.xpath('//table[contains(@class, "accountoverview-table")]')
def get_token(self):
return [{Attr('.', 'name')(input): Attr('.', 'value')(input)} \
for input in self.doc.xpath('//input[contains(@name, "Token")]')][0]
@method
class iter_accounts(ListElement):
# Tables have no headers so we must use ListElement.
......@@ -135,7 +148,7 @@ def obj_currency(self):
return Account.get_currency(CleanText('.//div[contains(text(), "Total des avoirs")]/following::strong[1]')(self))
class OldAccountsPage(LoggedPage, HTMLPage):
class OldAccountsPage(BinckPage):
'''
Old website accounts page. We can get rid of this
class when all users have access to the new website.
......@@ -154,10 +167,6 @@ def go_to_account(self, number):
def get_iban(self):
return CleanText('//div[@class="iban"]/text()', replace=[(' ', '')], default=NotAvailable)(self.doc)
def get_token(self):
return [{Attr('.', 'name')(input): Attr('.', 'value')(input)} \
for input in self.doc.xpath('//input[contains(@name, "Token")]')][0]
def is_investment(self):
# warning: the link can be present even in case of non-investement account
return CleanText('//a[contains(@href, "Portfolio")]', default=False)(self.doc)
......
......@@ -146,8 +146,8 @@ def _iter_history_base(self, account):
self.logger.debug('skipping coming %r', transaction.to_dict())
continue
history.append(transaction)
for transaction in sorted_transactions(history):
yield transaction
for transaction in sorted_transactions(history):
yield transaction
@need_login
def iter_coming_operations(self, account):
......
......@@ -116,11 +116,6 @@ class iter_history(ListElement):
class item(ItemElement):
klass = Transaction
def obj_id(self):
label = CleanText(Attr('./a[contains(@class, "accordion_collapse")]', "id"))(self)
label = ''.join(i for i in label if i.isdigit())
return label
obj_date = Date(CleanText('./div[contains(@class, "accordion_header")]/div[1]/p'))
obj_category = CleanText('./div[contains(@class, "accordion_header")]/div[2]/p[1]')
obj_label = CleanText('./div[contains(@class, "accordion_header")]/div[3]/p[1]')
......
......@@ -65,7 +65,7 @@ def get_iban(self):
if self.doc.xpath('//div[has-class("alert")]/p[contains(text(), "Une erreur est survenue")]') or \
self.doc.xpath('//div[has-class("alert")]/p[contains(text(), "Le compte est introuvable")]'):
return NotAvailable
return CleanText('//table[thead[tr[th[contains(text(), "Code I.B.A.N")]]]]/tbody/tr/td[2]', replace=[(' ', '')])(self.doc)
return CleanText('//div[strong[contains(text(),"IBAN")]]/div[contains(@class, "definition")]', replace=[(' ', '')])(self.doc)
class AuthenticationPage(HTMLPage):
......
......@@ -331,9 +331,13 @@ class ErrorCodePage(HTMLPage):
def on_load(self):
code = re.search(r'\/\?errorCode=(\d+)', self.url).group(1)
page = self.browser.open('/particuliers/compte-bancaire/comptes-en-ligne/bredconnect-compte-ligne?errorCode=%s' % code).page
# invalid login/password
if code == '20100':
msg = CleanText('//label[contains(@class, "error")]')(page.doc)
msg = CleanText('//label[contains(@class, "error")]', default=None)(page.doc)
# 20100: invalid login/password
# 139: dispobank user trying to connect to Bred
if code in ('20100', '139'):
raise BrowserIncorrectPassword(msg)
# 20104 & 1000: unknown error during login
elif code in ('20104', '1000'):
raise BrowserUnavailable(msg)
assert False, 'The % error is not handled.' % code
assert False, 'Error %s is not handled yet.' % code
......@@ -30,13 +30,16 @@
from weboob.browser.switch import SiteSwitch
from weboob.browser.url import URL
from weboob.capabilities.bank import Account, AddRecipientStep, Recipient, TransferBankError, Transaction, TransferStep
from weboob.capabilities.base import NotAvailable
from weboob.capabilities.base import NotAvailable, find_object
from weboob.capabilities.profile import Profile
from weboob.browser.exceptions import BrowserHTTPNotFound, ClientError, ServerError
from weboob.exceptions import (
BrowserIncorrectPassword, BrowserUnavailable, BrowserHTTPError, BrowserPasswordExpired, ActionNeeded
)
from weboob.tools.capabilities.bank.transactions import sorted_transactions, FrenchTransaction
from weboob.tools.capabilities.bank.transactions import (
sorted_transactions, FrenchTransaction, keep_only_card_transactions,
omit_deferred_transactions,
)
from weboob.tools.capabilities.bank.investments import create_french_liquidity
from weboob.tools.compat import urljoin, urlparse
from weboob.tools.value import Value
......@@ -49,7 +52,7 @@
ProTransferSummaryPage, ProAddRecipientOtpPage, ProAddRecipientPage,
SmsPage, SmsPageOption, SmsRequest, AuthentPage, RecipientPage, CanceledAuth, CaissedepargneKeyboard,
TransactionsDetailsPage, LoadingPage, ConsLoanPage, MeasurePage, NatixisLIHis, NatixisLIInv, NatixisRedirectPage,
SubscriptionPage, CreditCooperatifMarketPage, UnavailablePage,
SubscriptionPage, CreditCooperatifMarketPage, UnavailablePage, CardsPage, CardsComingPage, CardsOldWebsitePage,
)
from .linebourse_browser import LinebourseAPIBrowser
......@@ -81,6 +84,9 @@ class CaisseEpargne(LoginBrowser, StatesMixin):
pro_add_recipient_otp = URL('https://.*/Portail.aspx.*', ProAddRecipientOtpPage)
pro_add_recipient = URL('https://.*/Portail.aspx.*', ProAddRecipientPage)
measure_page = URL('https://.*/Portail.aspx.*', MeasurePage)
cards_old = URL('https://.*/Portail.aspx.*', CardsOldWebsitePage)
cards = URL('https://.*/Portail.aspx.*', CardsPage)
cards_coming = URL('https://.*/Portail.aspx.*', CardsComingPage)
authent = URL('https://.*/Portail.aspx.*', AuthentPage)
subscription = URL('https://.*/Portail.aspx\?tache=(?P<tache>).*', SubscriptionPage)
home = URL('https://.*/Portail.aspx.*', IndexPage)
......@@ -430,6 +436,30 @@ def get_accounts_list(self):
else:
assert False, "new domain that hasn't been seen so far ?"
self.home.go()
self.page.go_list()
self.page.go_cards()
if self.cards.is_here() or self.cards_old.is_here():
cards = list(self.page.iter_cards())
for card in cards:
card.parent = find_object(self.accounts, number=card._parent_id)
assert card.parent, 'card account %r parent was not found' % card
# If we are in the new site, we have to get each card coming transaction link.
if self.cards.is_here():
for card in cards:
info = card.parent._card_links
# If info is filled, that mean there are comings transaction
card._coming_info = None
if info:
self.page.go_list()
self.page.go_history(info)
card._coming_info = self.page.get_card_coming_info(card.number, info.copy())
self.accounts.extend(cards)
# Some accounts have no available balance or label and cause issues
# in the backend so we must exclude them from the accounts list:
self.accounts = [account for account in self.accounts if account.label and account.balance != NotAvailable]
......@@ -445,7 +475,6 @@ def get_loans_list(self):
if self.page.check_no_accounts() or self.page.check_no_loans():
return []
access_to_loans = False
for trial in range(5):
for _ in range(3):
self.home_tache.go(tache='CRESYNT0')
......@@ -461,12 +490,8 @@ def get_loans_list(self):
self.logger.warning('Access to loans failed, we try again')
else:
# We managed to reach the Loans JSON
access_to_loans = True
break
if not access_to_loans:
raise BrowserUnavailable()
for _ in range(3):
try:
self.home_tache.go(tache='CPTSYNT0')
......@@ -480,8 +505,13 @@ def get_loans_list(self):
return iter(self.loans)
# For all account, we fill up the history with transaction. For checking account, there will have
# also deferred_card transaction too.
# From this logic, if we send "account_card", that mean we recover all transactions from the parent
# checking account of the account_card, then we filter later the deferred transaction.
@need_login
def _get_history(self, info):
def _get_history(self, info, account_card=None):
# Only fetch deferred debit card transactions if `account_card` is not None
if isinstance(info['link'], list):
info['link'] = info['link'][0]
if not info['link'].startswith('HISTORIQUE'):
......@@ -500,6 +530,11 @@ def _get_history(self, info):
if 'netpro' in self.page.url and not self.page.is_history_of(info['id']):
self.page.go_history_netpro(info)
# In this case, we want the coming transaction for the new website
# (old website return coming directly in `get_coming()` )
if account_card and info and info['type'] == 'HISTORIQUE_CB':
self.page.go_coming(account_card._coming_info['link'])
info['link'] = [info['link']]
for i in range(self.HISTORY_MAX_PAGE):
......@@ -508,14 +543,25 @@ def _get_history(self, info):
# list of transactions on account page
transactions_list = []
list_form = []
card_and_forms = []
for tr in self.page.get_history():
transactions_list.append(tr)
if tr.type == tr.TYPE_CARD_SUMMARY:
list_form.append(self.page.get_form_to_detail(tr))
if account_card:
if self.card_matches(tr.card, account_card.number):
card_and_forms.append((tr.card, self.page.get_form_to_detail(tr)))
else:
self.logger.debug('will skip summary detail (%r) for different card %r', tr, account_card.number)
# For deferred card history only :
#
# Now that we find transactions that have TYPE_CARD_SUMMARY on the checking account AND the account_card number we want,
# we browse deferred card transactions that are resume by that list of TYPE_CARD_SUMMARY transaction.
# add detail card to list of transactions
for form in list_form:
# Checking account transaction:
# - 01/01 - Summary 5134XXXXXX103 - 900.00€ - TYPE_CARD_SUMMARY <-- We have to go in the form of this tr to get
# cards details transactions.
for card, form in card_and_forms:
form.submit()
if self.home.is_here() and self.page.is_access_error():
self.logger.warning('Access to card details is unavailable for this user')
......@@ -523,6 +569,8 @@ def _get_history(self, info):
assert self.transaction_detail.is_here()
for tr in self.page.get_detail():
tr.type = Transaction.TYPE_DEFERRED_CARD
if account_card:
tr.card = card
transactions_list.append(tr)
if self.new_website:
self.page.go_newsite_back_to_summary()
......@@ -592,6 +640,15 @@ def _get_history_invests(self, account):
def get_history(self, account):
self.home.go()
self.deleteCTX()
if account.type == account.TYPE_CARD:
def match_cb(tr):
return self.card_matches(tr.card, account.number)
hist = self._get_history(account.parent._info, account)
hist = keep_only_card_transactions(hist, match_cb)
return hist
if not hasattr(account, '_info'):
raise NotImplementedError
if account.type is Account.TYPE_LIFE_INSURANCE and 'measure_id' not in account._info:
......@@ -608,18 +665,37 @@ def get_history(self, account):
self.linebourse.session.cookies.update(self.session.cookies)
self.update_linebourse_token()
return self.linebourse.iter_history(account.id)
return self._get_history(account._info)
hist = self._get_history(account._info, False)
return omit_deferred_transactions(hist)
@need_login
def get_coming(self, account):
if account.type != account.TYPE_CARD:
return []
trs = []
if not hasattr(account, '_info'):
if not hasattr(account.parent, '_info'):
raise NotImplementedError()
for info in account._card_links:
for tr in self._get_history(info.copy()):
# We are on the old website
if hasattr(account, '_coming_eventargument'):
if not self.cards_old.is_here():
self.home.go()
self.page.go_list()
self.page.go_cards()
self.page.go_card_coming(account._coming_eventargument)
return sorted_transactions(self.page.iter_coming())
# We are on the new website.
info = account.parent._card_links
# if info is empty, that mean there are no coming yet
if info:
for tr in self._get_history(info.copy(), account):
tr.type = tr.TYPE_DEFERRED_CARD
tr.nopurge = True
trs.append(tr)
return sorted_transactions(trs)
......@@ -706,7 +782,7 @@ def get_profile(self):
@need_login
def iter_recipients(self, origin_account):
if origin_account.type == Account.TYPE_LOAN:
if origin_account.type in [Account.TYPE_LOAN, Account.TYPE_CARD]:
return []
if 'pro' in self.url:
......@@ -933,3 +1009,9 @@ def download_document(self, document):
self.page.go_document_list(sub_id=sub_id)
return self.page.download_document(document).content
def card_matches(self, a, b):
# For the same card, depending where we scrape it, we have
# more or less visible number. `X` are visible number, `*` hidden one's.
# tr.card: XXXX******XXXXXX, account.number: XXXXXX******XXXX
return (a[:4], a[-4:]) == (b[:4], b[-4:])
......@@ -208,7 +208,7 @@ def on_load(self):
if 'OIC_QCF' in self.browser.url:
# QCF is a mandatory test to make sure you know the basics about financials products
# however, you can still choose to postpone it. hence the continue link
link = Link('//span[@id="lea-prdvel-lien"]/p/b/a[contains(text(), "Continuer")]')(self.doc)
link = Link('//span[@id="lea-prdvel-lien"]//b/a[contains(text(), "Continuer")]')(self.doc)
if link:
self.logger.warning("By-passing QCF")
self.browser.location(link)
......@@ -284,7 +284,7 @@ def _get_account_info(self, a, accounts):
def is_account_inactive(self, account_id):
return self.doc.xpath('//tr[td[contains(text(), $id)]][@class="Inactive"]', id=account_id)
def _add_account(self, accounts, link, label, account_type, balance):
def _add_account(self, accounts, link, label, account_type, balance, number=None):
info = self._get_account_info(link, accounts)
if info is None:
self.logger.warning('Unable to parse account %r: %r' % (label, link))
......@@ -295,6 +295,7 @@ def _add_account(self, accounts, link, label, account_type, balance):
if is_rib_valid(info['id']):
account.iban = rib2iban(info['id'])
account._info = info
account.number = number
account.label = label
account.type = self.ACCOUNT_TYPES.get(label, info['acc_type'] if 'acc_type' in info else account_type)
if 'PERP' in account.label:
......@@ -309,16 +310,15 @@ def _add_account(self, accounts, link, label, account_type, balance):
account.currency = account.get_currency(balance) if balance and balance is not NotAvailable else NotAvailable
account._card_links = []
# Set coming history link to the parent account. At this point, we don't have card account yet.
if account._info['type'] == 'HISTORIQUE_CB' and account.id in accounts:
a = accounts[account.id]
if not a.coming:
a.coming = Decimal('0.0')
if account.balance and account.balance is not NotAvailable:
a.coming += account.balance
a._card_links.append(account._info)
a.coming = Decimal('0.0')
a._card_links = account._info
return
accounts[account.id] = account
return account
def get_balance(self, account):
if account.type not in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_PERP, Account.TYPE_CAPITALISATION):
......@@ -368,20 +368,27 @@ def get_list(self):
self.ACCOUNT_TYPES.get(CleanText('.')(tds[2])) or\
self.ACCOUNT_TYPES.get(CleanText('.')(tds[3]), Account.TYPE_UNKNOWN)
else:
# On the same row, there are many accounts (for example a
# check accound and a card one).
# On the same row, there could have many accounts (check account and a card one).
# For the card line, the number will be the same than the checking account, so we skip it.
if len(tds) > 4:
for i, a in enumerate(tds[2].xpath('./a')):
label = CleanText('.')(a)
balance = CleanText('.')(tds[-2].xpath('./a')[i])
self._add_account(accounts, a, label, account_type, balance)
# Only 4 tds on banque de la reunion website.
number = None
# if i > 0, that mean it's a card account. The number will be the same than it's
# checking parent account, we have to skip it.
if i == 0:
number = CleanText('.')(tds[-4].xpath('./a')[0])
self._add_account(accounts, a, label, account_type, balance, number)
# Only 4 tds on "banque de la reunion" website.
elif len(tds) == 4:
for i, a in enumerate(tds[1].xpath('./a')):
label = CleanText('.')(a)
balance = CleanText('.')(tds[-1].xpath('./a')[i])
self._add_account(accounts, a, label, account_type, balance)
self.logger.debug('we are on the %s website', 'old' if accounts else 'new')
if len(accounts) == 0:
# New website
self.browser.new_website = True
......@@ -405,7 +412,9 @@ def get_list(self):
label = CleanText('./strong')(tds[0])
balance = CleanText('.')(tds[-1])
self._add_account(accounts, a, label, account_type, balance)
account = self._add_account(accounts, a, label, account_type, balance)
if account:
account.number = CleanText('.')(tds[1])
return accounts.values()
......@@ -444,6 +453,8 @@ def get_loan_list(self):
account.currency = account.get_currency(CleanText('./a')(tds[4]))
accounts[account.id] = account
self.logger.debug('we are on the %s website', 'old' if accounts else 'new')
if len(accounts) == 0:
# New website
for table in self.doc.xpath('//div[@class="panel"]'):
......@@ -522,23 +533,58 @@ class item(ItemElement):
obj_next_payment_amount = MyDecimal(MyTableCell("next_payment_amount"))
obj_next_payment_date = Date(CleanText(MyTableCell("next_payment_date", default=''), default=NotAvailable), default=NotAvailable)
def submit_form(self, form, eventargument, eventtarget, scriptmanager):
form['__EVENTARGUMENT'] = eventargument
form['__EVENTTARGET'] = eventtarget
form['m_ScriptManager'] = scriptmanager
fix_form(form)
form.submit()
def go_list(self):
form = self.get_form(id='main')
eventargument = "CPTSYNT0"
form['__EVENTARGUMENT'] = "CPTSYNT0"
if "MM$m_CH$IsMsgInit" in form:
# Old website
eventtarget = "Menu_AJAX"
scriptmanager = "m_ScriptManager|Menu_AJAX"
else:
# New website
eventtarget = "MM$m_PostBack"
scriptmanager = "MM$m_UpdatePanel|MM$m_PostBack"
self.submit_form(form, eventargument, eventtarget, scriptmanager)
def go_cards(self):
form = self.get_form(id='main')
eventargument = ""
if "MM$m_CH$IsMsgInit" in form:
# Old website
form['__EVENTTARGET'] = "Menu_AJAX"
form['m_ScriptManager'] = "m_ScriptManager|Menu_AJAX"
eventtarget = "Menu_AJAX"
eventargument = "HISENCB0"
scriptmanager = "m_ScriptManager|Menu_AJAX"
else:
# New website
form['__EVENTTARGET'] = "MM$m_PostBack"
form['m_ScriptManager'] = "MM$m_UpdatePanel|MM$m_PostBack"
eventtarget = "MM$SYNTHESE$btnSyntheseCarte"
scriptmanager = "MM$m_UpdatePanel|MM$SYNTHESE$btnSyntheseCarte"
fix_form(form)
self.submit_form(form, eventargument, eventtarget, scriptmanager)
form.submit()
# only for old website
def go_card_coming(self, eventargument):
form = self.get_form(id='main')
eventtarget = "MM$HISTORIQUE_CB"
scriptmanager = "m_ScriptManager|Menu_AJAX"
self.submit_form(form, eventargument, eventtarget, scriptmanager)
# only for new website
def go_coming(self, eventargument):
form = self.get_form(id='main')
eventtarget = "MM$HISTORIQUE_CB"
scriptmanager = "MM$m_UpdatePanel|MM$HISTORIQUE_CB"
self.submit_form(form, eventargument, eventtarget, scriptmanager)
# On some pages, navigate to indexPage does not lead to the list of measures, so we need this form ...
def go_measure_list(self):
......@@ -679,7 +725,11 @@ def get_history(self):
continue
if 'tot dif' in t.raw.lower():
t._link = Link(tr.xpath('./td/a'))(self.doc)
t.deleted = True
# "Cb" for new site, "CB" for old one
mtc = re.match(r'(Cb|CB) (\d{4}\*+\d{6}) ', raw)
if mtc is not None:
t.card = mtc.group(2)
t.set_amount(credit, debit)
yield t
......@@ -786,6 +836,154 @@ def is_transfer_allowed(self):
return not self.doc.xpath('//ul/li[contains(text(), "Aucun compte tiers n\'est disponible")]')
class CardsPage(IndexPage):
def is_here(self):
return CleanText('//h3[normalize-space(text())="Mes cartes (cartes dont je suis le titulaire)"]')(self.doc)
@method
class iter_cards(TableElement):
head_xpath = '//table[@class="cartes"]/tbody/tr/th'
col_label = 'Carte'
col_number = ''
col_parent = 'Compte dépot associé'
col_coming = 'Encours'
item_xpath = '//table[@class="cartes"]/tbody/tr[not(th)]'
class item(ItemElement):
klass = Account
obj_type = Account.TYPE_CARD
obj_label = Format('%s %s', CleanText(TableCell('label')), Field('id'))
obj_number = CleanText(TableCell('number'))
obj_id = CleanText(TableCell('number'), replace=[('*', 'X')])
obj__parent_id = CleanText(TableCell('parent'))
obj_balance = 0
obj_currency = Currency(TableCell('coming'))
def obj_coming(self):
if CleanText(TableCell('coming'))(self) == '-':
raise SkipItem('immediate debit card?')
return CleanDecimal.French(TableCell('coming'), sign=lambda x: -1)(self)
def condition(self):
immediate_str = ''
# There are some card without any information. To exclude them, we keep only account
# with extra "option" (ex: coming transaction link, block bank card...)
if 'Faire opposition' in CleanText("./td[5]")(self):
# Only deferred card have this option to see coming transaction, even when
# there is 0 coming (Table element have no thead for the 5th column).
if 'Consulter mon encours carte' in CleanText("./td[5]")(self):
return True
# Card without 'Consulter mon encours carte' are immediate card. There are logged
# for now to make the debug easier
immediate_str = '[Immediate card]'
self.logger.warning('Skip card %s (no history/coming information) %s', Field('number')(self), immediate_str)
return False
class CardsComingPage(IndexPage):
def is_here(self):
return CleanText('//h2[text()="Encours de carte à débit différé"]')(self.doc)
def get_card_coming_info(self, number, info):
# If the xpath match, that mean there are only one card
# We have enought information in `info` to get its coming transaction
if CleanText('//tr[@id="MM_HISTORIQUE_CB_rptMois0_ctl01_trItem"]')(self.doc):
return info
# If the xpath match, that mean there are at least 2 cards
xpath = '//tr[@id="MM_HISTORIQUE_CB_rptMois0_trItem_0"]'
# In case of multiple card, first card coming's transactions are reachable
# with information in `info`.
if Regexp(CleanText(xpath), r'(\d{6}\*{6}\d{4})')(self.doc) == number:
return info
# For all card except the first one for the same check account, we have to get info through their href info
link = CleanText(Attr('//a[contains(text(),"%s")]' % number, 'href'))(self.doc)
infos = re.match(r'.*(DETAIL_OP_M0&[^\"]+).*', link)
info['link'] = infos.group(1)
return info
class CardsOldWebsitePage(IndexPage):
def is_here(self):
return CleanText('//span[@id="MM_m_CH_lblTitle" and contains(text(), "Historique de vos encours CB")]')(self.doc)
def get_account(self):
infos = CleanText('.//span[@id="MM_HISTORIQUE_CB"]/table[position()=1]//td')(self.doc)
result = re.search(r'.*(\d{11}).*', infos)
return result.group(1)
def get_date(self):
title = CleanText('//span[@id="MM_HISTORIQUE_CB_m_TableTitle3_lblTitle"]')(self.doc)
title_date = re.match('.*le (.*) sur .*', title)
return Date(dayfirst=True).filter(title_date.group(1))
@method
class iter_cards(TableElement):
head_xpath = '//table[@id="MM_HISTORIQUE_CB_m_ExDGOpeM0"]//tr[@class="DataGridHeader"]/td'
item_xpath = '//table[@id="MM_HISTORIQUE_CB_m_ExDGOpeM0"]//tr[not(contains(@class, "DataGridHeader")) and position() < last()]'
col_label = 'Libellé'
col_coming = 'Solde'
class item(ItemElement):
klass = Account
obj_type = Account.TYPE_CARD
obj_label = Format('%s %s', CleanText(TableCell('label')), CleanText(Field('number')))
obj_balance = 0
obj_coming = CleanDecimal.French(TableCell('coming'))
obj_currency = Currency(TableCell('coming'))
def obj__parent_id(self):
return self.page.get_account()
def obj_number(self):
return CleanText(TableCell('number'))(self).replace('*', 'X')
def obj_id(self):
number = Field('number')(self).replace('X', '')
account_id = '%s-%s' % (self.obj__parent_id(), number)
return account_id
def obj__coming_eventargument(self):
url = Attr('.//a', 'href')(self)
res = re.match(r'.*(DETAIL_OP_M0\&.*;\d{8})", .*', url)
return res.group(1)
def parse(self, obj):
# There are no thead name for this column.
self._cols['number'] = 3
@method
class iter_coming(TableElement):
head_xpath = '//table[@id="MM_HISTORIQUE_CB_m_ExDGDetailOpe"]//tr[@class="DataGridHeader"]/td'
item_xpath = '//table[@id="MM_HISTORIQUE_CB_m_ExDGDetailOpe"]//tr[not(contains(@class, "DataGridHeader"))]'
col_label = 'Libellé'
col_coming = 'Débit'
col_date = 'Date'
class item(ItemElement):
klass = Transaction
obj_type = Transaction.TYPE_DEFERRED_CARD
obj_label = CleanText(TableCell('label'))
obj_amount = CleanDecimal.French(TableCell('coming'))
obj_rdate = Date(CleanText(TableCell('date')), dayfirst=True)
def obj_date(self):
return self.page.get_date()
class ConsLoanPage(JsonPage):
def get_conso(self):
return self.doc
......@@ -992,7 +1190,9 @@ def parse(self, el):
# TODO use after 'I'?
_id = Regexp(CleanText('.'), r'- (\w+\d\w+)')(self) # at least one digit
accounts = list(self.page.browser.get_accounts_list()) + list(self.page.browser.get_loans_list())
match = [acc for acc in accounts if _id in acc.id]
# If it's an internal account, we should always find only one account with _id in it's id.
# Type card account contains their parent account id, and should not be listed in recipient account.
match = [acc for acc in accounts if _id in acc.id and acc.type != Account.TYPE_CARD]
assert len(match) == 1
match = match[0]
self.env['id'] = match.id
......
......@@ -57,7 +57,7 @@ def on_load(self):
# Need to update mail. Try to skip
msg = "Merci de renseigner votre adresse e-mail"
if CleanText('//p[@role="heading" and contains(text(), "%s")]' % msg)(self.doc):
url = Link('//a[contains(., "PASSER CETTE ETAPE")]')(self.doc)
url = Link('//a[contains(., "PASSER CETTE ETAPE")]', default=None)(self.doc)
if url:
self.browser.location(url)
else:
......
......@@ -33,7 +33,7 @@
from weboob.tools.capabilities.bank.transactions import sorted_transactions
from .pages import (
LogoutPage, InfosPage, AccountsPage, HistoryPage, LifeinsurancePage, MarketPage,
LogoutPage, AccountsPage, HistoryPage, LifeinsurancePage, MarketPage,
AdvisorPage, LoginPage, ProfilePage,
)
from .transfer_pages import TransferInfoPage, RecipientsListPage, TransferPage
......@@ -84,7 +84,6 @@ class CmsoParBrowser(LoginBrowser, StatesMixin):
logout = URL('/securityapi/revoke',
'/auth/errorauthn',
'/\/auth/errorauthn', LogoutPage)
infos = URL('/comptes/', InfosPage)
accounts = URL('/domiapi/oauth/json/accounts/synthese(?P<type>.*)', AccountsPage)
history = URL('/domiapi/oauth/json/accounts/(?P<page>.*)', HistoryPage)
loans = URL('/creditapi/rest/oauth/v1/synthese', AccountsPage)
......@@ -171,8 +170,7 @@ def iter_accounts(self):
accounts_eligibilite_debit = self.page.get_eligibilite_debit()
# First get all checking accounts...
data = dict(self.infos.stay_or_go().get_typelist())
self.accounts.go(data=json.dumps(data), type='comptes', headers=self.json_headers)
self.accounts.go(json={'typeListeCompte': 'COMPTE_SOLDE_COMPTES_CHEQUES'}, type='comptes')
self.page.check_response()
for key in self.page.get_keys():
for a in self.page.iter_accounts(key=key):
......
......@@ -55,15 +55,6 @@ class LogoutPage(RawPage):
pass
class InfosPage(LoggedPage, HTMLPage):
def get_typelist(self):
url = Attr(None, 'src').filter(self.doc.xpath('//script[contains(@src, "comptes/scripts")]'))
m = re.findall(r'synthesecomptes[^\w]+([^:]+)[^\w]+([^"]+)', self.browser.open(url).text)
for data in m:
if data[0] != 'method':
return {data[0]: data[1]}
class AccountsPage(LoggedPage, JsonPage):
TYPES = OrderedDict([('courant', Account.TYPE_CHECKING),
('pee', Account.TYPE_PEE),
......
......@@ -29,7 +29,7 @@
from weboob.capabilities.bank import Account, Investment
from weboob.capabilities.base import NotAvailable
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.tools.compat import urljoin
from weboob.tools.compat import urljoin, parse_qsl
from weboob.tools.capabilities.bank.investments import is_isin_valid
......@@ -284,9 +284,13 @@ def get_sso_url(self):
class TokenPage(CMSOPage, UpdateTokenMixin):
def on_load(self):
d = re.search(r'id_token=(?P<id_token>[^&]+)&access_token=(?P<access_token>[^&]+)', self.text).groupdict()
self.browser.token = d['id_token']
self.browser.csrf = d['access_token']
auth_query_params = re.search(r'parent\.location = ".*#(.*)";', self.text)
assert auth_query_params, 'Url query parameter with token for authentication was not found'
auth_query_params = auth_query_params.group(1)
params = dict(parse_qsl(auth_query_params))
self.browser.token = params['id_token']
self.browser.csrf = params['access_token']
class AuthCheckUser(HTMLPage):
......
......@@ -139,6 +139,7 @@ class item_account_generic(ItemElement):
('C/C', Account.TYPE_CHECKING),
('Start', Account.TYPE_CHECKING),
('Comptes courants', Account.TYPE_CHECKING),
('Service Accueil', Account.TYPE_CHECKING),
('Catip', Account.TYPE_DEPOSIT),
('Cic Immo', Account.TYPE_LOAN),
('Credit', Account.TYPE_LOAN),
......
......@@ -145,7 +145,7 @@ def execute_transfer(self, transfer, **params):
return self.browser.execute_transfer(transfer)
def transfer_check_label(self, old, new):
old = re.sub(r"[/<\?='!\+:]", '', old).strip()
old = re.sub(r"[/<\?='!\+:#]", '', old).strip()
old = old.encode('latin-1', errors='replace').decode('latin-1')
# if no reason given, the site changes the label
if not old and ("INTERNET-FAVEUR" in new):
......
......@@ -191,7 +191,8 @@ class AMFAmundiPage(HTMLPage):
CODE_TYPE = Investment.CODE_TYPE_AMF
def get_code(self):
return Regexp(CleanText('//td[@class="bannerColumn"]//li[contains(., "(C)")]', default=NotAvailable), r'(\d+)')(self.doc)
return Regexp(CleanText('//td[@class="bannerColumn"]//li[contains(., "(C)")]', default=NotAvailable),
r'(\d+)', default=NotAvailable)(self.doc)
class AMFSGPage(HTMLPage):
......@@ -384,6 +385,8 @@ def condition(self):
obj_label = Env('label')
def obj_type(self):
if Field('label')(self).startswith('ETOILE'):
return self.page.TYPES.get(Field('label')(self).split()[1].upper(), Account.TYPE_UNKNOWN)
return self.page.TYPES.get(Field('label')(self).split()[0].upper(), Account.TYPE_UNKNOWN)
def obj_balance(self):
......
......@@ -26,7 +26,7 @@
from weboob.browser.url import URL
from weboob.browser.exceptions import ClientError
from weboob.exceptions import BrowserIncorrectPassword, ActionNeeded, NoAccountsException
from weboob.capabilities.base import find_object
from weboob.capabilities.base import find_object, NotAvailable
from weboob.capabilities.bank import (
AccountNotFound, RecipientNotFound, AddRecipientStep, AddRecipientBankError,
Recipient, TransferBankError, AccountOwnerType,
......@@ -36,7 +36,7 @@
from .pages import (
LoginPage, CardsPage, CardHistoryPage, IncorrectLoginPage,
ProfileProPage, ProfileEntPage, ChangePassPage, SubscriptionPage, InscriptionPage,
ErrorPage, UselessPage,
ErrorPage, UselessPage, MainPage, MarketAccountPage, MarketInvestmentPage,
)
from .json_pages import (
AccountsJsonPage, BalancesJsonPage, HistoryJsonPage, BankStatementPage,
......@@ -107,6 +107,10 @@ def card_history(self, account, coming):
@need_login
def get_cb_operations(self, account):
if account.type in (account.TYPE_MARKET, ):
# market account transactions are in checking account
return
self.location('/Pgn/NavigationServlet?PageID=Cartes&MenuID=%sOPF&Classeur=1&NumeroPage=1&Rib=%s&Devise=%s' % (self.MENUID, account.id, account.currency))
if self.inscription_page.is_here():
......@@ -132,6 +136,8 @@ class SGEnterpriseBrowser(SGPEBrowser):
MENUID = 'BANREL'
CERTHASH = '2231d5ddb97d2950d5e6fc4d986c23be4cd231c31ad530942343a8fdcc44bb99'
main_page = URL('/icd-web/syd-front/index-comptes.html', MainPage)
accounts = URL('/icd/syd-front/data/syd-comptes-accederDepuisMenu.json', AccountsJsonPage)
intraday_accounts = URL('/icd/syd-front/data/syd-intraday-accederDepuisMenu.json', AccountsJsonPage)
......@@ -142,6 +148,13 @@ class SGEnterpriseBrowser(SGPEBrowser):
'/icd/syd-front/data/syd-intraday-chargerDetail.json', HistoryJsonPage)
history_next = URL('/icd/syd-front/data/syd-comptes-chargerProchainLotEcriture.json', HistoryJsonPage)
market_investment = URL(r'/Pgn/NavigationServlet\?.*PageID=CompteTitreDetailFrame',
r'/Pgn/NavigationServlet\?.*PageID=CompteTitreDetail',
MarketInvestmentPage)
market_accounts = URL(r'/Pgn/NavigationServlet\?.*PageID=CompteTitreFrame',
r'/Pgn/NavigationServlet\?.*PageID=CompteTitre',
MarketAccountPage)
profile = URL('/gae/afficherModificationMesDonnees.html', ProfileEntPage)
subscription = URL(r'/Pgn/NavigationServlet\?MenuID=BANRELRIE&PageID=ReleveRIE&NumeroPage=1&Origine=Menu', SubscriptionPage)
......@@ -177,14 +190,37 @@ def get_accounts_list(self):
acc.owner_type = AccountOwnerType.ORGANIZATION
yield acc
# retrieve market accounts if exist
for market_account in self.iter_market_accounts():
yield market_account
@need_login
def iter_history(self, account):
if account.type in (account.TYPE_MARKET, ):
# market account transactions are in checking account
return
value = self.history.go(data={'cl500_compte': account._id, 'cl200_typeReleve': 'valeur'}).get_value()
for tr in self.history.go(data={'cl500_compte': account._id, 'cl200_typeReleve': value}).iter_history(value=value):
yield tr
for tr in self.location('/icd/syd-front/data/syd-intraday-chargerDetail.json', data={'cl500_compte': account._id}).page.iter_history():
yield tr
@need_login
def iter_market_accounts(self):
self.main_page.go()
# retrieve market accounts if exist
market_accounts_link = self.page.get_market_accounts_link()
# there are no examples of entreprise space with market accounts yet
assert not market_accounts_link, 'There are market accounts, retrieve them.'
return []
@need_login
def iter_investment(self, account):
# there are no examples of entreprise space with market accounts yet
return []
@need_login
def iter_subscription(self):
subscriber = self.get_profile()
......@@ -257,6 +293,39 @@ def load_state(self, state):
self.need_reload_state = None
super(SGProfessionalBrowser, self).load_state(state)
@need_login
def iter_market_accounts(self):
self.main_page.go()
# retrieve market accounts if exist
market_accounts_link = self.page.get_market_accounts_link()
if market_accounts_link is NotAvailable:
return []
assert market_accounts_link, 'Market accounts link xpath may have changed'
# need to be on market accounts page to get the accounts iframe
self.location(market_accounts_link)
market_accounts_list_link = self.page.get_table_iframe_link()
if market_accounts_list_link is NotAvailable:
return []
assert market_accounts_link, 'Market accounts iframe link xpath may have changed'
self.location(market_accounts_list_link)
return self.page.iter_market_accounts()
@need_login
def iter_investment(self, account):
if account.type not in (account.TYPE_MARKET, ):
return []
assert account._url_data, 'This account has no url to retrieve investments'
# need to be on market accounts investment page to get the invetment iframe
self.location('/Pgn/NavigationServlet?%s' % account._url_data)
invests_list_link = self.page.get_table_iframe_link()
assert invests_list_link, 'It seems that this market account has no investment'
self.location(invests_list_link)
return self.page.iter_investment()
def copy_recipient_obj(self, recipient):
rcpt = Recipient()
rcpt.id = recipient.iban
......
......@@ -24,15 +24,17 @@
from io import BytesIO
from weboob.browser.pages import HTMLPage, LoggedPage
from weboob.browser.elements import ListElement, ItemElement, method
from weboob.browser.elements import ListElement, ItemElement, method, TableElement
from weboob.browser.filters.standard import (
CleanText, CleanDecimal, Date,
Env, Regexp, Field, Format,
Env, Regexp, Field, Format, TableCell,
)
from weboob.browser.filters.html import Attr
from weboob.browser.filters.html import Attr, Link
from weboob.tools.capabilities.bank.investments import is_isin_valid
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.capabilities.profile import Profile, Person
from weboob.capabilities.bill import Document, Subscription
from weboob.capabilities.bank import Account, Investment
from weboob.exceptions import ActionNeeded, BrowserIncorrectPassword, BrowserUnavailable
from weboob.tools.json import json
......@@ -269,3 +271,83 @@ def get_error(self):
class UselessPage(LoggedPage, SGPEPage):
pass
class MainPage(LoggedPage, SGPEPage):
def get_market_accounts_link(self):
market_accounts_link = Link('//li/a[@title="Comptes titres"]', default=None)(self.doc)
if market_accounts_link:
return market_accounts_link
elif self.doc.xpath('//span[contains(text(), "Comptes titres") and contains(@title, "pas habilité à utiliser ce service")]'):
return NotAvailable
# return None when we don't know if there are market accounts or not
# it will be handled in `browser.py`
class MarketAccountPage(LoggedPage, SGPEPage):
def get_table_iframe_link(self):
if self.doc.xpath('//div[contains(text(), "Aucun compte-titres")]'):
return NotAvailable
return Attr('//iframe[@id="frameTableau"]', 'src')(self.doc)
@method
class iter_market_accounts(TableElement):
item_xpath = '//table[@id="tab-corps"]//tr'
head_xpath = '//table[@id="tab-entete"]//td'
col_id = 'COMPTE'
col_label = 'INTITULE'
col_balance = 'EVALUATION'
class item(ItemElement):
def condition(self):
# table with empty row filled by empty `td`
return Field('number')(self)
klass = Account
obj_id = Format('%s_TITRE', CleanText(TableCell('id'), replace=[(' ', '')]))
obj_number = CleanText(TableCell('id'), replace=[(' ', '')])
obj_label = CleanText(TableCell('label'))
obj_balance = CleanDecimal.French(CleanText(TableCell('balance')))
obj_type = Account.TYPE_MARKET
# all `a` balises have same `href`
obj__url_data = Regexp(Link('(.//a)[1]'), r"lienParent\('(.*)'\)", default=NotAvailable)
class MarketInvestmentPage(LoggedPage, SGPEPage):
def get_table_iframe_link(self):
return Attr('//iframe[@id="frameTableau"]', 'src')(self.doc)
@method
class iter_investment(TableElement):
item_xpath = '//table[@id="tab-corps"]//tr'
head_xpath = '//table[@id="tab-entete"]//td'
col_code = 'CODE'
col_label = 'VALEUR'
col_valuation = 'MONTANT'
col_quantity = 'QUANTITE'
col_unitvalue = 'COURS'
class item(ItemElement):
def condition(self):
# table with empty row filled by empty `td`
return Field('valuation')(self)
klass = Investment
obj_code_type = Investment.CODE_TYPE_ISIN
obj_label = CleanText(TableCell('label'))
obj_valuation = CleanDecimal.French(CleanText(TableCell('valuation')))
obj_quantity = CleanDecimal.French(CleanText(TableCell('quantity')))
obj_unitvalue = CleanDecimal.French(CleanText(TableCell('unitvalue')))
def obj_code(self):
code = CleanText(TableCell('code'))(self)
# there is no example of invests without valid ISIN code
# wait for it to retrieve them corretly
assert is_isin_valid(code), 'This code is not a valid ISIN, please check what invest is it.'
return code