Skip to content
Commits on Source (39)
......@@ -56,6 +56,9 @@ def iter_accounts(self):
@need_login
def iter_investment(self, account):
if account.balance == 0:
self.logger.info('Account %s has a null balance, no investment available.', account.label)
return
headers = {'X-noee-authorization': 'noeprd %s' % self.token}
self.accounts.go(headers=headers)
for inv in self.page.iter_investments(account_id=account.id):
......
......@@ -515,7 +515,7 @@ class item(ItemElement):
obj_code = Regexp(CleanText(TableCell('code')), r'(.{12})', default=NotAvailable)
obj_code_type = lambda self: Investment.CODE_TYPE_ISIN if Field('code')(self) is not NotAvailable else NotAvailable
def obj_diff_percent(self):
def obj_diff_ratio(self):
diff_percent = MyDecimal(TableCell('diff')(self)[0])(self)
return diff_percent/100 if diff_percent != NotAvailable else diff_percent
......
......@@ -17,6 +17,8 @@
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from collections import OrderedDict
from functools import reduce
......@@ -34,38 +36,38 @@
class BanquePopulaireModule(Module, CapBankWealth, CapContact, CapProfile):
NAME = 'banquepopulaire'
MAINTAINER = u'Romain Bignon'
MAINTAINER = 'Romain Bignon'
EMAIL = 'romain@weboob.org'
VERSION = '1.6'
DESCRIPTION = u'Banque Populaire'
DESCRIPTION = 'Banque Populaire'
LICENSE = 'LGPLv3+'
website_choices = OrderedDict([(k, u'%s (%s)' % (v, k)) for k, v in sorted({
'www.ibps.alpes.banquepopulaire.fr': u'Alpes',
'www.ibps.alsace.banquepopulaire.fr': u'Alsace Lorraine Champagne',
'www.ibps.bpalc.banquepopulaire.fr' : u'Alsace Lorraine Champagne',
'www.ibps.bpaca.banquepopulaire.fr': u'Aquitaine Centre atlantique',
'www.ibps.atlantique.banquepopulaire.fr': u'Atlantique',
'www.ibps.bpgo.banquepopulaire.fr': u'Grand Ouest',
'www.ibps.loirelyonnais.banquepopulaire.fr': u'Auvergne Rhône Alpes',
'www.ibps.bpaura.banquepopulaire.fr': u'Auvergne Rhône Alpes',
'www.ibps.banquedesavoie.banquepopulaire.fr': u'Banque de Savoie',
'www.ibps.bpbfc.banquepopulaire.fr': u'Bourgogne Franche-Comté',
'www.ibps.bretagnenormandie.cmm.groupe.banquepopulaire.fr': u'Crédit Maritime Bretagne Normandie',
'www.ibps.atlantique.creditmaritime.groupe.banquepopulaire.fr': u'Crédit Maritime Atlantique',
'www.ibps.sudouest.creditmaritime.groupe.banquepopulaire.fr': u'Crédit Maritime du Littoral du Sud-Ouest',
'www.ibps.lorrainechampagne.banquepopulaire.fr': u'Lorraine Champagne',
'www.ibps.massifcentral.banquepopulaire.fr': u'Massif central',
'www.ibps.mediterranee.banquepopulaire.fr': u'Méditerranée',
'www.ibps.nord.banquepopulaire.fr': u'Nord',
'www.ibps.occitane.banquepopulaire.fr': u'Occitane',
'www.ibps.ouest.banquepopulaire.fr': u'Ouest',
'www.ibps.rivesparis.banquepopulaire.fr': u'Rives de Paris',
'www.ibps.sud.banquepopulaire.fr': u'Sud',
'www.ibps.valdefrance.banquepopulaire.fr': u'Val de France',
website_choices = OrderedDict([(k, '%s (%s)' % (v, k)) for k, v in sorted({
'www.ibps.alpes.banquepopulaire.fr': 'Alpes',
'www.ibps.alsace.banquepopulaire.fr': 'Alsace Lorraine Champagne',
'www.ibps.bpalc.banquepopulaire.fr' : 'Alsace Lorraine Champagne',
'www.ibps.bpaca.banquepopulaire.fr': 'Aquitaine Centre atlantique',
'www.ibps.atlantique.banquepopulaire.fr': 'Atlantique',
'www.ibps.bpgo.banquepopulaire.fr': 'Grand Ouest',
'www.ibps.loirelyonnais.banquepopulaire.fr': 'Auvergne Rhône Alpes',
'www.ibps.bpaura.banquepopulaire.fr': 'Auvergne Rhône Alpes',
'www.ibps.banquedesavoie.banquepopulaire.fr': 'Banque de Savoie',
'www.ibps.bpbfc.banquepopulaire.fr': 'Bourgogne Franche-Comté',
'www.ibps.bretagnenormandie.cmm.groupe.banquepopulaire.fr': 'Crédit Maritime Bretagne Normandie',
'www.ibps.atlantique.creditmaritime.groupe.banquepopulaire.fr': 'Crédit Maritime Atlantique',
'www.ibps.sudouest.creditmaritime.groupe.banquepopulaire.fr': 'Crédit Maritime du Littoral du Sud-Ouest',
'www.ibps.lorrainechampagne.banquepopulaire.fr': 'Lorraine Champagne',
'www.ibps.massifcentral.banquepopulaire.fr': 'Massif central',
'www.ibps.mediterranee.banquepopulaire.fr': 'Méditerranée',
'www.ibps.nord.banquepopulaire.fr': 'Nord',
'www.ibps.occitane.banquepopulaire.fr': 'Occitane',
'www.ibps.ouest.banquepopulaire.fr': 'Ouest',
'www.ibps.rivesparis.banquepopulaire.fr': 'Rives de Paris',
'www.ibps.sud.banquepopulaire.fr': 'Sud',
'www.ibps.valdefrance.banquepopulaire.fr': 'Val de France',
}.items(), key=lambda k_v: (k_v[1], k_v[0]))])
CONFIG = BackendConfig(Value('website', label=u'Région', choices=website_choices),
CONFIG = BackendConfig(Value('website', label='Région', choices=website_choices),
ValueBackendPassword('login', label='Identifiant', masked=False),
ValueBackendPassword('password', label='Mot de passee'))
ValueBackendPassword('password', label='Mot de passe'))
BROWSER = BanquePopulaire
def create_default_browser(self):
......
......@@ -23,7 +23,7 @@
from requests.exceptions import ConnectionError
from weboob.browser import LoginBrowser, URL, need_login
from weboob.exceptions import BrowserIncorrectPassword
from weboob.exceptions import BrowserIncorrectPassword, ActionNeeded
from weboob.capabilities.bank import Account
from weboob.capabilities.base import NotAvailable
from weboob.tools.decorators import retry
......@@ -111,8 +111,10 @@ def do_login(self):
error_message = self.page.get_error_message()
if error_message:
assert 'Saisie incorrecte' in error_message, error_message
raise BrowserIncorrectPassword(error_message)
if 'Saisie incorrecte' in error_message:
raise BrowserIncorrectPassword(error_message)
elif 'Votre accès est suspendu' in error_message:
raise ActionNeeded(error_message)
# can't login if there is ' ' in the 2 characters asked
if not self.page.login_secret(self.secret):
......
......@@ -306,10 +306,13 @@ def has_no_card(self):
def get_cards(self, account_id):
divs = self.doc.xpath('//div[@class="content-boxed"]')
assert len(divs)
msgs = re.compile(u'Vous avez fait opposition sur cette carte bancaire.' +
'|Votre carte bancaire a été envoyée.' +
'|BforBank a fait opposition sur votre carte' +
'|Pour des raisons de sécurité, la demande de réception du code confidentiel de votre carte par SMS est indisponible')
msgs = re.compile(
'Vous avez fait opposition sur cette carte bancaire.' +
'|Votre carte bancaire a été envoyée.' +
'|Carte bancaire commandée.' +
'|BforBank a fait opposition sur votre carte' +
'|Pour des raisons de sécurité, la demande de réception du code confidentiel de votre carte par SMS est indisponible'
)
divs = [d for d in divs if not msgs.search(CleanText('.//div[has-class("alert")]', default='')(d))]
divs = [d.xpath('.//div[@class="m-card-infos"]')[0] for d in divs]
divs = [d for d in divs if not d.xpath('.//div[@class="m-card-infos-body-text"][text()="Débit immédiat"]')]
......
......@@ -59,6 +59,7 @@ class BinckBrowser(LoginBrowser):
history = URL(r'/TransactionsOverview/GetTransactions',
r'/TransactionsOverview/FilteredOverview', HistoryPage)
questions = URL(r'/FDL_Complex_FR_Compte',
r'/FDL_NonComplex_FR_Compte',
r'FsmaMandatoryQuestionnairesOverview', QuestionPage)
change_pass = URL(r'/ChangePassword/Index',
r'/EditSetting/GetSetting\?code=MutationPassword', ChangePassPage)
......
......@@ -43,7 +43,7 @@ def on_load(self):
if self.doc.xpath(u'//h1[contains(text(), "Questionnaires connaissance et expérience")]'):
form = self.get_form('//form[@action="/FsmaMandatoryQuestionnairesOverview/PostponeQuestionnaires"]')
else:
form = self.get_form('//form[@action="/FDL_Complex_FR_Compte/Introduction/SkipQuestionnaire"]')
form = self.get_form('//form[contains(@action, "Complex_FR_Compte/Introduction/SkipQuestionnaire")]')
form.submit()
......@@ -217,7 +217,7 @@ class item(ItemElement):
obj_unitprice = Env('unitprice', default=NotAvailable)
obj_valuation = MyDecimal(Dict('ValueInEuro'))
obj_diff = MyDecimal(Dict('ResultValueInEuro'))
obj_diff_percent = Eval(lambda x: x / 100, MyDecimal(Dict('ResultPercentageInEuro')))
obj_diff_ratio = Eval(lambda x: x / 100, MyDecimal(Dict('ResultPercentageInEuro')))
obj_original_currency = Env('o_currency', default=NotAvailable)
obj_original_unitvalue = Env('o_unitvalue', default=NotAvailable)
obj_original_unitprice = Env('o_unitprice', default=NotAvailable)
......
......@@ -76,10 +76,14 @@ def do_login(self):
if self.error.is_here() or self.page.is_error():
raise BrowserIncorrectPassword()
if self.type == '2' and self.page.is_corporate():
self.logger.info('Manager corporate connection')
raise SiteSwitch('corporate')
# ti corporate and ge corporate are not detected the same way ..
if 'corporate' in self.page.url:
self.logger.info('Carholder corporate connection')
self.is_corporate = True
else:
self.logger.info('Cardholder connection')
def ti_card_go(self):
if self.is_corporate:
......
......@@ -79,6 +79,14 @@ def iter_history(self, account):
(date.today() - timedelta(days=90)).strftime('%Y%m%d'),
date.today().strftime('%Y%m%d'))
@need_login
def iter_documents(self, subscription):
raise NotImplementedError()
@need_login
def iter_subscription(self):
raise NotImplementedError()
@need_login
def iter_coming_operations(self, account):
return self.get_transactions(account.id,
......
......@@ -33,7 +33,7 @@
from .pages import (
LoginPage, AuthPage, AccountsPage, AccountHistoryViewPage, AccountHistoryPage,
ActionNeededPage, TransactionPage, MarketPage, InvestPage
ActionNeededPage, TransactionPage, MarketPage, InvestPage,
)
......@@ -124,6 +124,14 @@ def iter_history(self, account):
return []
return self._iter_history_base(account)
@need_login
def iter_documents(self, subscription):
raise NotImplementedError()
@need_login
def iter_subscription(self):
raise NotImplementedError()
def _iter_history_base(self, account):
dformat = "%Y%m%d"
......
......@@ -230,6 +230,7 @@ class AccountHistoryPage(LoggedPage, JsonPage):
u'0083': Transaction.TYPE_DEFERRED_CARD,
u'0813': Transaction.TYPE_LOAN_PAYMENT,
u'0568': Transaction.TYPE_TRANSFER,
u'1194': Transaction.TYPE_DEFERRED_CARD, # PAYBACK typed as DEFERRED_CARD
}
@method
......
......@@ -33,6 +33,10 @@
from weboob.capabilities.base import find_object, strict_find_object
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword, Value, ValueBool
from weboob.capabilities.bill import (
Subscription, CapDocument, SubscriptionNotFound, DocumentNotFound, Document,
DocumentTypes,
)
from .enterprise.browser import BNPEnterprise
from .company.browser import BNPCompany
......@@ -42,7 +46,7 @@
__all__ = ['BNPorcModule']
class BNPorcModule(Module, CapBankWealth, CapBankTransferAddRecipient, CapMessages, CapContact, CapProfile):
class BNPorcModule(Module, CapBankWealth, CapBankTransferAddRecipient, CapMessages, CapContact, CapProfile, CapDocument):
NAME = 'bnporc'
MAINTAINER = u'Romain Bignon'
EMAIL = 'romain@weboob.org'
......@@ -61,6 +65,13 @@ class BNPorcModule(Module, CapBankWealth, CapBankTransferAddRecipient, CapMessag
'ent2': 'Entreprises et PME (nouveau site)'}))
STORAGE = {'seen': []}
accepted_document_types = (
DocumentTypes.STATEMENT,
DocumentTypes.REPORT,
DocumentTypes.BILL,
DocumentTypes.OTHER,
)
# Store the messages *list* for this duration
CACHE_THREADS = timedelta(seconds=3 * 60 * 60)
......@@ -74,6 +85,14 @@ def create_default_browser(self):
self.BROWSER = b[self.config['website'].get()]
return self.create_browser(self.config)
def iter_resources(self, objs, split_path):
if Account in objs:
self._restrict_level(split_path)
return self.iter_accounts()
if Subscription in objs:
self._restrict_level(split_path)
return self.iter_subscription()
def iter_accounts(self):
return self.browser.iter_accounts()
......@@ -205,4 +224,27 @@ def set_message_read(self, message):
self.storage.get('seen', default=[]).append(message.thread.id)
self.storage.save()
def get_subscription(self, _id):
return find_object(self.iter_subscription(), id=_id, error=SubscriptionNotFound)
def iter_documents(self, subscription):
if not isinstance(subscription, Subscription):
subscription = self.get_subscription(subscription)
return self.browser.iter_documents(subscription)
def iter_subscription(self):
return self.browser.iter_subscription()
def get_document(self, _id):
subscription_id = _id.split('_')[0]
subscription = self.get_subscription(subscription_id)
return find_object(self.iter_documents(subscription), id=_id, error=DocumentNotFound)
def download_document(self, document):
if not isinstance(document, Document):
document = self.get_document(document)
return self.browser.open(document.url).content
OBJECTS = {Thread: fill_thread}
......@@ -30,13 +30,14 @@
AccountNotFound, Account, AddRecipientStep, AddRecipientTimeout,
TransferInvalidRecipient, Loan,
)
from weboob.capabilities.bill import Subscription
from weboob.capabilities.profile import ProfileMissing
from weboob.tools.decorators import retry
from weboob.tools.capabilities.bank.transactions import sorted_transactions
from weboob.tools.json import json
from weboob.browser.exceptions import ServerError
from weboob.browser.elements import DataError
from weboob.exceptions import BrowserIncorrectPassword
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable
from weboob.tools.value import Value, ValueBool
from weboob.tools.capabilities.bank.investments import create_french_liquidity
......@@ -50,6 +51,7 @@
UselessPage, TransferAssertionError, LoanDetailsPage,
)
from .document_pages import DocumentsPage, TitulairePage
__all__ = ['BNPPartPro', 'HelloBank']
......@@ -92,12 +94,14 @@ class BNPParibasBrowser(JsonBrowserMixin, LoginBrowser):
r'/fr/espace-pro/changer-son-mot-de-passe',
r'/fr/espace-client/100-connexions',
r'/fr/espace-prive/mot-de-passe-expire',
r'/fr/client/mdp-expire',
r'/fr/client/100-connexion',
r'/fr/systeme/page-indisponible', ConnectionThresholdPage)
accounts = URL(r'udc-wspl/rest/getlstcpt', AccountsPage)
loan_details = URL(r'caraccomptes-wspl/rpc/(?P<loan_type>.*)', LoanDetailsPage)
ibans = URL(r'rib-wspl/rpc/comptes', AccountsIBANPage)
history = URL(r'rop2-wspl/rest/releveOp', HistoryPage)
history_old = URL(r'rop-wspl/rest/releveOp', HistoryPage)
transfer_init = URL(r'virement-wspl/rest/initialisationVirement', TransferInitPage)
lifeinsurances = URL(r'mefav-wspl/rest/infosContrat', LifeInsurancesPage)
......@@ -121,6 +125,9 @@ class BNPParibasBrowser(JsonBrowserMixin, LoginBrowser):
advisor = URL(r'/conseiller-wspl/rest/monConseiller', AdvisorPage)
titulaire = URL(r'/demat-wspl/rest/listerTitulairesDemat', TitulairePage)
document = URL(r'/demat-wspl/rest/rechercheCriteresDemat', DocumentsPage)
profile = URL(r'/kyc-wspl/rest/informationsClient', ProfilePage)
list_detail_card = URL(r'/udcarte-wspl/rest/listeDetailCartes', ListDetailCardPage)
......@@ -275,14 +282,21 @@ def iter_history(self, account, coming=False):
if not self.card_to_transaction_type:
self.list_detail_card.go()
self.card_to_transaction_type = self.page.get_card_to_transaction_type()
self.history.go(data=JSON({
data = JSON({
"ibanCrypte": account.id,
"pastOrPending": 1,
"triAV": 0,
"startDate": (datetime.now() - relativedelta(years=1)).strftime('%d%m%Y'),
"endDate": datetime.now().strftime('%d%m%Y')
}))
})
try:
self.history.go(data=data)
except BrowserUnavailable:
# old url is still used for certain connections bu we don't know which one is,
# so the same HistoryPage is attained by the old url in another URL object
data[1]['startDate'] = (datetime.now() - relativedelta(years=3)).strftime('%d%m%Y')
# old url authorizes up to 3 years of history
self.history_old.go(data=data)
if coming:
return sorted_transactions(self.page.iter_coming())
......@@ -510,6 +524,40 @@ def iter_threads(self):
def get_thread(self, thread):
raise NotImplementedError()
@need_login
def iter_documents(self, subscription):
titulaires = self.titulaire.go().get_titulaires()
# Calling '/demat-wspl/rest/listerDocuments' before the request on 'document'
# is necessary when you specify an ikpi, otherwise no documents are returned
self.location('/demat-wspl/rest/listerDocuments')
# When we only have one titulaire, no need to use the ikpi parameter in the request,
# all document are provided with this simple request
data = {
'dateDebut': (datetime.now() - relativedelta(years=3)).strftime('%d/%m/%Y'),
'dateFin': datetime.now().strftime('%d/%m/%Y'),
}
# Ikpi is necessary for multi titulaires accounts to get each document of each titulaires
if len(titulaires) > 1:
data['ikpiPersonne'] = subscription._iduser
self.document.go(json=data)
return self.page.iter_documents(sub_id=subscription.id, sub_number=subscription._number, baseurl=self.BASEURL)
@need_login
def iter_subscription(self):
acc_list = self.iter_accounts()
for acc in acc_list:
sub = Subscription()
sub.label = acc.label
sub.subscriber = acc._subscriber
sub.id = acc.id
# number is the hidden number of an account like "****1234"
# and it's used in the parsing of the docs in iter_documents
sub._number = acc.number
# iduser is the ikpi affiliate to the account,
# usefull for multi titulaires connexions
sub._iduser = acc._iduser
yield sub
class BNPPartPro(BNPParibasBrowser):
BASEURL_TEMPLATE = r'https://%s.bnpparibas/'
......
# -*- coding: utf-8 -*-
# Copyright(C) 2009-2019 Romain Bignon
#
# This file is part of a weboob module.
#
# This weboob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This weboob module 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
import re
from weboob.browser.elements import DictElement, ItemElement, method
from weboob.browser.filters.json import Dict
from weboob.browser.filters.standard import Format, Date, Env
from weboob.browser.pages import JsonPage, LoggedPage
from weboob.capabilities.bill import Document, DocumentTypes
from weboob.tools.compat import urlencode
patterns = {
r'Relevé': DocumentTypes.STATEMENT,
r'Livret(s) A': DocumentTypes.STATEMENT,
r'développement durable': DocumentTypes.STATEMENT,
r'Synthèse': DocumentTypes.STATEMENT,
r'Echelles/Décomptes': DocumentTypes.STATEMENT,
r'épargne logement': DocumentTypes.STATEMENT,
r'Livret(s) jeune': DocumentTypes.STATEMENT,
r'Compte(s) sur Livret': DocumentTypes.STATEMENT,
r'Récapitulatifs annuels': DocumentTypes.REPORT,
r"Avis d'exécution": DocumentTypes.REPORT,
r'Factures': DocumentTypes.BILL,
}
def get_document_type(family):
for patt, type in patterns.items():
if re.search(re.escape(patt), family):
return type
return DocumentTypes.OTHER
class TitulairePage(LoggedPage, JsonPage):
def get_titulaires(self):
return set([t['idKpiTitulaire'] for t in self.doc['data']['listeTitulairesDemat']['listeTitulaires']])
class DocumentsPage(LoggedPage, JsonPage):
@method
class iter_documents(DictElement):
item_xpath = 'data/rechercheCriteresDemat/*/*/listeDocument'
ignore_duplicate = True
class item(ItemElement):
klass = Document
def condition(self):
# There is two type of json, the one with the ibancrypte in it
# and the one with the idcontrat in it, here we check if
# the document belong to the subscritpion.
if 'ibanCrypte' in self.el:
return Env('sub_id')(self) in Dict('ibanCrypte')(self)
else:
return Env('sub_number')(self) in Dict('idContrat')(self)
obj_date = Date(Dict('dateDoc'), dayfirst=True)
obj_format = 'pdf'
obj_id = Format('%s_%s', Env('sub_id'), Dict('idDoc'))
def obj_label(self):
if 'ibanCrypte' in self.el:
return '%s %s N° %s' % (Dict('dateDoc')(self), Dict('libelleSousFamille')(self), Dict('numeroCompteAnonymise')(self))
else:
return '%s %s N° %s' % (Dict('dateDoc')(self), Dict('libelleSousFamille')(self), Dict('idContrat')(self))
def obj_url(self):
keys_to_copy = {
'idDocument' :'idDoc',
'dateDocument': 'dateDoc',
'idLocalisation': 'idLocalisation',
'viDocDocument': 'viDocDocument',
}
# Here we parse the json with ibancrypte in it, for most cases
if 'ibanCrypte' in self.el:
url = 'demat-wspl/rest/consultationDocumentDemat?'
keys_to_copy.update({
'typeCpt': 'typeCompte',
'familleDoc': 'famDoc',
'ibanCrypte': 'ibanCrypte',
'typeDoc': 'typeDoc',
'consulted': 'consulted',
})
request_params = {'typeFamille': 'R001', 'ikpiPersonne': ''}
# Here we parse the json with idcontrat in it. For the cases present
# on privee.mabanque where sometimes the doc url is different
else:
url = 'demat-wspl/rest/consultationDocumentSpecialBpfDemat?'
keys_to_copy.update({
'heureDocument': 'heureDoc',
'numClient': 'numClient',
'typeReport': 'typeReport',
})
request_params = {'ibanCrypte': ''}
for k, v in keys_to_copy.items():
request_params[k] = Dict(v)(self)
return Env('baseurl')(self) + url + urlencode(request_params)
def obj_type(self):
return get_document_type(Dict('libelleSousFamille')(self))
......@@ -30,7 +30,8 @@
from weboob.browser.elements import DictElement, ListElement, TableElement, ItemElement, method
from weboob.browser.filters.json import Dict
from weboob.browser.filters.standard import (
Format, Eval, Regexp, CleanText, Date, CleanDecimal, Field, Coalesce, Map, Env, Currency,
Format, Eval, Regexp, CleanText, Date, CleanDecimal, Field, Coalesce, Map, Env,
Currency,
)
from weboob.browser.filters.html import TableCell
from weboob.browser.pages import JsonPage, LoggedPage, HTMLPage
......@@ -107,7 +108,10 @@ def looks_legit(self, password):
return True
def on_load(self):
msg = CleanText('//div[@class="confirmation"]//span[span]')(self.doc)
msg = (
CleanText('//div[@class="confirmation"]//span[span]')(self.doc) or
CleanText('//p[contains(text(), "Vous avez atteint la date de fin de vie de votre code secret")]')(self.doc)
)
self.logger.warning('Password expired.')
if not self.browser.rotating_password:
......@@ -306,7 +310,6 @@ def obj_company_siren(self):
class AccountsPage(BNPPage):
@method
class iter_accounts(DictElement):
item_xpath = 'data/infoUdc/familleCompte'
......@@ -315,6 +318,12 @@ class iter_accounts_details(DictElement):
item_xpath = 'compte'
class item(ItemElement):
def validate(self, obj):
# We skip loans with a balance of 0 because the JSON returned gives
# us no info (only `null` values on all fields), so there is nothing
# useful to display
return obj.type != Account.TYPE_LOAN or obj.balance != 0
FAMILY_TO_TYPE = {
1: Account.TYPE_CHECKING,
2: Account.TYPE_SAVINGS,
......@@ -354,6 +363,8 @@ class item(ItemElement):
obj_balance = Dict('soldeDispo')
obj_coming = Dict('soldeAVenir')
obj_number = Dict('value')
obj__subscriber = Format('%s %s', Dict('titulaire/nom'), Dict('titulaire/prenom'))
obj__iduser = Dict('titulaire/ikpi')
def obj_iban(self):
iban = Map(Dict('key'), Env('ibans')(self), default=NotAvailable)(self)
......@@ -381,11 +392,15 @@ class fill_loan_details(ItemElement):
obj_rate = Dict('data/tauxRemboursement')
obj_nb_payments_left = Dict('data/nbRemboursementRestant')
obj_next_payment_date = Date(Dict('data/dateProchainAmortissement'), dayfirst=True)
obj__subscriber = Format('%s %s', Dict('data/titulaire/nom'), Dict('data/titulaire/prenom'))
obj__iduser = None
@method
class fill_revolving_details(ItemElement):
obj_total_amount = Dict('data/montantDisponible')
obj_rate = Dict('data/tauxInterets')
obj__subscriber = Format('%s %s', Dict('data/titulaire/nom'), Dict('data/titulaire/prenom'))
obj__iduser = None
class AccountsIBANPage(BNPPage):
......@@ -441,6 +456,8 @@ class RecipientsPage(BNPPage):
@method
class iter_recipients(DictElement):
item_xpath = 'data/infoBeneficiaire/listeBeneficiaire'
# We ignore duplicate because BNP allows differents recipients with the same iban
ignore_duplicate = True
class item(MyRecipient):
# For the moment, only yield ready to transfer on recipients.
......@@ -745,6 +762,8 @@ class item(ItemElement):
obj_balance = CleanDecimal(TableCell('balance'), replace_dots=True)
obj_coming = None
obj_iban = None
obj__subscriber = None
obj__iduser = None
def obj_type(self):
for k, v in self.page.ACCOUNT_TYPES.items():
......
......@@ -54,6 +54,7 @@ class Transaction(FrenchTransaction):
(re.compile(r'^(?P<category>FRAIS POUR)(?P<text>.*)'), FrenchTransaction.TYPE_BANK),
(re.compile(r'^(?P<text>(?P<category>REMUNERATION).*)'), FrenchTransaction.TYPE_BANK),
(re.compile(r'^(?P<category>REMISE DE CHEQUES?) (?P<text>.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile(r'^(?P<category>VERSEMENT DAB) (?P<text>.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile(r'^(?P<text>DEBIT CARTE BANCAIRE DIFFERE.*)'), FrenchTransaction.TYPE_CARD_SUMMARY),
(re.compile(r'^(?P<category>COTISATION TRIMESTRIELLE).*'), FrenchTransaction.TYPE_BANK),
(re.compile(r'^REMISE COMMERCIALE.*'), FrenchTransaction.TYPE_BANK),
......
......@@ -136,6 +136,7 @@ def obj_type(self):
'livrets?': Account.TYPE_SAVINGS,
'epargnes? logement': Account.TYPE_SAVINGS,
"autres produits d'epargne": Account.TYPE_SAVINGS,
'compte relais': Account.TYPE_SAVINGS,
'comptes? titres? et pea': Account.TYPE_MARKET,
'compte-titres': Account.TYPE_MARKET,
'assurances? vie': Account.TYPE_LIFE_INSURANCE,
......
......@@ -52,6 +52,7 @@
SmsPage, SmsPageOption, SmsRequest, AuthentPage, RecipientPage, CanceledAuth, CaissedepargneKeyboard,
TransactionsDetailsPage, LoadingPage, ConsLoanPage, MeasurePage, NatixisLIHis, NatixisLIInv, NatixisRedirectPage,
SubscriptionPage, CreditCooperatifMarketPage, UnavailablePage, CardsPage, CardsComingPage, CardsOldWebsitePage, TransactionPopupPage,
OldLeviesPage, NewLeviesPage,
)
from .linebourse_browser import LinebourseAPIBrowser
......@@ -86,6 +87,8 @@ class CaisseEpargne(LoginBrowser, StatesMixin):
cards_old = URL('https://.*/Portail.aspx.*', CardsOldWebsitePage)
cards = URL('https://.*/Portail.aspx.*', CardsPage)
cards_coming = URL('https://.*/Portail.aspx.*', CardsComingPage)
old_checkings_levies = URL(r'https://.*/Portail.aspx.*', OldLeviesPage)
new_checkings_levies = URL(r'https://.*/Portail.aspx.*', NewLeviesPage)
authent = URL('https://.*/Portail.aspx.*', AuthentPage)
subscription = URL('https://.*/Portail.aspx\?tache=(?P<tache>).*', SubscriptionPage)
transaction_popup = URL(r'https://.*/Portail.aspx.*', TransactionPopupPage)
......@@ -458,7 +461,7 @@ def get_accounts_list(self):
- In CardsPage, there are cards (with "Business" in the label) without checking account on the
website (neither history nor coming), so we skip them.
- Some card on the CardsPage that have a checking account parent, but if we follow the link to
reach it with CardsComingPage, we find an other card that not in CardsPage.
reach it with CardsComingPage, we find an other card that is not in CardsPage.
"""
if self.new_website:
for account in self.accounts:
......@@ -736,33 +739,51 @@ def match_cb(tr):
@need_login
def get_coming(self, account):
if account.type != account.TYPE_CARD:
return []
if account.type == account.TYPE_CHECKING:
return self.get_coming_checking(account)
elif account.type == account.TYPE_CARD:
return self.get_coming_card(account)
return []
def get_coming_checking(self, account):
# The accounts list or account history page does not contain comings for checking accounts
# We need to go to a specific levies page where we can find past and coming levies (such as recurring ones)
trs = []
self.home.go()
self.page.go_cards() # need to go to cards page to have access to the nav bar where we can choose LeviesPage from
if not self.page.levies_page_enabled():
return trs
self.page.go_levies() # need to go to a general page where we find levies for all accounts before requesting a specific account
if not self.page.comings_enabled(account.id):
return trs
self.page.go_levies(account.id)
if self.new_checkings_levies.is_here() or self.old_checkings_levies.is_here():
today = datetime.datetime.today().date()
# Today transactions are in this page but also in history page, we need to ignore it as a coming
for tr in self.page.iter_coming():
if tr.date > today:
trs.append(tr)
return trs
def get_coming_card(self, account):
trs = []
if not hasattr(account.parent, '_info'):
raise NotImplementedError()
# 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 is empty, that means there are no comings yet
if info:
for tr in self._get_history(info.copy(), account):
tr.type = tr.TYPE_DEFERRED_CARD
trs.append(tr)
return sorted_transactions(trs)
@need_login
......
......@@ -524,12 +524,12 @@ class item(ItemElement):
obj_label = Env('label')
obj_type = Loan.TYPE_LOAN
obj_total_amount = MyDecimal(MyTableCell("total_amount"))
obj_rate = Eval(lambda x: x / 100, MyDecimal(MyTableCell("rate", default=NotAvailable), default=NotAvailable))
obj_balance = MyDecimal(MyTableCell("balance"), sign=lambda x: -1)
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", default=''), default=NotAvailable), default=NotAvailable)
obj_rate = MyDecimal(MyTableCell("rate", default=NotAvailable), default=NotAvailable)
def submit_form(self, form, eventargument, eventtarget, scriptmanager):
form['__EVENTARGUMENT'] = eventargument
......@@ -538,6 +538,36 @@ def submit_form(self, form, eventargument, eventtarget, scriptmanager):
fix_form(form)
form.submit()
def go_levies(self, account_id=None):
form = self.get_form(id='main')
if account_id:
# Go to an account specific levies page
eventargument = ""
if "MM$m_CH$IsMsgInit" in form:
# Old website
form['MM$SYNTHESE_SDD_RECUS$m_ExDropDownList'] = account_id
eventtarget = "MM$SYNTHESE_SDD_RECUS$m_ExDropDownList"
scriptmanager = "MM$m_UpdatePanel|MM$SYNTHESE_SDD_RECUS$m_ExDropDownList"
else:
# New website
form['MM$SYNTHESE_SDD_RECUS$ddlCompte'] = account_id
eventtarget = "MM$SYNTHESE_SDD_RECUS$ddlCompte"
scriptmanager = "MM$m_UpdatePanel|MM$SYNTHESE_SDD_RECUS$ddlCompte"
self.submit_form(form, eventargument, eventtarget, scriptmanager,)
else:
# Go to an general levies page page where all levies are found
if "MM$m_CH$IsMsgInit" in form:
# Old website
eventargument = "SDDRSYN0"
eventtarget = "Menu_AJAX"
scriptmanager = "m_ScriptManager|Menu_AJAX"
else:
# New website
eventargument = "SDDRSYN0&codeMenu=WPS1"
eventtarget = "MM$Menu_Ajax"
scriptmanager = "MM$m_UpdatePanel|MM$Menu_Ajax"
self.submit_form(form, eventargument, eventtarget, scriptmanager,)
def go_list(self):
form = self.get_form(id='main')
......@@ -842,6 +872,13 @@ def go_pro_transfer_availability(self):
def is_transfer_allowed(self):
return not self.doc.xpath('//ul/li[contains(text(), "Aucun compte tiers n\'est disponible")]')
def levies_page_enabled(self):
""" Levies page does not exist in the nav bar for every connections """
return (
CleanText('//a/span[contains(text(), "Suivre mes prélèvements reçus")]')(self.doc) or # new website
CleanText('//a[contains(text(), "Suivre les prélèvements reçus")]')(self.doc) # old website
)
class TransactionPopupPage(LoggedPage, HTMLPage):
def is_here(self):
......@@ -851,6 +888,74 @@ def complete_label(self):
return CleanText('''//div[@class="scrollPane"]/table[//caption[contains(text(), "Détail de l'opération")]]//tr[2]''')(self.doc)
class NewLeviesPage(IndexPage):
""" Scrape new website 'Prélèvements' page for comings for checking accounts """
def is_here(self):
return CleanText('//h2[contains(text(), "Suivez vos prélèvements reçus")]')(self.doc)
def comings_enabled(self, account_id):
""" Check if a specific account can be selected on the general levies page """
return account_id in CleanText('//span[@id="MM_SYNTHESE_SDD_RECUS"]//select/option/@value')(self.doc)
@method
class iter_coming(TableElement):
head_xpath = '//div[contains(@id, "ListePrelevement_0")]/table[contains(@summary, "Liste des prélèvements en attente")]//tr/th'
item_xpath = '//div[contains(@id, "ListePrelevement_0")]/table[contains(@summary, "Liste des prélèvements en attente")]//tr[contains(@id, "trRowDetail")]'
col_label = 'Libellé/Référence'
col_coming = 'Montant'
col_date = 'Date'
class item(ItemElement):
klass = Transaction
# Transaction typing will mostly not work since transaction as comings will only display the debiting organism in the label
# Labels will bear recognizable patterns only when they move from future to past, where they will be typed by iter_history
# when transactions change state from coming to history 'Prlv' is append to their label, this will help the backend for the matching
obj_raw = Transaction.Raw(Format('Prlv %s', Field('label')))
obj_label = CleanText(TableCell('label'))
obj_amount = CleanDecimal.French(TableCell('coming'), sign=lambda x: -1)
obj_date = Date(CleanText(TableCell('date')), dayfirst=True)
def condition(self):
return not CleanText('''//p[contains(text(), "Vous n'avez pas de prélèvement en attente d'exécution.")]''')(self)
class OldLeviesPage(IndexPage):
""" Scrape old website 'Prélèvements' page for comings for checking accounts """
def is_here(self):
return CleanText('//span[contains(text(), "Suivez vos prélèvements reçus")]')(self.doc)
def comings_enabled(self, account_id):
""" Check if a specific account can be selected on the general levies page """
return account_id in CleanText('//span[@id="MM_SYNTHESE_SDD_RECUS"]//select/option/@value')(self.doc)
@method
class iter_coming(TableElement):
head_xpath = '''//span[contains(text(), "Prélèvements en attente d'exécution")]/ancestor::table[1]/following-sibling::table[1]//tr[contains(@class, "DataGridHeader")]//td'''
item_xpath = '''//span[contains(text(), "Prélèvements en attente d'exécution")]/ancestor::table[1]/following-sibling::table[1]//tr[contains(@class, "DataGridHeader")]//following-sibling::tr'''
col_label = 'Libellé/Référence'
col_coming = 'Montant'
col_date = 'Date'
class item(ItemElement):
klass = Transaction
# Transaction typing will mostly not work since transaction as comings will only display the debiting organism in the label
# Labels will bear recognizable patterns only when they move from future to past, where they will be typed by iter_history
# when transactions change state from coming to history 'Prlv' is append to their label, this will help the backend for the matching
obj_raw = Transaction.Raw(Format('Prlv %s', Field('label')))
obj_label = CleanText(TableCell('label'))
obj_amount = CleanDecimal.French(TableCell('coming'), sign=lambda x: -1)
obj_date = Date(CleanText(TableCell('date')), dayfirst=True)
def condition(self):
return not CleanText('''//table[@id="MM_SYNTHESE_SDD_RECUS_rpt_dgList_0"]//td[contains(text(), "Vous n'avez pas de prélèvements")]''')(self)
class CardsPage(IndexPage):
def is_here(self):
return CleanText('//h3[normalize-space(text())="Mes cartes (cartes dont je suis le titulaire)"]')(self.doc)
......@@ -1119,6 +1224,10 @@ def find_elements(self):
class item(ItemElement):
klass = Transaction
def condition(self):
# Eliminate transactions without amount
return Dict('montantBrut')(self)
obj_raw = Transaction.Raw(Dict('type/libelleLong'))
obj_amount = Eval(float_to_decimal, Dict('montantBrut/valeur'))
......@@ -1169,7 +1278,7 @@ def obj_diff(self):
return Eval(float_to_decimal, Dict('montantPlusValue/valeur'))(self)
return NotAvailable
def obj_diff_percent(self):
def obj_diff_ratio(self):
if Dict('tauxPlusValue')(self):
return Eval(lambda x: float_to_decimal(x) / 100, Dict('tauxPlusValue'))(self)
return NotAvailable
......
......@@ -613,17 +613,17 @@ def parse(self, obj):
self.env['vdate'] = NotAvailable
if CleanText('//table[@class="ca-table"][caption[span[b[text()="Historique des opérations"]]]]//tr[count(td) = 4]')(self):
# History table with 4 columns
self.env['raw'] = CleanText('./td[2]', children=False)(self)
self.env['raw'] = CleanText('./td[2]')(self)
self.env['amount'] = CleanDecimal.French('./td[last()]')(self)
elif CleanText('//table[@class="ca-table"][caption[span[b[text()="Historique des opérations"]]]]//tr[count(td) = 5]')(self):
# History table with 5 columns
self.env['raw'] = CleanText('./td[3]', children=False)(self)
self.env['raw'] = CleanText('./td[3]')(self)
self.env['amount'] = CleanDecimal.French('./td[last()]')(self)
elif CleanText('//table[@class="ca-table"][caption[span[b[text()="Historique des opérations"]]]]//tr[count(td) = 6]')(self):
# History table with 6 columns (contains vdate)
self.env['raw'] = CleanText('./td[4]', children=False)(self)
self.env['raw'] = CleanText('./td[4]')(self)
self.env['vdate'] = DateGuesser(CleanText('./td[2]'), Env('date_guesser'))(self)
self.env['amount'] = CleanDecimal.French('./td[last()]')(self)
......@@ -635,11 +635,11 @@ def parse(self, obj):
)(self)
if CleanText('//table[@class="ca-table"][caption[span[b[text()="Historique des opérations"]]]]//th[a[contains(text(), "Valeur")]]')(self):
# With vdate column ('Valeur')
self.env['raw'] = CleanText('./td[4]', children=False)(self)
self.env['raw'] = CleanText('./td[4]')(self)
self.env['vdate'] = DateGuesser(CleanText('./td[2]'), Env('date_guesser'))(self)
else:
# Without any vdate column
self.env['raw'] = CleanText('./td[3]', children=False)(self)
self.env['raw'] = CleanText('./td[3]')(self)
else:
assert False, 'This type of history table is not handled yet!'
......@@ -833,7 +833,7 @@ class item(ItemElement):
CleanDecimal.French('.//span[@class="box"][span[span[text()="Répartition"]]]/span[2]/span')
)
def obj_diff_percent(self):
def obj_diff_ratio(self):
# Euro funds have '-' instead of a diff_percent value
if CleanText('.//span[@class="box"][span[span[text()="+/- value latente (%)"]]]/span[2]/span')(self) == '-':
return NotAvailable
......