Skip to content
Commits on Source (82)
Hide whitespace changes
Inline Side-by-side
modules/amazon/browser.py
View file @ 09df761c
......@@ -21,7 +21,9 @@
from datetime import date
from weboob.browser import LoginBrowser, URL, need_login, StatesMixin
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable, ImageCaptchaQuestion, BrowserQuestion
from weboob.exceptions import (
BrowserIncorrectPassword, BrowserUnavailable, ImageCaptchaQuestion, BrowserQuestion, ActionNeeded
)
from weboob.tools.value import Value
from weboob.browser.browsers import ClientError
......@@ -85,6 +87,11 @@ def push_security_otp(self, pin_code):
self.location('/ap/signin', data=res_form, headers=self.otp_headers)
def handle_security(self):
otp_type = self.page.get_otp_type()
if otp_type == '/ap/signin':
# this otp will be always present until user deactivate it
raise ActionNeeded('You have enabled otp in your options, please deactivate it before synchronize')
if self.page.doc.xpath('//span[@class="a-button-text"]'):
self.page.send_code()
......
modules/amazon/pages.py
View file @ 09df761c
......@@ -45,9 +45,20 @@ def get_sub_link(self):
class SecurityPage(HTMLPage):
def get_otp_type(self):
# amazon send us otp in two cases:
# - if it's the first time we connect to this account for an ip => manage it normally
# - if user has activated otp in his options => raise ActionNeeded, an ask user to deactivate it
form = self.get_form(xpath='//form[.//h1]')
url = form.url.replace(self.browser.BASEURL, '')
# verify: this otp is sent by amazon when we connect to the account for the first time from a new ip or computer
# /ap/signin: this otp is a user activated otp which is always present
assert url in ('verify', '/ap/signin'), url
return url
def get_otp_message(self):
message = self.doc.xpath('//div[@class="a-box-inner"]/p')
return message[0] if message else None
return CleanText('//div[@class="a-box-inner"]/p')(self.doc)
def send_code(self):
form = self.get_form()
......
modules/bolden/pages.py
View file @ 09df761c
......@@ -61,8 +61,8 @@ class item(ItemElement):
obj_label = 'Compte Bolden'
obj_type = Account.TYPE_MARKET
obj_currency = 'EUR'
obj_balance = CleanDecimal('//div[p[has-class("investor-state") and contains(text(),"Total compte Bolden :")]]/p[has-class("investor-status")]', replace_dots=True)
obj_valuation_diff = CleanDecimal('//p[has-class("rent-amount strong dashboard-text")]', replace_dots=True)
obj_balance = CleanDecimal.French('//div[p[has-class("investor-state") and contains(text(),"Total compte Bolden :")]]/p[has-class("investor-status")]')
obj_valuation_diff = CleanDecimal.French('//div[has-class("rent-total")]')
@method
class iter_investments(TableElement):
......@@ -93,7 +93,7 @@ def obj__docurl(self):
return urljoin(self.page.url, Link('.//a')(TableCell('doc')(self)[0]))
def get_liquidity(self):
return CleanDecimal('//div[p[contains(text(), "Fonds disponibles")]]/p[@class="investor-status strong"]', replace_dots=True)(self.doc)
return CleanDecimal.French('//div[p[contains(text(), "Fonds disponibles")]]/p[has-class("investor-status")]')(self.doc)
class OperationsPage(LoggedPage, HTMLPage):
......
modules/bp/browser.py
View file @ 09df761c
......@@ -74,7 +74,10 @@ class BPBrowser(LoginBrowser, StatesMixin):
'/voscomptes/canalXHTML/pret/encours/detaillerOffrePretConsoListe-encoursPrets.ea',
'/voscomptes/canalXHTML/pret/creditRenouvelable/init-consulterCreditRenouvelable.ea',
'/voscomptes/canalXHTML/pret/encours/rechercherPret-encoursPrets.ea',
'/voscomptes/canalXHTML/sso/commun/init-integration.ea\?partenaire',
'/voscomptes/canalXHTML/sso/lbpf/souscriptionCristalFormAutoPost.jsp',
AccountList)
par_accounts_revolving = URL('https://espaceclientcreditconso.labanquepostale.fr/sav/accueil.do', AccountList)
accounts_rib = URL(r'.*voscomptes/canalXHTML/comptesCommun/imprimerRIB/init-imprimer_rib.ea.*',
'/voscomptes/canalXHTML/comptesCommun/imprimerRIB/init-selection_rib.ea', AccountRIB)
......@@ -103,10 +106,11 @@ class BPBrowser(LoginBrowser, StatesMixin):
par_account_checking_history = URL('/voscomptes/canalXHTML/CCP/releves_ccp/init-releve_ccp.ea\?typeRecherche=10&compte.numero=(?P<accountId>.*)',
'/voscomptes/canalXHTML/CCP/releves_ccp/afficher-releve_ccp.ea', AccountHistory)
deferred_card_history = URL(r'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>\w+)&indexCarte=(?P<cardIndex>\d+)&typeListe=(?P<type>\d+)', AccountHistory)
deferred_card_history_multi = URL(r'/voscomptes/canalXHTML/CB/releveCB/preparerRecherche-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>\w+)&indexCarte=(?P<cardIndex>\d+)&typeListe=(?P<type>\d+)', AccountHistory) # &typeRecherche=10
deferred_card_history = URL(r'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>\w+)&indexCompte=(?P<cardIndex>\d+)&typeListe=(?P<type>\d+)', AccountHistory)
deferred_card_history_multi = URL(r'/voscomptes/canalXHTML/CB/releveCB/preparerRecherche-mouvementsCarteDD.ea\?indexCompte=(?P<accountId>\w+)&indexCarte=(?P<cardIndex>\d+)&typeListe=(?P<type>\d+)', AccountHistory) # &typeRecherche=10
par_account_checking_coming = URL('/voscomptes/canalXHTML/CCP/releves_ccp_encours/preparerRecherche-releve_ccp_encours.ea\?compte.numero=(?P<accountId>.*)&typeRecherche=1',
'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>.*)&typeListe=1&typeRecherche=10', AccountHistory)
'/voscomptes/canalXHTML/CB/releveCB/init-mouvementsCarteDD.ea\?compte.numero=(?P<accountId>.*)&typeListe=1&typeRecherche=10',
'/voscomptes/canalXHTML/CCP/releves_ccp_encours/preparerRecherche-releve_ccp_encours.ea\?indexCompte', AccountHistory)
par_account_savings_and_invests_history = URL('/voscomptes/canalXHTML/CNE/releveCNE/init-releve_cne.ea\?typeRecherche=10&compte.numero=(?P<accountId>.*)',
'/voscomptes/canalXHTML/CNE/releveCNE/releveCNE-releve_cne.ea', AccountHistory)
......@@ -218,6 +222,10 @@ def do_login(self):
@need_login
def get_accounts_list(self):
if self.session.cookies.get('indicateur'):
# Malformed cookie to delete to reach other spaces
del self.session.cookies['indicateur']
if self.accounts is None:
accounts = []
to_check = []
......@@ -237,7 +245,7 @@ def get_accounts_list(self):
for account in self.page.iter_accounts():
if account.type == Account.TYPE_LOAN:
self.location(account.url)
if 'CreditRenouvelable' not in account.url:
if 'initSSO' not in account.url:
for loan in self.page.iter_loans():
loan.currency = account.currency
accounts.append(loan)
......@@ -248,9 +256,11 @@ def get_accounts_list(self):
student_loan.currency = account.currency
accounts.append(student_loan)
else:
for loan in self.page.iter_revolving_loans():
loan.currency = account.currency
accounts.append(loan)
# The main revolving page is not accessible, we can reach it by this new way
self.location(self.absurl('/voscomptes/canalXHTML/sso/lbpf/souscriptionCristalFormAutoPost.jsp'))
self.page.go_revolving()
revolving_loan = self.page.get_revolving_attributes(account)
accounts.append(revolving_loan)
page.go()
elif account.type == Account.TYPE_PERP:
......@@ -308,7 +318,7 @@ def get_history(self, account):
self.go_linebourse(account)
return self.linebourse.iter_history(account.id)
if account.type == Account.TYPE_LOAN:
if account.type in (Account.TYPE_LOAN, Account.TYPE_REVOLVING_CREDIT):
return []
if account.type == Account.TYPE_CARD:
......
modules/bp/pages/accountlist.py
View file @ 09df761c
......@@ -30,7 +30,7 @@
from weboob.browser.elements import ListElement, ItemElement, method, TableElement
from weboob.browser.pages import LoggedPage, RawPage, PartialHTMLPage, HTMLPage
from weboob.browser.filters.html import Link, TableCell
from weboob.browser.filters.standard import CleanText, CleanDecimal, Regexp, Env, Field, BrowserURL, Currency, Async, Date, Format
from weboob.browser.filters.standard import CleanText, CleanDecimal, Regexp, Env, Field, Currency, Async, Date, Format
from weboob.exceptions import BrowserUnavailable
from weboob.tools.compat import urljoin, unicode
......@@ -58,6 +58,8 @@ def condition(self):
def obj_url(self):
url = Link(u'./a', default=NotAvailable)(self)
if url:
if 'CreditRenouvelable' in url:
url = Link(u'.//a[contains(text(), "espace de gestion crédit renouvelable")]')(self.el)
return urljoin(self.page.url, url)
return url
......@@ -74,9 +76,9 @@ def obj_coming(self):
has_coming = False
coming = 0
self.page.browser.open(Field('url')(self))
coming_operations = self.page.browser.open(
BrowserURL('par_account_checking_coming', accountId=Field('id'))(self))
details_page = self.page.browser.open(Field('url')(self))
coming_op_link = Regexp(Link(u'//a[contains(text(), "Opérations à venir")]'), r'../(.*)')(details_page.page.doc)
coming_operations = self.page.browser.open(self.page.browser.BASEURL + '/voscomptes/canalXHTML/CCP/' + coming_op_link)
if CleanText('//span[@id="amount_total"]')(coming_operations.page.doc):
has_coming = True
......@@ -91,10 +93,11 @@ def obj_coming(self):
return NotAvailable
def obj_iban(self):
response = self.page.browser.open(
'/voscomptes/canalXHTML/comptesCommun/imprimerRIB/init-imprimer_rib.ea?numeroCompte=%s' % Field('id')(
self))
return response.page.get_iban()
rib_link = Link('//a[abbr[contains(text(), "RIB")]]', default=NotAvailable)(self.el)
if rib_link:
response = self.page.browser.open(rib_link)
return response.page.get_iban()
return NotAvailable
def obj_type(self):
types = {'comptes? bancaires?': Account.TYPE_CHECKING,
......@@ -137,6 +140,10 @@ def on_load(self):
raise BrowserUnavailable()
def go_revolving(self):
form = self.get_form()
form.submit()
@property
def no_accounts(self):
return len(self.doc.xpath('//iframe[contains(@src, "/comptes_contrats/sans_")] |\
......@@ -156,6 +163,22 @@ class item_account(item_account_generic):
def condition(self):
return item_account_generic.condition(self)
def get_revolving_attributes(self, account):
loan = Loan()
loan.id = account.id
loan.label = '%s - %s' %(account.label, account.id)
loan.currency = account.currency
loan.url = account.url
loan.available_amount = CleanDecimal('//tr[td[contains(text(), "Montant Maximum Autorisé") or contains(text(), "Montant autorisé")]]/td[2]')(self.doc)
loan.used_amount = loan.used_amount = CleanDecimal('//tr[td[contains(text(), "Montant Utilisé") or contains(text(), "Montant utilisé")]]/td[2]')(self.doc)
loan.available_amount = CleanDecimal(Regexp(CleanText('//tr[td[contains(text(), "Montant Disponible") or contains(text(), "Montant disponible")]]/td[2]'), r'(.*) au'))(self.doc)
loan._has_cards = False
loan.type = Account.TYPE_REVOLVING_CREDIT
return loan
@method
class iter_revolving_loans(ListElement):
item_xpath = '//div[@class="bloc Tmargin"]//dl'
......
modules/caissedepargne/browser.py
View file @ 09df761c
......@@ -333,7 +333,7 @@ def loans_conso(self):
days = ('Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun')
month = ('Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec')
now = datetime.datetime.today()
d = '%s %s %s %s:%s:%s GMT 0100 (CET)' % (days[now.weekday()], month[now.month - 1], now.year, now.hour, format(now.minute, "02"), now.second)
d = '%s %s %s %s %s:%s:%s GMT+0100 (heure normale d’Europe centrale)' % (days[now.weekday()], now.day, month[now.month - 1], now.year, now.hour, format(now.minute, "02"), now.second)
if self.home.is_here():
msg = self.page.loan_unavailable_msg()
if msg:
......@@ -597,7 +597,7 @@ def get_coming(self, account):
@need_login
def get_investment(self, account):
self.deleteCTX()
if account.type not in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_MARKET, Account.TYPE_PEA) or 'measure_id' in account._info:
if account.type not in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_CAPITALISATION, Account.TYPE_MARKET, Account.TYPE_PEA) or 'measure_id' in account._info:
raise NotImplementedError()
if account.type == Account.TYPE_PEA and account.label == 'PEA NUMERAIRE':
......@@ -622,7 +622,7 @@ def get_investment(self, account):
yield investment
return
elif account.type == Account.TYPE_LIFE_INSURANCE:
elif account.type in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_CAPITALISATION):
if "MILLEVIE" in account.label:
self.page.go_life_insurance(account)
label = account.label.split()[-1]
......
modules/caissedepargne/pages.py
View file @ 09df761c
......@@ -296,6 +296,8 @@ def _add_account(self, accounts, link, label, account_type, balance):
account.type = self.ACCOUNT_TYPES.get(label, info['acc_type'] if 'acc_type' in info else account_type)
if 'PERP' in account.label:
account.type = Account.TYPE_PERP
if 'NUANCES CAPITALISATI' in account.label:
account.type = Account.TYPE_CAPITALISATION
balance = balance or self.get_balance(account)
account.balance = Decimal(FrenchTransaction.clean_amount(balance)) if balance and balance is not NotAvailable else NotAvailable
......@@ -315,10 +317,10 @@ def _add_account(self, accounts, link, label, account_type, balance):
accounts[account.id] = account
def get_balance(self, account):
if account.type not in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_PERP):
if account.type not in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_PERP, Account.TYPE_CAPITALISATION):
return NotAvailable
page = self.go_history(account._info).page
balance = page.doc.xpath('.//tr[td[ends-with(@id,"NumContrat")]/a[contains(text(),$id)]]/td[@class="somme"]', id=account.id)
balance = page.doc.xpath('.//tr[td[contains(@id,"NumContrat")]]/td[@class="somme"]/a[contains(@href, $id)]', id=account.id)
if len(balance) > 0:
balance = CleanText('.')(balance[0])
balance = balance if balance != u'' else NotAvailable
......@@ -408,7 +410,7 @@ def go_loans_conso(self, tr):
if m:
account = m.group(1)
form = self.get_form(name="main")
form = self.get_form(id="main")
form['__EVENTTARGET'] = 'MM$SYNTHESE_CREDITS'
form['__EVENTARGUMENT'] = 'ACTIVDESACT_CREDITCONSO&%s' % account
form['m_ScriptManager'] = 'MM$m_UpdatePanel|MM$SYNTHESE_CREDITS'
......@@ -459,6 +461,9 @@ def get_loan_list(self):
account._card_links = []
if "renouvelables" in CleanText('.')(title):
if 'JSESSIONID' in self.browser.session.cookies:
# Need to delete this to access the consumer loans space (a new one will be created)
del self.browser.session.cookies['JSESSIONID']
self.go_loans_conso(tr)
d = self.browser.loans_conso()
if d:
......@@ -502,11 +507,10 @@ class item(ItemElement):
obj_currency = Currency(MyTableCell("balance"))
obj_last_payment_date = Date(CleanText(MyTableCell("last_payment_date")))
obj_next_payment_amount = MyDecimal(MyTableCell("next_payment_amount"))
obj_next_payment_date = Date(CleanText(MyTableCell("next_payment_date")))
obj_next_payment_date = Date(CleanText(MyTableCell("next_payment_date", default=''), default=NotAvailable), default=NotAvailable)
def go_list(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTARGUMENT'] = "CPTSYNT0"
......@@ -525,7 +529,7 @@ def go_list(self):
# On some pages, navigate to indexPage does not lead to the list of measures, so we need this form ...
def go_measure_list(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTARGUMENT'] = "MESLIST0"
form['__EVENTTARGET'] = 'Menu_AJAX'
......@@ -537,7 +541,7 @@ def go_measure_list(self):
# This function goes to the accounts page of one measure giving its id
def go_measure_accounts_list(self, measure_id):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTARGUMENT'] = "CPTSYNT0"
......@@ -556,7 +560,7 @@ def go_measure_accounts_list(self, measure_id):
form.submit()
def go_loan_list(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTARGUMENT'] = "CRESYNT0"
......@@ -573,7 +577,7 @@ def go_loan_list(self):
form.submit()
def go_history(self, info, is_cbtab=False):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = 'MM$%s' % (info['type'] if is_cbtab else 'SYNTHESE')
form['__EVENTARGUMENT'] = info['link']
......@@ -587,7 +591,7 @@ def go_history(self, info, is_cbtab=False):
def get_form_to_detail(self, transaction):
m = re.match('.*\("(.*)", "(DETAIL_OP&[\d]+).*\)\)', transaction._link)
# go to detailcard page
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = m.group(1)
form['__EVENTARGUMENT'] = m.group(2)
fix_form(form)
......@@ -650,7 +654,7 @@ def go_next(self):
# <a id="MM_HISTORIQUE_CB_lnkSuivante" class="next" href="javascript:WebForm_DoPostBackWithOptions(new WebForm_PostBackOptions(&quot;MM$HISTORIQUE_CB$lnkSuivante&quot;, &quot;&quot;, true, &quot;&quot;, &quot;&quot;, false, true))">Suivant<span class="arrow">></span></a>
link = self.doc.xpath('//a[contains(@id, "lnkSuivante")]')
if len(link) == 0 or 'disabled' in link[0].attrib:
if len(link) == 0 or 'disabled' in link[0].attrib or link[0].attrib.get('class') == 'aspNetDisabled':
return False
account_type = 'COMPTE'
......@@ -658,7 +662,7 @@ def go_next(self):
if m:
account_type = m.group(1)
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = "MM$HISTORIQUE_%s$lnkSuivante" % account_type
form['__EVENTARGUMENT'] = ''
......@@ -679,7 +683,7 @@ def go_life_insurance(self, account):
link = self.doc.xpath('//tr[td[contains(., ' + account.id + ') ]]//a')[0]
m = re.search("PostBackOptions?\([\"']([^\"']+)[\"'],\s*['\"]((REDIR_ASS_VIE)?[\d\w&]+)?['\"]", link.attrib.get('href', ''))
if m is not None:
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = m.group(1)
form['__EVENTARGUMENT'] = m.group(2)
......@@ -712,7 +716,7 @@ def go_transfer(self, account):
else:
link = link[0]
m = re.search("PostBackOptions?\([\"']([^\"']+)[\"'],\s*['\"]([^\"']+)?['\"]", link.attrib.get('href', ''))
form = self.get_form(name='main')
form = self.get_form(id='main')
if 'MM$HISTORIQUE_COMPTE$btnCumul' in form:
del form['MM$HISTORIQUE_COMPTE$btnCumul']
form['__EVENTTARGET'] = m.group(1)
......@@ -728,7 +732,7 @@ def loan_unavailable_msg(self):
return msg
def go_subscription(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['m_ScriptManager'] = 'MM$m_UpdatePanel|MM$Menu_Ajax'
form['__EVENTTARGET'] = 'MM$Menu_Ajax'
link = Link('//a[contains(@title, "e-Documents") or contains(@title, "Relevés en ligne")]')(self.doc)
......@@ -1005,7 +1009,7 @@ def get_recipient_value(self, recipient):
return recipient_value[0]
def init_transfer(self, account, recipient, transfer):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['MM$VIREMENT$SAISIE_VIREMENT$ddlCompteDebiter'] = self.get_origin_account_value(account)
form['MM$VIREMENT$SAISIE_VIREMENT$ddlCompteCrediter'] = self.get_recipient_value(recipient)
form['MM$VIREMENT$SAISIE_VIREMENT$txtLibelleVirement'] = transfer.label
......@@ -1021,7 +1025,7 @@ class iter_recipients(MyRecipients):
pass
def continue_transfer(self, origin_label, recipient, label):
form = self.get_form(name='main')
form = self.get_form(id='main')
type_ = 'intra' if recipient.category == u'Interne' else 'sepa'
fill = lambda s, t: s % (t.upper(), t.capitalize())
form['__EVENTTARGET'] = 'MM$VIREMENT$m_WizardBar$m_lnkNext$m_lnkButton'
......@@ -1032,7 +1036,7 @@ def continue_transfer(self, origin_label, recipient, label):
form.submit()
def go_add_recipient(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
link = self.doc.xpath(u'//a[span[contains(text(), "Ajouter un compte bénéficiaire")]]')[0]
m = re.search("PostBackOptions?\([\"']([^\"']+)[\"'],\s*['\"]([^\"']+)?['\"]", link.attrib.get('href', ''))
form['__EVENTTARGET'] = m.group(1)
......@@ -1051,7 +1055,7 @@ def is_here(self):
return bool(CleanText(u'//h2[contains(text(), "Confirmer mon virement")]')(self.doc))
def confirm(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = 'MM$VIREMENT$m_WizardBar$m_lnkNext$m_lnkButton'
form.submit()
......@@ -1152,7 +1156,7 @@ class iter_recipients(MyRecipients):
pass
def init_transfer(self, account, recipient, transfer):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['MM$VIREMENT$SAISIE_VIREMENT$ddlCompteDebiter'] = self.get_origin_account_value(account)
form['MM$VIREMENT$SAISIE_VIREMENT$ddlCompteCrediterPro'] = self.get_recipient_value(recipient)
form['MM$VIREMENT$SAISIE_VIREMENT$Libelle'] = transfer.label
......@@ -1166,7 +1170,7 @@ def init_transfer(self, account, recipient, transfer):
form.submit()
def go_add_recipient(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = 'MM$VIREMENT$SAISIE_VIREMENT$ddlCompteCrediterPro'
form['MM$VIREMENT$SAISIE_VIREMENT$ddlCompteCrediterPro'] = 'AC'
form.submit()
......@@ -1228,7 +1232,7 @@ def is_here(self):
return bool(CleanText(u'//h2[contains(text(), "Authentification réussie")]')(self.doc))
def go_on(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = 'MM$RETOUR_OK_SOL$m_ChoiceBar$lnkRight'
form.submit()
......@@ -1247,7 +1251,7 @@ def is_here(self):
//h2[contains(text(), "Confirmer l\'ajout d\'un compte bénéficiaire")]')(self.doc))
def post_recipient(self, recipient):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = '%s$m_WizardBar$m_lnkNext$m_lnkButton' % self.EVENTTARGET
form['%s$m_RibIban$txtTitulaireCompte' % self.FORM_FIELD_ADD] = recipient.label
for i in range(len(recipient.iban) // 4 + 1):
......@@ -1255,7 +1259,7 @@ def post_recipient(self, recipient):
form.submit()
def confirm_recipient(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = 'MM$WIZARD_AJOUT_COMPTE_EXTERNE$m_WizardBar$m_lnkNext$m_lnkButton'
form.submit()
......@@ -1270,7 +1274,7 @@ def is_here(self):
return self.need_auth() and self.doc.xpath('//span[@id="MM_ANR_WS_AUTHENT_ANR_WS_AUTHENT_SAISIE_lblProcedure1"]')
def set_browser_form(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = 'MM$ANR_WS_AUTHENT$m_WizardBar$m_lnkNext$m_lnkButton'
self.browser.recipient_form = dict((k, v) for k, v in form.items())
self.browser.recipient_form['url'] = form.url
......@@ -1306,8 +1310,9 @@ class get_detail(TableElement):
def next_page(self):
# only for new website, don't have any accounts with enough deferred card transactions on old webiste
if self.page.doc.xpath('//a[contains(@id, "lnkSuivante") and not(contains(@disabled,"disabled"))]'):
form = self.page.get_form(name='main')
if self.page.doc.xpath('//a[contains(@id, "lnkSuivante") and not(contains(@disabled,"disabled")) \
and not(contains(@class, "aspNetDisabled"))]'):
form = self.page.get_form(id='main')
form['__EVENTTARGET'] = "MM$ECRITURE_GLOBALE$lnkSuivante"
form['__EVENTARGUMENT'] = ''
fix_form(form)
......@@ -1329,12 +1334,12 @@ def go_form_to_summary(self):
# return to first page
to_history = Link(self.doc.xpath(u'//a[contains(text(), "Retour à l\'historique")]'))(self.doc)
n = re.match('.*\([\'\"](MM\$.*?)[\'\"],.*\)$', to_history)
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = n.group(1)
form.submit()
def go_newsite_back_to_summary(self):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['__EVENTTARGET'] = "MM$ECRITURE_GLOBALE$lnkRetourHisto"
form.submit()
......@@ -1363,7 +1368,7 @@ def condition(self):
def go_document_list(self, sub_id):
target = Attr('//select[contains(@id, "ClientsBancaires")]', 'id')(self.doc)
form = self.get_form(name='main')
form = self.get_form(id='main')
form['m_ScriptManager'] = target
if 'palatine' in self.browser.BASEURL:
form['MM$CONSULTATION_NUMERISATION_PALATINE$cboClientsBancaires'] = sub_id
......@@ -1383,16 +1388,25 @@ class iter_documents(ListElement):
class item(ItemElement):
klass = Document
obj_label = Format('%s %s', CleanText('./preceding::h3[1]'), CleanText('./span'))
obj_date = Date(CleanText('./span'), dayfirst=True)
obj_type = DocumentTypes.OTHER
obj_format = 'pdf'
obj_url = Regexp(Link('.'), r'WebForm_PostBackOptions\("(\S*)"')
obj_id = Format('%s_%s_%s', Env('sub_id'), CleanText('./span', symbols='/'), Regexp(Field('url'), r'ctl(.*)'))
obj_id = Format('%s_%s_%s', Env('sub_id'), CleanText('./span', symbols='/', replace=[(' ', '_')]), Regexp(Field('url'), r'ctl(.*)'))
obj__event_id = Regexp(Attr('.', 'onclick'), r"val\('(.*)'\);", default=None)
def obj_label(self):
if 'Récapitulatif de frais bancaires' in CleanText('./span')(self.el):
return CleanText('./span')(self.el)
return Format('%s %s', CleanText('./preceding::h3[1]'), CleanText('./span'))(self.el)
def obj_date(self):
if 'Récapitulatif de frais bancaires' in CleanText('./span')(self.el):
year = Regexp(CleanText('./span'), r'(\d{4})')(self.el)
return Date(dayfirst=True).filter('31/12/%s' %year)
return Date(CleanText('./span'), dayfirst=True)(self.el)
def download_document(self, document):
form = self.get_form(name='main')
form = self.get_form(id='main')
form['m_ScriptManager'] = document.url
form['__EVENTTARGET'] = document.url
form['MM$COMPTE_EDOCUMENTS$ctrlEDocumentsConsultationDocument$eventId'] = document._event_id
......
modules/cmso/module.py
View file @ 09df761c
......@@ -19,9 +19,9 @@
from __future__ import unicode_literals
from weboob.capabilities.bank import CapBankWealth, AccountNotFound
from weboob.capabilities.bank import CapBankTransfer, CapBankWealth, Account, AccountNotFound, RecipientNotFound
from weboob.capabilities.contact import CapContact
from weboob.capabilities.base import find_object
from weboob.capabilities.base import find_object, strict_find_object
from weboob.capabilities.profile import CapProfile
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import Value, ValueBackendPassword
......@@ -33,7 +33,7 @@
__all__ = ['CmsoModule']
class CmsoModule(Module, CapBankWealth, CapContact, CapProfile):
class CmsoModule(Module, CapBankTransfer, CapBankWealth, CapContact, CapProfile):
NAME = 'cmso'
MAINTAINER = 'Romain Bignon'
EMAIL = 'romain@weboob.org'
......@@ -69,6 +69,39 @@ def iter_coming(self, account):
def iter_investment(self, account):
return self.browser.iter_investment(account)
def iter_transfer_recipients(self, origin_account):
if self.config['website'].get() != "par":
raise NotImplementedError()
if not isinstance(origin_account, Account):
origin_account = self.get_account(origin_account)
return self.browser.iter_recipients(origin_account)
def init_transfer(self, transfer, **params):
if self.config['website'].get() != "par":
raise NotImplementedError()
self.logger.info('Going to do a new transfer')
account = strict_find_object(
self.iter_accounts(),
error=AccountNotFound,
iban=transfer.account_iban,
id=transfer.account_id
)
recipient = strict_find_object(
self.iter_transfer_recipients(account.id),
error=RecipientNotFound,
iban=transfer.recipient_iban,
id=transfer.recipient_id
)
return self.browser.init_transfer(account, recipient, transfer.amount, transfer.label, transfer.exec_date)
def execute_transfer(self, transfer, **params):
return self.browser.execute_transfer(transfer, **params)
def iter_contacts(self):
if self.config['website'].get() != "par":
raise NotImplementedError()
......
modules/cmso/par/browser.py
View file @ 09df761c
......@@ -22,6 +22,7 @@
import re
import json
from datetime import date
from functools import wraps
from weboob.browser import LoginBrowser, URL, need_login, StatesMixin
......@@ -33,8 +34,9 @@
from .pages import (
LogoutPage, InfosPage, AccountsPage, HistoryPage, LifeinsurancePage, MarketPage,
AdvisorPage, LoginPage, RecipientsPage, ProfilePage,
AdvisorPage, LoginPage, ProfilePage,
)
from .transfer_pages import TransferInfoPage, RecipientsListPage, TransferPage
def retry(exc_check, tries=4):
......@@ -92,7 +94,11 @@ class CmsoParBrowser(LoginBrowser, StatesMixin):
'https://www.*/domiweb/prive/particulier', MarketPage)
advisor = URL('/edrapi/v(?P<version>\w+)/oauth/(?P<page>\w+)', AdvisorPage)
recipients = URL(r'/domiapi/oauth/json/transfer/transferinfos', RecipientsPage)
# transfer
transfer_info = URL(r'/domiapi/oauth/json/transfer/transferinfos', TransferInfoPage)
recipients_list = URL(r'/domiapi/oauth/json/transfer/beneficiariesListTransfer', RecipientsListPage)
init_transfer_page = URL(r'/domiapi/oauth/json/transfer/controlTransferOperation', TransferPage)
execute_transfer_page = URL(r'/domiapi/oauth/json/transfer/transferregister', TransferPage)
profile = URL(r'/domiapi/oauth/json/edr/infosPerson', ProfilePage)
......@@ -158,7 +164,7 @@ def iter_accounts(self):
seen = {}
self.recipients.go(data='{"beneficiaryType":"INTERNATIONAL"}', headers=self.json_headers)
self.transfer_info.go(json={"beneficiaryType":"INTERNATIONAL"})
numbers = self.page.get_numbers()
# First get all checking accounts...
......@@ -305,6 +311,74 @@ def iter_investment(self, account):
return []
raise NotImplementedError()
@retry((ClientError, ServerError))
@need_login
def iter_recipients(self, account):
self.transfer_info.go(json={"beneficiaryType":"INTERNATIONAL"})
if account.type in (Account.TYPE_LOAN, ):
return
if not account._eligible_debit:
return
# internal recipient
for rcpt in self.page.iter_titu_accounts():
if rcpt.id != account.id:
yield rcpt
for rcpt in self.page.iter_manda_accounts():
if rcpt.id != account.id:
yield rcpt
for rcpt in self.page.iter_legal_rep_accounts():
if rcpt.id != account.id:
yield rcpt
# external recipient
for rcpt in self.page.iter_external_recipients():
yield rcpt
@need_login
def init_transfer(self, account, recipient, amount, reason, exec_date):
self.recipients_list.go(json={"beneficiaryType":"INTERNATIONAL"})
transfer_data = {
'beneficiaryIndex': self.page.get_rcpt_index(recipient),
'debitAccountIndex': account._index,
'devise': account.currency,
'deviseReglement': account.currency,
'montant': amount,
'nature': 'externesepa',
'transferToBeneficiary': True,
}
if exec_date and exec_date > date.today():
transfer_data['date'] = int(exec_date.strftime('%s')) * 1000
else:
transfer_data['immediate'] = True
# check if recipient is internal or external
if recipient.id != recipient.iban:
transfer_data['nature'] = 'interne'
transfer_data['transferToBeneficiary'] = False
self.init_transfer_page.go(json=transfer_data)
transfer = self.page.handle_transfer(account, recipient, amount, reason, exec_date)
# transfer_data is used in execute_transfer
transfer._transfer_data = transfer_data
return transfer
@need_login
def execute_transfer(self, transfer, **params):
assert transfer._transfer_data
transfer._transfer_data.update({
'enregistrerNouveauBeneficiaire': False,
'creditLabel': 'de %s' % transfer.account_label if not transfer.label else transfer.label,
'debitLabel': 'vers %s' % transfer.recipient_label,
'typeFrais': 'SHA'
})
self.execute_transfer_page.go(json=transfer._transfer_data)
transfer.id = self.page.get_transfer_confirm_id()
return transfer
@retry((ClientError, ServerError))
@need_login
def get_advisor(self):
......
modules/cmso/par/pages.py
View file @ 09df761c
......@@ -135,6 +135,8 @@ class item(ItemElement):
# Iban is available without last 5 numbers, or by sms
obj_iban = NotAvailable
obj__index = Dict('index')
# to know if we can do transfer on account
obj__eligible_debit = Dict('eligibiliteDebit', default=False)
def obj_balance(self):
balance = CleanDecimal(Dict('soldeEuro', default="0"))(self)
......@@ -194,6 +196,8 @@ class item(ItemElement):
obj_coming = CleanDecimal(Dict('AVenir', default=None), default=NotAvailable)
obj__index = Dict('index')
obj__owner = Dict('nomTitulaire')
# to know if we can do transfer on account
obj__eligible_debit = Dict('eligibiliteDebit', default=False)
def obj_id(self):
type = Field('type')(self)
......@@ -549,35 +553,6 @@ class update_agency(ItemElement):
obj_address = Format('%s %s', Dict('adresse1'), Dict('adresse3'))
class RecipientsPage(LoggedPage, JsonPage):
def get_numbers(self):
# If account information is not available when asking for the
# recipients (server error for ex.), return an empty dictionary
# that will be filled later after being returned the json of the
# account page (containing the accounts IDs too).
if 'listCompteTitulaireCotitulaire' not in self.doc and 'exception' in self.doc:
return {}
ret = {}
ret.update({
d['index']: d['numeroContratSouscrit']
for d in self.doc['listCompteTitulaireCotitulaire']
})
ret.update({
d['index']: d['numeroContratSouscrit']
for p in self.doc['listCompteMandataire'].values()
for d in p
})
ret.update({
d['index']: d['numeroContratSouscrit']
for p in self.doc['listCompteLegalRep'].values()
for d in p
})
return ret
class ProfilePage(LoggedPage, JsonPage):
# be careful, this page is used in CmsoProBrowser too!
......
modules/cmso/par/transfer_pages.py 0 → 100644
View file @ 09df761c
# -*- coding: utf-8 -*-
# Copyright(C) 2019 Sylvie Ye
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
import datetime as dt
from weboob.browser.pages import JsonPage, LoggedPage
from weboob.browser.elements import DictElement, ItemElement, method
from weboob.browser.filters.standard import CleanText, CleanDecimal, Currency
from weboob.browser.filters.json import Dict
from weboob.capabilities.bank import Recipient, Transfer, TransferBankError
from weboob.capabilities.base import NotAvailable
class MyRecipientItemElement(ItemElement):
def condition(self):
return Dict('eligibiliteCredit', default=False)
klass = Recipient
obj_id = Dict('numeroContratSouscrit')
obj_label = Dict('lib')
obj_iban = NotAvailable
obj_enabled_at = dt.date.today()
obj_category = 'Interne'
obj__index = Dict('index')
class RecipientsListPage(LoggedPage, JsonPage):
def get_rcpt_index(self, recipient):
if recipient.category == 'Externe':
for el in self.doc['listBeneficiaries']:
for rcpt in el:
# in this list, recipient iban is like FR111111111111111XXXXX
if rcpt['iban'][:-5] == recipient.iban[:-5] and rcpt['nom'] == recipient.label:
return rcpt['index']
return recipient._index
class TransferInfoPage(LoggedPage, JsonPage):
def get_numbers(self):
# If account information is not available when asking for the
# recipients (server error for ex.), return an empty dictionary
# that will be filled later after being returned the json of the
# account page (containing the accounts IDs too).
if 'listCompteTitulaireCotitulaire' not in self.doc and 'exception' in self.doc:
return {}
ret = {}
ret.update({
d['index']: d['numeroContratSouscrit']
for d in self.doc['listCompteTitulaireCotitulaire']
})
ret.update({
d['index']: d['numeroContratSouscrit']
for p in self.doc['listCompteMandataire'].values()
for d in p
})
ret.update({
d['index']: d['numeroContratSouscrit']
for p in self.doc['listCompteLegalRep'].values()
for d in p
})
return ret
@method
class iter_titu_accounts(DictElement):
item_xpath = 'listCompteTitulaireCotitulaire'
class item(MyRecipientItemElement):
pass
@method
class iter_manda_accounts(DictElement):
item_xpath = 'listCompteMandataire/*'
class item(MyRecipientItemElement):
pass
@method
class iter_legal_rep_accounts(DictElement):
item_xpath = 'listCompteLegalRep/*'
class item(MyRecipientItemElement):
pass
@method
class iter_external_recipients(DictElement):
item_xpath = 'listBeneficiaries'
class item(ItemElement):
klass = Recipient
obj_id = obj_iban = Dict('iban')
obj_label = Dict('nom')
obj_category = 'Externe'
obj_enabled_at = dt.date.today()
obj__index = Dict('index')
def condition(self):
return Dict('actif', default=False)(self)
class TransferPage(LoggedPage, JsonPage):
def on_load(self):
if self.doc.get('exception') and not self.doc.get('debitAccountOwner'):
if Dict('exception/type')(self.doc) == 1:
# technical error
assert False, 'Error with code %s occured during init_transfer: %s' % \
(Dict('exception/code')(self.doc), Dict('exception/message')(self.doc))
elif Dict('exception/type')(self.doc) == 2:
# user error
TransferBankError(message=Dict('exception/message')(self.doc))
def handle_transfer(self, account, recipient, amount, reason, exec_date):
transfer = Transfer()
transfer.amount = CleanDecimal(Dict('amount'))(self.doc)
transfer.currency = Currency(Dict('codeDevise'))(self.doc)
transfer.label = reason
if exec_date:
transfer.exec_date = dt.date.fromtimestamp(int(Dict('date')(self.doc))//1000)
transfer.account_id = account.id
transfer.account_label = CleanText(Dict('debitAccountLabel'))(self.doc)
transfer.account_balance = CleanDecimal(Dict('debitAccountBalance'))(self.doc)
transfer.recipient_id = recipient.id
transfer.recipient_iban = recipient.iban
transfer.recipient_label = CleanText(Dict('creditAccountOwner'))(self.doc)
return transfer
def get_transfer_confirm_id(self):
return self.doc.get('numeroOperation')
modules/cragr/api/browser.py
View file @ 09df761c
......@@ -20,22 +20,28 @@
from __future__ import unicode_literals
from decimal import Decimal
import re
from weboob.capabilities.bank import (
Account,
)
from weboob.capabilities.base import find_object, empty, NotAvailable
from weboob.capabilities.bank import Account, Transaction
from weboob.capabilities.base import empty, NotAvailable
from weboob.browser import LoginBrowser, URL, need_login
from weboob.exceptions import BrowserUnavailable, BrowserIncorrectPassword
from weboob.browser.exceptions import ServerError
from weboob.exceptions import BrowserUnavailable, BrowserIncorrectPassword, ActionNeeded
from weboob.browser.exceptions import ServerError, BrowserHTTPNotFound
from weboob.capabilities.bank import Loan
from weboob.tools.capabilities.bank.iban import is_iban_valid
from weboob.tools.capabilities.bank.transactions import sorted_transactions
from .pages import (
LoginPage, LoggedOutPage, KeypadPage, SecurityPage, ContractsPage, AccountsPage, AccountDetailsPage,
IbanPage, HistoryPage, ProfilePage,
LoginPage, LoggedOutPage, KeypadPage, SecurityPage, ContractsPage, FirstConnectionPage, AccountsPage, AccountDetailsPage,
TokenPage, IbanPage, HistoryPage, CardsPage, CardHistoryPage, NetfincaRedirectionPage, PredicaRedirectionPage,
PredicaInvestmentsPage, ProfilePage, ProfileDetailsPage, ProProfileDetailsPage,
)
from weboob.tools.capabilities.bank.investments import create_french_liquidity
from .netfinca_browser import NetfincaBrowser
__all__ = ['CragrAPI']
......@@ -44,32 +50,84 @@ class CragrAPI(LoginBrowser):
login_page = URL(r'particulier/acceder-a-mes-comptes.html$', LoginPage)
keypad = URL(r'particulier/acceder-a-mes-comptes.authenticationKeypad.json', KeypadPage)
security_check = URL(r'particulier/acceder-a-mes-comptes.html/j_security_check', SecurityPage)
first_connection = URL(r'.*/operations/interstitielles/premiere-connexion.html', FirstConnectionPage)
logged_out = URL(r'.*', LoggedOutPage)
token_page = URL(r'libs/granite/csrf/token.json', TokenPage)
contracts_page = URL(r'particulier/operations/.rechargement.contexte.html\?idBamIndex=(?P<id_contract>)',
r'association/operations/.rechargement.contexte.html\?idBamIndex=(?P<id_contract>)',
r'professionnel/operations/.rechargement.contexte.html\?idBamIndex=(?P<id_contract>)', ContractsPage)
r'professionnel/operations/.rechargement.contexte.html\?idBamIndex=(?P<id_contract>)',
r'agriculteur/operations/.rechargement.contexte.html\?idBamIndex=(?P<id_contract>)',
r'entreprise/operations/.rechargement.contexte.html\?idBamIndex=(?P<id_contract>)', ContractsPage)
accounts_page = URL(r'particulier/operations/synthese.html',
r'association/operations/synthese.html',
r'professionnel/operations/synthese.html', AccountsPage)
r'professionnel/operations/synthese.html',
r'agriculteur/operations/synthese.html',
r'entreprise/operations/synthese.html', AccountsPage)
account_details = URL(r'particulier/operations/synthese/jcr:content.produits-valorisation.json/(?P<category>)',
r'association/operations/synthese/jcr:content.produits-valorisation.json/(?P<category>)',
r'professionnel/operations/synthese/jcr:content.produits-valorisation.json/(?P<category>)', AccountDetailsPage)
r'professionnel/operations/synthese/jcr:content.produits-valorisation.json/(?P<category>)',
r'agriculteur/operations/synthese/jcr:content.produits-valorisation.json/(?P<category>)',
r'entreprise/operations/synthese/jcr:content.produits-valorisation.json/(?P<category>)', AccountDetailsPage)
account_iban = URL(r'particulier/operations/operations-courantes/editer-rib/jcr:content.ibaninformation.json',
r'association/operations/operations-courantes/editer-rib/jcr:content.ibaninformation.json',
r'professionnel/operations/operations-courantes/editer-rib/jcr:content.ibaninformation.json', IbanPage)
history_page = URL(r'particulier/operations/synthese/detail-comptes/jcr:content.n3.compte.infos.json',
r'association/operations/synthese/detail-comptes/jcr:content.n3.compte.infos.json',
r'professionnel/operations/synthese/detail-comptes/jcr:content.n3.compte.infos.json', HistoryPage)
r'professionnel/operations/operations-courantes/editer-rib/jcr:content.ibaninformation.json',
r'agriculteur/operations/operations-courantes/editer-rib/jcr:content.ibaninformation.json',
r'entreprise/operations/operations-courantes/editer-rib/jcr:content.ibaninformation.json', IbanPage)
cards = URL(r'particulier/operations/moyens-paiement/mes-cartes/jcr:content.listeCartesParCompte.json',
r'association/operations/moyens-paiement/mes-cartes/jcr:content.listeCartesParCompte.json',
r'professionnel/operations/moyens-paiement/mes-cartes/jcr:content.listeCartesParCompte.json',
r'agriculteur/operations/moyens-paiement/mes-cartes/jcr:content.listeCartesParCompte.json',
r'entreprise/operations/moyens-paiement/mes-cartes/jcr:content.listeCartesParCompte.json', CardsPage)
history = URL(r'particulier/operations/synthese/detail-comptes/jcr:content.n3.operations.json',
r'association/operations/synthese/detail-comptes/jcr:content.n3.operations.json',
r'professionnel/operations/synthese/detail-comptes/jcr:content.n3.operations.json',
r'agriculteur/operations/synthese/detail-comptes/jcr:content.n3.operations.json',
r'entreprise/operations/synthese/detail-comptes/jcr:content.n3.operations.json', HistoryPage)
card_history = URL(r'particulier/operations/synthese/detail-comptes/jcr:content.n3.operations.encours.carte.debit.differe.json',
r'association/operations/synthese/detail-comptes/jcr:content.n3.operations.encours.carte.debit.differe.json',
r'professionnel/operations/synthese/detail-comptes/jcr:content.n3.operations.encours.carte.debit.differe.json',
r'agriculteur/operations/synthese/detail-comptes/jcr:content.n3.operations.encours.carte.debit.differe.json',
r'entreprise/operations/synthese/detail-comptes/jcr:content.n3.operations.encours.carte.debit.differe.json', CardHistoryPage)
netfinca_redirection = URL(r'particulier/operations/moco/catitres/jcr:content.init.html',
r'association/operations/moco/catitres/jcr:content.init.html',
r'professionnel/operations/moco/catitres/jcr:content.init.html',
r'agriculteur/operations/moco/catitres/jcr:content.init.html',
r'entreprise/operations/moco/catitres/jcr:content.init.html',
r'particulier/operations/moco/catitres/_jcr_content.init.html',
r'association/operations/moco/catitres/_jcr_content.init.html',
r'professionnel/operations/moco/catitres/_jcr_content.init.html',
r'agriculteur/operations/moco/catitres/_jcr_content.init.html',
r'entreprise/operations/moco/catitres/_jcr_content.init.html', NetfincaRedirectionPage)
predica_redirection = URL(r'particulier/operations/moco/predica/jcr:content.init.html',
r'association/operations/moco/predica/jcr:content.init.html',
r'professionnel/operations/moco/predica/jcr:content.init.html',
r'agriculteur/operations/moco/predica/jcr:content.init.html',
r'entreprise/operations/moco/predica/jcr:content.init.html', PredicaRedirectionPage)
predica_investments = URL(r'https://npcprediweb.predica.credit-agricole.fr/rest/detailEpargne/contrat/', PredicaInvestmentsPage)
profile_page = URL(r'particulier/operations/synthese/jcr:content.npc.store.client.json',
r'association/operations/synthese/jcr:content.npc.store.client.json',
r'professionnel/operations/synthese/jcr:content.npc.store.client.json', ProfilePage)
r'professionnel/operations/synthese/jcr:content.npc.store.client.json',
r'agriculteur/operations/synthese/jcr:content.npc.store.client.json',
r'entreprise/operations/synthese/jcr:content.npc.store.client.json', ProfilePage)
profile_details = URL(r'particulier/operations/profil/infos-personnelles/gerer-coordonnees.html', ProfileDetailsPage)
pro_profile_details = URL(r'association/operations/profil/infos-personnelles/controler-coordonnees.html',
r'professionnel/operations/profil/infos-personnelles/controler-coordonnees.html',
r'agriculteur/operations/profil/infos-personnelles/controler-coordonnees.html',
r'entreprise/operations/profil/infos-personnelles/controler-coordonnees.html', ProProfileDetailsPage)
def __init__(self, website, *args, **kwargs):
super(CragrAPI, self).__init__(*args, **kwargs)
......@@ -78,16 +136,17 @@ def __init__(self, website, *args, **kwargs):
self.BASEURL = 'https://%s/' % self.region
self.accounts_url = None
def do_login(self):
self.keypad.go()
keypad_password = self.page.build_password(self.password[:6])
keypad_id = self.page.get_keypad_id()
assert keypad_password, 'Could not obtain keypad password'
assert keypad_id, 'Could not obtain keypad id'
# Netfinca browser:
self.weboob = kwargs.pop('weboob')
dirname = self.responses_dirname
self.netfinca = NetfincaBrowser('', '', logger=self.logger, weboob=self.weboob, responses_dirname=dirname, proxy=self.PROXIES)
self.login_page.go()
# Get the form data to POST the security check:
form = self.page.get_login_form(self.username, keypad_password, keypad_id)
def deinit(self):
super(CragrAPI, self).deinit()
self.netfinca.deinit()
def do_login(self):
form = self.get_security_form()
try:
self.security_check.go(data=form)
except ServerError as exc:
......@@ -95,32 +154,65 @@ def do_login(self):
error = exc.response.json().get('error')
if error:
message = error.get('message', '')
if 'Votre identification est incorrecte' in message:
wrongpass_messages = ("Votre identification est incorrecte", "Vous n'avez plus droit")
if any(value in message for value in wrongpass_messages):
raise BrowserIncorrectPassword()
assert False, 'Unhandled Server Error encountered: %s' % error.get('message', '')
# accounts_url may contain '/particulier', '/professionnel' or '/association'
if 'obtenir un nouveau code' in message:
raise ActionNeeded(message)
technical_errors = ('Un incident technique', 'identifiant et votre code personnel')
if any(value in message for value in technical_errors):
# If it is a technical error, we try login again
form = self.get_security_form()
try:
self.security_check.go(data=form)
except ServerError as exc:
error = exc.response.json().get('error')
if error:
message = error.get('message', '')
if 'Un incident technique' in message:
raise BrowserUnavailable(message)
# accounts_url may contain '/particulier', '/professionnel', '/entreprise', '/agriculteur' or '/association'
self.accounts_url = self.page.get_accounts_url()
assert self.accounts_url, 'Could not get accounts url from security check'
self.location(self.accounts_url)
assert self.accounts_page.is_here(), 'We failed to login after the security check!'
assert self.accounts_page.is_here(), 'We failed to login after the security check: response URL is %s' % self.url
# Once the security check is passed, we are logged in.
def get_security_form(self):
self.keypad.go()
keypad_password = self.page.build_password(self.password[:6])
keypad_id = self.page.get_keypad_id()
assert keypad_password, 'Could not obtain keypad password'
assert keypad_id, 'Could not obtain keypad id'
self.login_page.go()
# Get the form data to POST the security check:
form = self.page.get_login_form(self.username, keypad_password, keypad_id)
return form
@need_login
def get_accounts_list(self):
# Determine how many spaces are present on the connection:
self.location(self.accounts_url)
if not self.accounts_page.is_here():
# We have been logged out.
self.do_login()
total_spaces = self.page.count_spaces()
self.logger.info('The total number of spaces on this connection is %s.' % total_spaces)
# Complete accounts list is required to match card parent accounts
# and to avoid accounts that are present on several spaces
all_accounts = {}
deferred_cards = {}
for contract in range(total_spaces):
# This request often returns a 500 error so we retry several times.
try:
self.contracts_page.go(id_contract=contract)
self.go_to_account_space(contract)
except ServerError:
self.logger.warning('Server returned error 500 when trying to access space %s, we try again' % contract)
try:
self.contracts_page.go(id_contract=contract)
self.go_to_account_space(contract)
except ServerError:
self.logger.warning('Server returned error 500 twice when trying to access space %s, this space will be skipped' % contract)
continue
......@@ -135,9 +227,19 @@ def get_accounts_list(self):
account._contract = contract
account.owner_type = self.page.get_owner_type()
# Some accounts have no balance in the main JSON, so we must
# get all the (id, balance) pairs in the account_details JSON:
categories = {int(account._category) for account in accounts_list if account._category != None}
''' Other accounts have no balance in the main JSON, so we must get all
the (_id_element_contrat, balance) pairs in the account_details JSON.
Account categories always correspond to the same account types:
# Category 1: Checking accounts,
# Category 2: To be determined,
# Category 3: Savings,
# Category 4: Loans & Credits,
# Category 5: Insurances (skipped),
# Category 6: To be determined,
# Category 7: Market accounts. '''
categories = {int(account._category) for account in accounts_list if account._category not in (None, '5')}
account_balances = {}
loan_ids = {}
for category in categories:
......@@ -145,7 +247,6 @@ def get_accounts_list(self):
account_balances.update(self.page.get_account_balances())
loan_ids.update(self.page.get_loan_ids())
# Getting IBANs for checking accounts
if main_account.type == Account.TYPE_CHECKING:
params = {
'compteIdx': int(main_account._index),
......@@ -155,76 +256,265 @@ def get_accounts_list(self):
iban = self.page.get_iban()
if is_iban_valid(iban):
main_account.iban = iban
yield main_account
for card in main_account._cards:
card.parent = main_account
card.currency = main_account.currency
card.owner_type = main_account.owner_type
card._contract = contract
yield card
if main_account.id not in all_accounts:
all_accounts[main_account.id] = main_account
yield main_account
for account in accounts_list:
if empty(account.balance):
account.balance = account_balances.get(account.id, NotAvailable)
account.balance = account_balances.get(account._id_element_contrat, NotAvailable)
if account.type == Account.TYPE_CHECKING:
try:
params = {
'compteIdx': int(account._index),
'grandeFamilleCode': 1,
}
self.account_iban.go(params=params)
iban = self.page.get_iban()
if is_iban_valid(iban):
account.iban = iban
except ServerError:
self.logger.warning('Could not fetch IBAN for checking account "%s %s"', account.label, account.id)
pass
# TO-DO: Create Loan() object with its related attributes
params = {
'compteIdx': int(account._index),
'grandeFamilleCode': int(account._category),
}
self.account_iban.go(params=params)
iban = self.page.get_iban()
if is_iban_valid(iban):
account.iban = iban
# Loans have a specific ID that we need to fetch
# so the backend can match loans properly.
# If no there is no loan ID, we keep the account ID.
if account.type == Account.TYPE_LOAN:
account.id = loan_ids.get(account.id, account.id)
account.balance = -account.balance
account.id = account.number = loan_ids.get(account._id_element_contrat, account.id)
account = self.switch_account_to_loan(account)
elif account.type == Account.TYPE_REVOLVING_CREDIT:
account.id = loan_ids.get(account.id, account.id)
account.balance = 0
yield account
account.id = account.number = loan_ids.get(account._id_element_contrat, account.id)
account = self.switch_account_to_revolving(account)
if account.id not in all_accounts:
all_accounts[account.id] = account
yield account
# Fetch all deferred credit cards for this space
self.cards.go()
for card in self.page.iter_card_parents():
card.number = card.id
card.parent = all_accounts.get(card._parent_id, NotAvailable)
card.currency = card.parent.currency
card.owner_type = card.parent.owner_type
card._category = card.parent._category
card._contract = contract
if card.id not in deferred_cards:
deferred_cards[card.id] = card
# We must check if cards are unique on their parent account;
# if not, we cannot retrieve their summaries in iter_history.
parent_accounts = []
for card in deferred_cards.values():
parent_accounts.append(card.parent.id)
for card in deferred_cards.values():
if parent_accounts.count(card.parent.id) == 1:
card._unique = True
else:
card._unique = False
yield card
def switch_account_to_loan(self, account):
loan = Loan()
copy_attrs = ('id', 'number', 'label', 'type', 'currency', '_index', '_category', '_contract', '_id_element_contrat', 'owner_type')
for attr in copy_attrs:
setattr(loan, attr, getattr(account, attr))
loan.balance = -account.balance
return loan
def switch_account_to_revolving(self, account):
loan = Loan()
copy_attrs = ('id', 'number', 'label', 'type', 'currency', '_index', '_category', '_contract', '_id_element_contrat', 'owner_type')
for attr in copy_attrs:
setattr(loan, attr, getattr(account, attr))
loan.balance = Decimal(0)
loan.available_amount = account.balance
return loan
@need_login
def go_to_account_space(self, contract):
# TO-DO: Figure out a way to determine whether
# we already are on the right account space
self.contracts_page.go(id_contract=contract)
assert self.accounts_page.is_here()
if not self.accounts_page.is_here():
# We have been logged out.
self.do_login()
self.contracts_page.go(id_contract=contract)
assert self.accounts_page.is_here()
@need_login
def get_card(self, id):
return find_object(self.get_cards(), id=id)
@need_login
def get_cards(self, accounts_list=None):
# accounts_list is only used by get_list
raise BrowserUnavailable()
@need_login
def get_history(self, account):
raise BrowserUnavailable()
def get_history(self, account, coming=False):
if account.type == Account.TYPE_CARD:
card_transactions = []
self.go_to_account_space(account._contract)
# Deferred cards transactions have a specific JSON.
# Only three months of history available for cards.
value = 0 if coming else 1
params = {
'grandeFamilleCode': int(account._category),
'compteIdx': int(account.parent._index),
'carteIdx': int(account._index),
'rechercheEncoursDebite': value
}
self.card_history.go(params=params)
for tr in self.page.iter_card_history():
card_transactions.append(tr)
# If the card if not unique on the parent id, it is impossible
# to know which summary corresponds to which card.
if not coming and card_transactions and account._unique:
# Get card summaries from parent account
# until we reach the oldest card transaction
last_transaction = card_transactions[-1]
before_last_transaction = False
params = {
'compteIdx': int(account.parent._index),
'grandeFamilleCode': int(account.parent._category),
'idDevise': str(account.parent.currency),
'idElementContrat': str(account.parent._id_element_contrat),
}
self.history.go(params=params)
for tr in self.page.iter_history():
if tr.date < last_transaction.date:
before_last_transaction = True
break
if tr.type == Transaction.TYPE_CARD_SUMMARY:
tr.amount = -tr.amount
card_transactions.append(tr)
while self.page.has_next_page() and not before_last_transaction:
next_index = self.page.get_next_index()
params = {
'grandeFamilleCode': int(account.parent._category),
'compteIdx': int(account.parent._index),
'idDevise': str(account.parent.currency),
'startIndex': next_index,
'count': 100,
}
self.history.go(params=params)
for tr in self.page.iter_history():
if tr.date < last_transaction.date:
before_last_transaction = True
break
if tr.type == Transaction.TYPE_CARD_SUMMARY:
tr.amount = -tr.amount
card_transactions.append(tr)
for tr in sorted_transactions(card_transactions):
yield tr
return
# These three parameters are required to get the transactions for non_card accounts
if empty(account._index) or empty(account._category) or empty(account._id_element_contrat):
return
self.go_to_account_space(account._contract)
params = {
'compteIdx': int(account._index),
'grandeFamilleCode': int(account._category),
'idDevise': str(account.currency),
'idElementContrat': str(account._id_element_contrat),
}
self.history.go(params=params)
for tr in self.page.iter_history():
yield tr
# Get other transactions 100 by 100:
while self.page.has_next_page():
next_index = self.page.get_next_index()
params = {
'grandeFamilleCode': int(account._category),
'compteIdx': int(account._index),
'idDevise': str(account.currency),
'startIndex': next_index,
'count': 100,
}
self.history.go(params=params)
for tr in self.page.iter_history():
yield tr
@need_login
def iter_investment(self, account):
raise BrowserUnavailable()
if account.type in (Account.TYPE_PERP, Account.TYPE_PERCO, Account.TYPE_LIFE_INSURANCE, Account.TYPE_CAPITALISATION):
if account.label == "Vers l'avenir":
# Website crashes when clicking on these Life Insurances...
return
self.go_to_account_space(account._contract)
token = self.token_page.go().get_token()
data = {
'situation_travail': 'CONTRAT',
'idelco': account.id,
':cq_csrf_token': token,
}
self.predica_redirection.go(data=data)
self.predica_investments.go()
for inv in self.page.iter_investments():
yield inv
elif account.type == Account.TYPE_PEA and account.label == 'Compte espèce PEA':
yield create_french_liquidity(account.balance)
return
elif account.type in (Account.TYPE_PEA, Account.TYPE_MARKET):
# Do not try to get to Netfinca if there is no money
# on the account or the server will return an error 500
if account.balance == 0:
return
self.go_to_account_space(account._contract)
token = self.token_page.go().get_token()
data = {
'situation_travail': 'BANCAIRE',
'num_compte': account.id,
'code_fam_produit': account._fam_product_code,
'code_fam_contrat_compte': account._fam_contract_code,
':cq_csrf_token': token,
}
# For some market accounts, investments are not even accessible,
# and the only way to know if there are investments is to try
# to go to the Netfinca space with the accounts parameters.
try:
self.netfinca_redirection.go(data=data)
except BrowserHTTPNotFound:
self.logger.info('Investments are not available for this account.')
self.go_to_account_space(account._contract)
return
url = self.page.get_url()
if 'netfinca' in url:
self.location(url)
self.netfinca.session.cookies.update(self.session.cookies)
self.netfinca.accounts.go()
for inv in self.netfinca.iter_investments(account):
if inv.code == 'XX-liquidity' and account.type == Account.TYPE_PEA:
# Liquidities are already fetched on the "PEA espèces"
continue
yield inv
@need_login
def iter_advisor(self):
raise BrowserUnavailable()
self.go_to_account_space(0)
owner_type = self.page.get_owner_type()
self.profile_page.go()
if owner_type == 'PRIV':
advisor = self.page.get_advisor()
self.profile_details.go()
self.page.fill_advisor(obj=advisor)
return advisor
elif owner_type == 'ORGA':
advisor = self.page.get_advisor()
self.pro_profile_details.go()
self.page.fill_advisor(obj=advisor)
return advisor
@need_login
def get_profile(self):
#self.profile.go()
raise BrowserUnavailable()
# There is one profile per space, so we only fetch the first one
self.go_to_account_space(0)
owner_type = self.page.get_owner_type()
self.profile_page.go()
if owner_type == 'PRIV':
profile = self.page.get_user_profile()
self.profile_details.go()
self.page.fill_profile(obj=profile)
return profile
elif owner_type == 'ORGA':
profile = self.page.get_company_profile()
self.pro_profile_details.go()
self.page.fill_profile(obj=profile)
return profile
@need_login
def iter_transfer_recipients(self, account):
......
modules/cragr/api/netfinca_browser.py 0 → 100644
View file @ 09df761c
# -*- coding: utf-8 -*-
# Copyright(C) 2012-2019 Budget Insight
from weboob.browser import AbstractBrowser
class NetfincaBrowser(AbstractBrowser):
PARENT = 'netfinca'
BASEURL = 'https://www.cabourse.credit-agricole.fr'
modules/cragr/api/pages.py
View file @ 09df761c
......@@ -22,19 +22,23 @@
from decimal import Decimal
import re
import json
import dateutil
from weboob.browser.pages import HTMLPage, JsonPage, LoggedPage
from weboob.exceptions import BrowserUnavailable
from weboob.exceptions import ActionNeeded
from weboob.capabilities import NotAvailable
from weboob.capabilities.bank import (
Account, AccountOwnerType,
Account, AccountOwnerType, Transaction, Investment,
)
from weboob.capabilities.profile import Person, Company
from weboob.capabilities.contact import Advisor
from weboob.browser.elements import DictElement, ItemElement, method
from weboob.browser.filters.standard import (
CleanText, CleanDecimal, Currency as CleanCurrency, Format, Field, Map, Eval
CleanText, CleanDecimal, Currency as CleanCurrency, Format, Field, Map, Eval, Env, Regexp, Date,
)
from weboob.browser.filters.html import Attr
from weboob.browser.filters.json import Dict
from weboob.tools.capabilities.bank.investments import is_isin_valid
def float_to_decimal(f):
......@@ -54,7 +58,7 @@ def get_keypad_id(self):
class LoginPage(HTMLPage):
def get_login_form(self, username, keypad_password, keypad_id):
form = self.get_form(id="loginForm")
form['j_username'] = username
form['j_username'] = username[:11]
form['j_password'] = keypad_password
form['keypadId'] = keypad_id
return form
......@@ -64,9 +68,12 @@ class LoggedOutPage(HTMLPage):
def is_here(self):
return self.doc.xpath('//b[text()="FIN DE CONNEXION"]')
class FirstConnectionPage(LoggedPage, HTMLPage):
def on_load(self):
self.logger.warning('We have been logged out!')
raise BrowserUnavailable()
message = CleanText('//p[contains(text(), "votre première visite")]')(self.doc)
if message:
raise ActionNeeded(message)
class SecurityPage(JsonPage):
......@@ -74,6 +81,11 @@ def get_accounts_url(self):
return Dict('url')(self.doc)
class TokenPage(LoggedPage, JsonPage):
def get_token(self):
return Dict('token')(self.doc)
class ContractsPage(LoggedPage, HTMLPage):
pass
......@@ -102,12 +114,15 @@ class ContractsPage(LoggedPage, HTMLPage):
'DAV TIGERE': Account.TYPE_SAVINGS,
'CPTEXCPRO': Account.TYPE_SAVINGS,
'CPTEXCENT': Account.TYPE_SAVINGS,
'DAT': Account.TYPE_SAVINGS,
'PRET PERSO': Account.TYPE_LOAN,
'P. ENTREPR': Account.TYPE_LOAN,
'P. HABITAT': Account.TYPE_LOAN,
'PRET 0%': Account.TYPE_LOAN,
'INV PRO': Account.TYPE_LOAN,
'TRES. PRO': Account.TYPE_LOAN,
'CT ATT HAB': Account.TYPE_LOAN,
'PRET CEL': Account.TYPE_LOAN,
'PEA': Account.TYPE_PEA,
'PEAP': Account.TYPE_PEA,
'DAV PEA': Account.TYPE_PEA,
......@@ -165,8 +180,10 @@ class get_main_account(ItemElement):
obj_balance = Eval(float_to_decimal, Dict('comptePrincipal/solde'))
obj_currency = CleanCurrency(Dict('comptePrincipal/idDevise'))
obj__index = Dict('comptePrincipal/index')
obj__category = None # Main accounts have no category
obj__category = Dict('comptePrincipal/grandeFamilleProduitCode', default=None)
obj__id_element_contrat = CleanText(Dict('comptePrincipal/idElementContrat'))
obj__fam_product_code = CleanText(Dict('comptePrincipal/codeFamilleProduitBam'))
obj__fam_contract_code = CleanText(Dict('comptePrincipal/codeFamilleContratBam'))
def obj_type(self):
_type = Map(CleanText(Dict('comptePrincipal/libelleUsuelProduit')), ACCOUNT_TYPES, Account.TYPE_UNKNOWN)(self)
......@@ -174,22 +191,6 @@ def obj_type(self):
self.logger.warning('We got an untyped account: please add "%s" to ACCOUNT_TYPES.' % CleanText(Dict('comptePrincipal/libelleUsuelProduit'))(self))
return _type
class obj__cards(DictElement):
item_xpath = 'comptePrincipal/cartesDD'
class item(ItemElement):
klass = Account
def obj_id(self):
return CleanText(Dict('idCarte'))(self).replace(' ', '')
obj_label = Format('Carte %s %s', Field('id'), CleanText(Dict('titulaire')))
obj_type = Account.TYPE_CARD
obj_coming = Eval(float_to_decimal, Dict('encoursCarteM'))
obj_balance = CleanDecimal(0)
obj__index = Dict('index')
obj__category = None
@method
class iter_accounts(DictElement):
item_xpath = 'grandesFamilles/*/elementsContrats'
......@@ -199,18 +200,28 @@ class item(ItemElement):
klass = Account
obj_id = CleanText(Dict('numeroCompteBam'))
obj_number = CleanText(Dict('numeroCompteBam'))
def obj_id(self):
# Loan ids may be duplicated so we use the contract number for now:
if Field('type')(self) == Account.TYPE_LOAN:
return CleanText(Dict('idElementContrat'))(self)
return CleanText(Dict('numeroCompte'))(self)
obj_number = CleanText(Dict('numeroCompte'))
obj_label = CleanText(Dict('libelleProduit'))
obj_currency = CleanCurrency(Dict('idDevise'))
obj__index = Dict('index')
obj__category = Dict('grandeFamilleProduitCode', default=None)
obj__id_element_contrat = CleanText(Dict('idElementContrat'))
obj__fam_product_code = CleanText(Dict('codeFamilleProduitBam'))
obj__fam_contract_code = CleanText(Dict('codeFamilleContratBam'))
def obj_type(self):
if CleanText(Dict('libelleUsuelProduit'))(self) in ('HABITATION',):
# No need to log warning for "assurance" accounts
return NotAvailable
_type = Map(CleanText(Dict('libelleUsuelProduit')), ACCOUNT_TYPES, Account.TYPE_UNKNOWN)(self)
if _type == Account.TYPE_UNKNOWN:
self.logger.warning('We got an untyped account: please add "%s" to ACCOUNT_TYPES.' % CleanText(Dict('libelleUsuelProduit'))(self))
self.logger.warning('There is an untyped account: please add "%s" to ACCOUNT_TYPES.' % CleanText(Dict('libelleUsuelProduit'))(self))
return _type
def obj_balance(self):
......@@ -227,22 +238,30 @@ def condition(self):
class AccountDetailsPage(LoggedPage, JsonPage):
def get_account_balances(self):
# We use the 'idElementContrat' key because it is unique
# whereas the account id may not be unique for Loans
account_balances = {}
for el in self.doc:
# Insurances have no balance, we skip them
if el.get('typeProduit') == 'assurance':
continue
value = el.get('solde', el.get('encoursActuel', el.get('valorisationContrat', el.get('montantRestantDu', el.get('capitalDisponible')))))
assert value is not None, 'Could not find the account balance'
account_balances[Dict('numeroCompte')(el)] = float_to_decimal(value)
if value is None:
continue
account_balances[Dict('idElementContrat')(el)] = float_to_decimal(value)
return account_balances
def get_loan_ids(self):
# We use the 'idElementContrat' key because it is unique
# whereas the account id may not be unique for Loans
loan_ids = {}
for el in self.doc:
if el.get('numeroCredit'):
# Loans
loan_ids[Dict('numeroCompte')(el)] = Dict('numeroCredit')(el)
loan_ids[Dict('idElementContrat')(el)] = Dict('numeroCredit')(el)
elif el.get('numeroContrat'):
# Revolving credits
loan_ids[Dict('numeroCompte')(el)] = Dict('numeroContrat')(el)
loan_ids[Dict('idElementContrat')(el)] = Dict('numeroContrat')(el)
return loan_ids
......@@ -252,12 +271,191 @@ def get_iban(self):
class HistoryPage(LoggedPage, JsonPage):
pass
def has_next_page(self):
return Dict('hasNext')(self.doc)
def get_next_index(self):
return Dict('nextSetStartIndex')(self.doc)
class InvestmentPage(LoggedPage, JsonPage):
pass
@method
class iter_history(DictElement):
item_xpath = 'listeOperations'
class item(ItemElement):
TRANSACTION_TYPES = {
'PAIEMENT PAR CARTE': Transaction.TYPE_CARD,
'REMISE CARTE': Transaction.TYPE_CARD,
'PRELEVEMENT CARTE': Transaction.TYPE_CARD_SUMMARY,
'RETRAIT AU DISTRIBUTEUR': Transaction.TYPE_WITHDRAWAL,
"RETRAIT MUR D'ARGENT": Transaction.TYPE_WITHDRAWAL,
'FRAIS': Transaction.TYPE_BANK,
'COTISATION': Transaction.TYPE_BANK,
'VIREMENT': Transaction.TYPE_TRANSFER,
'VIREMENT EN VOTRE FAVEUR': Transaction.TYPE_TRANSFER,
'VIREMENT EMIS': Transaction.TYPE_TRANSFER,
'CHEQUE EMIS': Transaction.TYPE_CHECK,
'REMISE DE CHEQUE': Transaction.TYPE_DEPOSIT,
'PRELEVEMENT': Transaction.TYPE_ORDER,
'PRELEVT': Transaction.TYPE_ORDER,
'PRELEVMNT': Transaction.TYPE_ORDER,
'REMBOURSEMENT DE PRET': Transaction.TYPE_LOAN_PAYMENT,
}
klass = Transaction
obj_raw = Format('%s %s %s', CleanText(Dict('libelleTypeOperation')), CleanText(Dict('libelleOperation')), CleanText(Dict('libelleComplementaire')))
obj_label = Format('%s %s', CleanText(Dict('libelleTypeOperation')), CleanText(Dict('libelleOperation')))
obj_amount = Eval(float_to_decimal, Dict('montant'))
obj_type = Map(CleanText(Dict('libelleTypeOperation')), TRANSACTION_TYPES, Transaction.TYPE_UNKNOWN)
def obj_date(self):
return dateutil.parser.parse(Dict('dateValeur')(self))
def obj_rdate(self):
return dateutil.parser.parse(Dict('dateOperation')(self))
class CardsPage(LoggedPage, JsonPage):
@method
class iter_card_parents(DictElement):
item_xpath = 'comptes'
class iter_cards(DictElement):
item_xpath = 'listeCartes'
def parse(self, el):
self.env['parent_id'] = Dict('idCompte')(el)
class item(ItemElement):
klass = Account
def obj_id(self):
return CleanText(Dict('idCarte'))(self).replace(' ', '')
def condition(self):
assert CleanText(Dict('codeTypeDebitPaiementCarte'))(self) in ('D', 'I')
return CleanText(Dict('codeTypeDebitPaiementCarte'))(self)=='D'
obj_label = Format('Carte %s %s', Field('id'), CleanText(Dict('titulaire')))
obj_type = Account.TYPE_CARD
obj_coming = Eval(lambda x: -float_to_decimal(x), Dict('encoursCarteM'))
obj_balance = CleanDecimal(0)
obj__parent_id = Env('parent_id')
obj__index = Dict('index')
obj__id_element_contrat = None
class CardHistoryPage(LoggedPage, JsonPage):
@method
class iter_card_history(DictElement):
item_xpath = None
class item(ItemElement):
klass = Transaction
obj_raw = CleanText(Dict('libelleOperation'))
obj_label = CleanText(Dict('libelleOperation'))
obj_amount = Eval(float_to_decimal, Dict('montant'))
obj_type = Transaction.TYPE_DEFERRED_CARD
def obj_date(self):
return dateutil.parser.parse(Dict('datePrelevement')(self))
def obj_rdate(self):
return dateutil.parser.parse(Dict('dateOperation')(self))
class NetfincaRedirectionPage(LoggedPage, HTMLPage):
def get_url(self):
return Regexp(Attr('//body', 'onload'), r'document.location="([^"]+)"')(self.doc)
class PredicaRedirectionPage(LoggedPage, HTMLPage):
def on_load(self):
form = self.get_form()
form.submit()
class PredicaInvestmentsPage(LoggedPage, JsonPage):
@method
class iter_investments(DictElement):
item_xpath = 'listeSupports/support'
class item(ItemElement):
klass = Investment
obj_label = CleanText(Dict('lcspt'))
obj_valuation = Eval(float_to_decimal, Dict('mtvalspt'))
def obj_portfolio_share(self):
portfolio_share = Dict('txrpaspt', default=None)(self)
if portfolio_share:
return Eval(lambda x: float_to_decimal(x / 100), portfolio_share)(self)
return NotAvailable
def obj_unitvalue(self):
unit_value = Dict('mtliqpaaspt', default=None)(self)
if unit_value:
return Eval(float_to_decimal, unit_value)(self)
return NotAvailable
def obj_quantity(self):
quantity = Dict('qtpaaspt', default=None)(self)
if quantity:
return Eval(float_to_decimal, quantity)(self)
return NotAvailable
def obj_code(self):
code = Dict('cdsptisn')(self)
if is_isin_valid(code):
return code
return NotAvailable
def obj_code_type(self):
if is_isin_valid(Field('code')(self)):
return Investment.CODE_TYPE_ISIN
return NotAvailable
class ProfilePage(LoggedPage, JsonPage):
pass
\ No newline at end of file
@method
class get_user_profile(ItemElement):
klass = Person
obj_name = CleanText(Dict('displayName', default=NotAvailable))
obj_phone = CleanText(Dict('branchPhone', default=NotAvailable))
obj_birth_date = Date(Dict('birthdate', default=NotAvailable))
@method
class get_company_profile(ItemElement):
klass = Company
obj_name = CleanText(Dict('displayName', default=NotAvailable))
obj_phone = CleanText(Dict('branchPhone', default=NotAvailable))
obj_registration_date = Date(Dict('birthdate', default=NotAvailable))
@method
class get_advisor(ItemElement):
klass = Advisor
def obj_name(self):
# If no advisor is displayed, we return the agency advisor.
if Dict('advisorGivenName')(self) and Dict('advisorFamilyName')(self):
return Format('%s %s', CleanText(Dict('advisorGivenName')), CleanText(Dict('advisorFamilyName')))(self)
return Format('%s %s', CleanText(Dict('branchManagerGivenName')), CleanText(Dict('branchManagerFamilyName')))(self)
class ProfileDetailsPage(LoggedPage, HTMLPage):
@method
class fill_profile(ItemElement):
obj_email = CleanText('//p[contains(@class, "Data mail")]', default=NotAvailable)
obj_address = CleanText('//p[strong[contains(text(), "Adresse")]]/text()[2]', default=NotAvailable)
@method
class fill_advisor(ItemElement):
obj_phone = CleanText('//div[@id="blockConseiller"]//a[contains(@class, "advisorNumber")]', default=NotAvailable)
class ProProfileDetailsPage(ProfileDetailsPage):
pass
modules/cragr/module.py
View file @ 09df761c
......@@ -84,7 +84,7 @@ class CragrModule(Module, CapBankWealth, CapBankTransferAddRecipient, CapContact
}.items())])
CONFIG = BackendConfig(Value('website', label=u'Région', choices=website_choices),
ValueBackendPassword('login', label=u'N° de compte', masked=False),
ValueBackendPassword('login', label=u'N° de compte', masked=False, regexp=r'\d+'),
ValueBackendPassword('password', label=u'Code personnel', regexp=r'\d{6}'))
BROWSER = ProxyBrowser
......@@ -98,7 +98,8 @@ def create_default_browser(self):
site_conf = self.COMPAT_DOMAINS.get(site_conf, site_conf)
return self.create_browser(site_conf,
self.config['login'].get(),
self.config['password'].get())
self.config['password'].get(),
weboob=self.weboob)
def iter_accounts(self):
return self.browser.get_accounts_list()
......@@ -114,15 +115,17 @@ def to_date(obj):
return obj.date()
return obj
for tr in self.browser.get_history(account):
for tr in self.browser.get_history(account, coming):
tr_coming = to_date(tr.date) > today
if coming == tr_coming:
yield tr
elif coming:
break
def iter_history(self, account):
if account.type == Account.TYPE_CARD:
return self._history_filter(account, False)
return self.browser.get_history(account)
return self.browser.get_history(account, False)
def iter_coming(self, account):
if account.type == Account.TYPE_CARD:
......
modules/cragr/web/browser.py
View file @ 09df761c
......@@ -412,7 +412,7 @@ def market_accounts_matching(self, accounts_list, market_accounts_list):
account.balance = self.page.get_pea_balance()
@need_login
def get_history(self, account):
def get_history(self, account, coming=False):
if account.type in (Account.TYPE_MARKET, Account.TYPE_PEA, Account.TYPE_LIFE_INSURANCE, Account.TYPE_PERP):
self.logger.warning('This account is not supported')
raise NotImplementedError()
......
modules/creditdunord/module.py
View file @ 09df761c
......@@ -83,7 +83,7 @@ def iter_history(self, account):
yield tr
def iter_coming(self, account):
account = self.get_account(account.id)
account = self.get_account_for_history(account.id)
for tr in self.browser.get_history(account, coming=True):
if tr._is_coming:
yield tr
......
modules/creditmutuel/browser.py
View file @ 09df761c
......@@ -312,8 +312,11 @@ def list_operations(self, page, account):
form.pop(k, None)
form.submit()
# IndexError when form xpath returns [], StopIteration if next called on empty iterable
except (IndexError, StopIteration, FormNotFound):
except (StopIteration, FormNotFound):
self.logger.warning('Could not get history on new website')
except IndexError:
# 6 months history is not available
pass
while self.page:
try:
......
modules/creditmutuel/pages.py
View file @ 09df761c
......@@ -907,7 +907,11 @@ def obj_type(self):
def obj_original_amount(self):
m = re.search(r'(([\s-]\d+)+,\d+)', CleanText(TableCell('commerce'))(self))
if m and not 'FRAIS' in CleanText(TableCell('commerce'))(self):
return Decimal(m.group(1).replace(',', '.').replace(' ', '')).quantize(Decimal('0.01'))
matched_text = m.group(1)
submatch = re.search(r'\d+-(.*)', matched_text)
if submatch:
matched_text = submatch.group(1)
return Decimal(matched_text.replace(',', '.').replace(' ', '')).quantize(Decimal('0.01'))
return NotAvailable
def obj_original_currency(self):
......@@ -979,7 +983,11 @@ def obj_type(self):
def obj_original_amount(self):
m = re.search(r'(([\s-]\d+)+,\d+)', CleanText(TableCell('commerce'))(self))
if m and not 'FRAIS' in CleanText(TableCell('commerce'))(self):
return Decimal(m.group(1).replace(',', '.').replace(' ', '')).quantize(Decimal('0.01'))
matched_text = m.group(1)
submatch = re.search(r'\d+-(.*)', matched_text)
if submatch:
matched_text = submatch.group(1)
return Decimal(matched_text.replace(',', '.').replace(' ', '')).quantize(Decimal('0.01'))
return NotAvailable
def obj_original_currency(self):
......@@ -1105,11 +1113,11 @@ def obj_commission(self):
@method
class iter_investment(TableElement):
item_xpath = '//table[has-class("liste")]/tbody/tr[count(td)>=7]'
head_xpath = '//table[has-class("liste")]/thead/tr/th'
item_xpath = '//table[has-class("liste") and not (@summary="Avances")]/tbody/tr[count(td)>=7]'
head_xpath = '//table[has-class("liste") and not (@summary="Avances")]/thead/tr/th'
col_label = 'Support'
col_unitprice = re.compile(r"^Prix d'achat moyen")
col_unitprice = re.compile(r'Prix')
col_vdate = re.compile(r'Date de cotation')
col_unitvalue = 'Valeur de la part'
col_quantity = 'Nombre de parts'
......
modules/fortuneo/browser.py
View file @ 09df761c
......@@ -32,7 +32,7 @@
from .pages.login import LoginPage, UnavailablePage
from .pages.accounts_list import (
AccountsList, AccountHistoryPage, CardHistoryPage, InvestmentHistoryPage, PeaHistoryPage, LoanPage, ProfilePage, ProfilePageCSV, SecurityPage,
AccountsList, AccountHistoryPage, CardHistoryPage, InvestmentHistoryPage, PeaHistoryPage, LoanPage, ProfilePage, ProfilePageCSV, SecurityPage, FakeActionPage,
)
from .pages.transfer import (
RegisterTransferPage, ValidateTransferPage, ConfirmTransferPage, RecipientsPage, RecipientSMSPage
......@@ -84,7 +84,7 @@ class Fortuneo(LoginBrowser, StatesMixin):
r'fr/prive/mes-comptes/compte-courant/.*/init-confirmer-saisie-virement.jsp',
r'/fr/prive/mes-comptes/compte-courant/.*/confirmer-saisie-virement.jsp',
ConfirmTransferPage)
fake_action_page = URL(r'fr/prive/mes-comptes/synthese-globale/synthese-mes-comptes.jsp', FakeActionPage)
profile = URL(r'/fr/prive/informations-client.jsp', ProfilePage)
profile_csv = URL(r'/PdfStruts\?*', ProfilePageCSV)
......@@ -160,7 +160,17 @@ def get_accounts_list(self):
self.process_action_needed()
assert self.accounts_page.is_here()
return self.page.get_list()
accounts_list = self.page.get_list()
if self.fake_action_page.is_here():
# A false action needed is present, it's a choice to make Fortuno your main bank.
# To avoid it, we need to first detect it on the account_page
# Then make a post request to mimic the click on choice 'later'
# And to finish we must to reload the page with a POST to get the accounts
# before going on the accounts_page, which will have the data.
self.location(self.absurl('ReloadContext?action=1&', base=True), method='POST')
self.accounts_page.go()
accounts_list = self.page.get_list()
return accounts_list
def process_action_needed(self):
# we have to go in an iframe to know if there are CGUs
......