pax_global_header 0000666 0000000 0000000 00000000064 14575653726 0014536 g ustar 00root root 0000000 0000000 52 comment=5f3d558793b537a74480241ac6981479f5938cd3
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/ 0000775 0000000 0000000 00000000000 14575653726 0023670 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/ 0000775 0000000 0000000 00000000000 14575653726 0025340 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/bred/ 0000775 0000000 0000000 00000000000 14575653726 0026254 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/bred/bred/ 0000775 0000000 0000000 00000000000 14575653726 0027170 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/bred/bred/__init__.py0000664 0000000 0000000 00000000074 14575653726 0031302 0 ustar 00root root 0000000 0000000 from .browser import BredBrowser
__all__ = ['BredBrowser']
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/bred/bred/browser.py 0000664 0000000 0000000 00000100105 14575653726 0031222 0 ustar 00root root 0000000 0000000 # Copyright(C) 2014 Romain Bignon
#
# This file is part of a woob module.
#
# This woob 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 woob 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 woob module. If not, see .
import json
import time
import operator
import random
from datetime import date
from decimal import Decimal
from woob.exceptions import (
AuthMethodNotImplemented, AppValidation,
AppValidationExpired, AppValidationCancelled,
BrowserQuestion, BrowserIncorrectPassword,
BrowserUnavailable, ActionNeeded,
)
from woob.capabilities.bank import (
Account, AddRecipientStep, AddRecipientBankError,
TransferBankError, AccountOwnerType,
)
from woob.browser import need_login, URL
from woob.browser.mfa import TwoFactorBrowser
from woob.browser.exceptions import ClientError
from woob.capabilities import NotAvailable
from woob.capabilities.base import find_object
from woob.tools.capabilities.bank.investments import create_french_liquidity
from woob.tools.value import Value
from woob_modules.linebourse.browser import LinebourseAPIBrowser
from .pages import (
HomePage, LoginPage, AccountsTwoFAPage, InitAuthentPage, AuthentResultPage,
SendSmsPage, CheckOtpPage, TrustedDevicesPage, UniversePage,
TokenPage, MoveUniversePage, SwitchPage,
LoansPage, AccountsPage, IbanPage, LifeInsurancesPage,
SearchPage, ProfilePage, ErrorPage, ErrorCodePage, LinebourseLoginPage,
UnavailablePage, ErrorMsgPage,
)
from .transfer_pages import (
RecipientListPage, EmittersListPage, ListAuthentPage,
AddRecipientPage, TransferPage,
)
__all__ = ['BredBrowser']
class BredBrowser(TwoFactorBrowser):
BASEURL = 'https://www.bred.fr'
HAS_CREDENTIALS_ONLY = True
LINEBOURSE_BROWSER = LinebourseAPIBrowser
home = URL(r'/$', HomePage)
login = URL(r'/transactionnel/Authentication', LoginPage)
error = URL(r'.*gestion-des-erreurs/erreur-pwd',
r'.*gestion-des-erreurs/opposition',
r'/pages-gestion-des-erreurs/erreur-technique',
r'/pages-gestion-des-erreurs/message-tiers-oppose', ErrorPage)
universe = URL(r'/transactionnel/services/applications/menu/getMenuUnivers', UniversePage)
token = URL(r'/transactionnel/services/rest/User/nonce\?random=(?P.*)', TokenPage)
move_universe = URL(r'/transactionnel/services/applications/listes/(?P.*)/default', MoveUniversePage)
switch = URL(r'/transactionnel/services/rest/User/switch', SwitchPage)
loans = URL(r'/transactionnel/services/applications/prets/liste', LoansPage)
accounts = URL(r'/transactionnel/services/rest/Account/accounts', AccountsPage)
iban = URL(r'/transactionnel/services/rest/Account/account/(?P.*)/iban', IbanPage)
linebourse_login = URL(r'/transactionnel/v2/services/applications/SSO/linebourse', LinebourseLoginPage)
life_insurances = URL(r'/transactionnel/services/applications/avoirsPrepar/getAvoirs', LifeInsurancesPage)
search = URL(r'/transactionnel/services/applications/operations/getSearch/', SearchPage)
profile = URL(r'/transactionnel/services/rest/User/user', ProfilePage)
error_code = URL(r'/.*\?errorCode=.*', ErrorCodePage)
error_msg_page = URL ( r'/authentification\?source=no&errorCode=(?P\d)', ErrorMsgPage)
unavailable_page = URL(r'/ERREUR/', UnavailablePage)
accounts_twofa = URL(r'/transactionnel/v2/services/rest/Account/accounts', AccountsTwoFAPage)
list_authent = URL(r'/transactionnel/services/applications/authenticationstrong/listeAuthent/(?P\w+)', ListAuthentPage)
init_authent = URL(r'/transactionnel/services/applications/authenticationstrong/init', InitAuthentPage)
authent_result = URL(r'/transactionnel/services/applications/authenticationstrong/result/(?P[^/]+)/(?P\w+)', AuthentResultPage)
trusted_devices = URL(r'/transactionnel/services/applications/trustedDevices', TrustedDevicesPage)
check_otp = URL(r'/transactionnel/services/applications/authenticationstrong/(?P\w+)/check', CheckOtpPage)
send_sms = URL(r'/transactionnel/services/applications/authenticationstrong/sms/send', SendSmsPage)
recipient_list = URL(r'/transactionnel/v2/services/applications/virement/getComptesCrediteurs', RecipientListPage)
emitters_list = URL(r'/transactionnel/v2/services/applications/virement/getComptesDebiteurs', EmittersListPage)
add_recipient = URL(r'/transactionnel/v2/services/applications/beneficiaires/updateBeneficiaire', AddRecipientPage)
create_transfer = URL(r'/transactionnel/v2/services/applications/virement/confirmVirement', TransferPage)
validate_transfer = URL(r'/transactionnel/v2/services/applications/virement/validVirement', TransferPage)
__states__ = (
'auth_method', 'need_reload_state', 'authent_id', 'device_id',
'context', 'recipient_transfer_limit',
)
def __init__(self, accnum, config, *args, **kwargs):
self.config = config
kwargs['username'] = self.config['login'].get()
# Bred only use first 8 char (even if the password is set to be bigger)
# The js login form remove after 8th char. No comment.
kwargs['password'] = self.config['password'].get()[:8]
super(BredBrowser, self).__init__(config, *args, **kwargs)
self.accnum = accnum
self.universes = None
self.current_univers = None
self.need_reload_state = None
self.context = None
self.device_id = None
self.auth_method = None
self.authent_id = None
self.recipient_transfer_limit = None
# Some accounts are detailed on linebourse. The only way to know which is to go on linebourse.
# The parameters to do so depend on the universe.
self.linebourse_urls = {}
self.linebourse_tokens = {}
dirname = self.responses_dirname
if dirname:
dirname += '/bourse'
self.linebourse = self.LINEBOURSE_BROWSER(
'https://www.linebourse.fr',
logger=self.logger,
responses_dirname=dirname,
proxy=self.PROXIES,
)
self.AUTHENTICATION_METHODS = {
'resume': self.handle_polling, # validation in mobile app
'otp_sms': self.handle_otp_sms, # OTP in SMS
'otp_app': self.handle_otp_app, # OTP in mobile app
}
def load_state(self, state):
# Is the browser being reloaded for 2FA?
is_2fa = state.get('resume') or state.get('otp_sms') or state.get('otp_app')
if state.get('need_reload_state') or state.get('device_id') or is_2fa:
state.pop('url', None)
super(BredBrowser, self).load_state(state)
self.need_reload_state = None
def init_login(self):
if self.device_id:
# will not raise 2FA if the one realized with this id is still valid
self.session.headers['x-trusted-device-id'] = self.device_id
if 'hsess' not in self.session.cookies:
self.home.go() # set session token
assert 'hsess' in self.session.cookies, "Session token not correctly set"
# hard-coded authentication payload
data = {
'identifiant': self.username,
'password': self.password,
}
self.login.go(data=data)
if self.error_code.is_here():
code = self.page.get_code()
self.error_msg_page.go(errorcode=code)
msg = self.page.get_msg()
# 20100: invalid login/password
if code == '20100':
raise BrowserIncorrectPassword(msg)
elif code == '20109':
raise ActionNeeded(msg)
# 20104 & 1000: unknown error during login
elif code in ('20104', '1000'):
raise BrowserUnavailable(msg)
raise AssertionError('Error %s is not handled yet.' % code)
try:
# It's an accounts page if SCA already done
# Need to first go there to trigger it, since LoginPage doesn't do that.
self.accounts_twofa.go()
except ClientError as e:
if e.response.status_code == 449:
self.check_interactive()
self.context = e.response.json()['content']
self.trigger_connection_twofa()
raise
def trigger_connection_twofa(self):
# Needed to record the device doing the SCA and keep it valid.
self.device_id = ''.join([str(random.randint(0, 9)) for _ in range(50)]) # Python2 compatible
# self.device_id = ''.join(random.choices(string.digits, k=50)) # better but needs Python3
self.auth_method = self.get_connection_twofa_method()
if self.auth_method == 'notification':
self.update_headers()
data = {
'context': self.context['contextAppli'], # 'accounts_access'
'type_auth': 'NOTIFICATION',
'type_phone': 'P',
}
self.init_authent.go(json=data)
self.authent_id = self.page.get_authent_id()
if self.context.get('message'):
raise AppValidation(self.context['message'])
raise AppValidation("Veuillez valider l'accès sur votre application.")
elif self.auth_method == 'sms':
self.update_headers()
data = {
'context': self.context['context'],
'contextAppli': self.context['contextAppli'],
}
self.send_sms.go(json=data)
if self.context.get('message'):
raise BrowserQuestion(
Value('otp_sms', label=self.context['message']),
)
raise BrowserQuestion(
Value('otp_sms', label="Veuillez entrer le code reçu au %s" % self.context['liste']['numTel'])
)
elif self.auth_method == 'otp':
if self.context.get('message'):
raise BrowserQuestion(
Value('otp_app', label=self.context['message'])
)
raise BrowserQuestion(
Value('otp_app', label="Veuillez entrer le code reçu sur votre application."),
)
def get_connection_twofa_method(self):
# The order and tests are taken from the bred website code.
# Keywords in scripts.js: showSMS showEasyOTP showOTP
methods = self.context['liste']
# Overriding default order of tests with 'preferred_sca' configuration item
preferred_sca_value = self.config.get('preferred_sca')
# Some children don't have this mechanism
if preferred_sca_value:
preferred_auth_methods = tuple(preferred_sca_value.get().split())
for auth_method in preferred_auth_methods:
if methods.get(auth_method):
return auth_method
if methods.get('sms'):
return 'sms'
elif methods.get('notification') and methods.get('otp'):
return 'notification'
elif methods.get('otp'):
return 'otp'
message = self.context['message']
raise AuthMethodNotImplemented('Unhandled strong authentification method: %s' % message)
def update_headers(self):
timestamp = int(time.time() * 1000)
if self.device_id:
self.session.headers['x-trusted-device-id'] = self.device_id
self.token.go(timestamp=timestamp)
self.session.headers['x-token-bred'] = self.page.get_content()
def handle_polling(self, enrol=True):
for _ in range(60): # 5' timeout duration on website
self.update_headers()
self.authent_result.go(
authent_id=self.authent_id,
context=self.context['contextAppli'],
json={}, # yes, needed
)
status = self.page.get_status()
if not status:
# When the notification expires, we get a generic error message
# instead of a status like 'PENDING'
self.context = None
raise AppValidationExpired('La validation par application mobile a expiré.')
elif status == 'ABORTED':
self.context = None
raise AppValidationCancelled("La validation par application a été annulée par l'utilisateur.")
elif status == 'AUTHORISED':
self.context = None
if enrol:
self.enrol_device()
return
assert status == 'PENDING', "Unhandled app validation status : '%s'" % status
time.sleep(5)
self.context = None
raise AppValidationExpired('La validation par application mobile a expiré.')
def handle_otp_sms(self):
self.validate_connection_otp_auth(self.otp_sms)
def handle_otp_app(self):
self.validate_connection_otp_auth(self.otp_app)
def validate_connection_otp_auth(self, auth_value):
self.update_headers()
data = {
'context': self.context['context'],
'contextAppli': self.context['contextAppli'],
'otp': auth_value,
}
self.check_otp.go(
auth_method=self.auth_method,
json=data,
)
error = self.page.get_error()
if error:
raise BrowserIncorrectPassword('Error when validating OTP: %s' % error)
self.context = None
self.enrol_device()
def enrol_device(self):
# Add device_id to list of trusted devices to avoid SCA for 90 days
# User will see a 'BI' entry on this list and can delete it on demand.
self.update_headers()
device_name = 'Accès Budget-Insight pour agrégation'
device_name_value = self.config.get('device_name')
if device_name_value:
device_name = device_name_value.get()
data = {
'uuid': self.device_id, # Called an uuid but it's just a 50 digits long string.
'deviceName': device_name, # clear message for user
'biometricEnabled': False,
'securedBiometricEnabled': False,
'notificationEnabled': False,
}
self.trusted_devices.go(json=data)
error = self.page.get_error()
if error:
raise BrowserIncorrectPassword('Error when enroling trusted device: %s' % error)
@need_login
def get_universes(self):
"""Get universes (particulier, pro, etc)"""
self.update_headers()
self.universe.go(headers={'Accept': 'application/json'})
return self.page.get_universes()
def move_to_universe(self, univers):
if univers == self.current_univers:
return
self.move_universe.go(key=univers)
self.update_headers()
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
self.switch.go(
data=json.dumps({'all': 'false', 'univers': univers}),
headers=headers,
)
self.current_univers = univers
def linebourse_login_for_universe(self, universe):
self.linebourse_login.go()
if self.linebourse_login.is_here():
linebourse_url = self.page.get_linebourse_url()
if linebourse_url:
self.linebourse_urls[universe] = linebourse_url
self.linebourse_tokens[universe] = self.page.get_linebourse_token()
def universe_to_owner_type(self, universe_key):
# P particulire
# M immobilier
# E entreprise
# R pro
# S pro+
if universe_key in ['P', 'M']:
return AccountOwnerType.PRIVATE
elif universe_key in ['E', 'R', 'S']:
return AccountOwnerType.ORGANIZATION
else:
self.logger.error("New universe to convert to owner_type: %s", universe_key)
return NotAvailable
@need_login
def get_accounts_list(self):
accounts = []
for universe_key in sorted(self.get_universes()):
self.move_to_universe(universe_key)
universe_accounts = []
owner_type = self.universe_to_owner_type(universe_key)
universe_accounts.extend(self.get_list())
universe_accounts.extend(self.get_life_insurance_list())
universe_accounts.extend(self.get_loans_list())
linebourse_accounts = self.get_linebourse_accounts(universe_key)
for account in universe_accounts:
account._is_in_linebourse = False
account.owner_type = owner_type
# Accound id looks like 'bred_account_id.folder_id'
# We only want bred_account_id and we need to clean it to match it to linebourse IDs.
account_id = account.id.strip('0').split('.')[0]
for linebourse_account in linebourse_accounts:
if account_id in linebourse_account:
account._is_in_linebourse = True
accounts.extend(universe_accounts)
# Life insurances are sometimes in multiple universes, we have to remove duplicates
unique_accounts = {account.id: account for account in accounts}.values()
# Fill parents with resulting accounts when relevant:
for account in unique_accounts:
if account.type not in [Account.TYPE_CARD, Account.TYPE_LIFE_INSURANCE]:
continue
account.parent = find_object(
unique_accounts, _number=account._parent_number, type=Account.TYPE_CHECKING
)
return sorted(unique_accounts, key=operator.attrgetter('_univers'))
@need_login
def get_linebourse_accounts(self, universe_key):
self.move_to_universe(universe_key)
if universe_key not in self.linebourse_urls:
self.linebourse_login_for_universe(universe_key)
if universe_key in self.linebourse_urls:
self.linebourse.location(
self.linebourse_urls[universe_key],
data={'SJRToken': self.linebourse_tokens[universe_key]}
)
self.linebourse.session.headers['X-XSRF-TOKEN'] = self.linebourse.session.cookies.get('XSRF-TOKEN')
params = {'_': '{}'.format(int(time.time() * 1000))}
try:
self.linebourse.go_account_codes(params=params)
except self.LINEBOURSE_BROWSER.LinebourseNoSpace:
return []
if self.linebourse.account_codes.is_here():
return self.linebourse.page.get_accounts_list()
return []
@need_login
def get_loans_list(self):
self.loans.go()
return self.page.iter_loans(current_univers=self.current_univers)
@need_login
def get_list(self):
self.accounts.go()
return self.page.iter_accounts(accnum=self.accnum, current_univers=self.current_univers)
@need_login
def get_life_insurance_list(self):
self.life_insurances.go()
self.page.check_error()
return self.page.iter_lifeinsurances(univers=self.current_univers)
@need_login
def _make_api_call(self, account, start_date, end_date, offset, max_length=50):
self.update_headers()
call_payload = {
"account": account._number,
"poste": account._nature,
"sousPoste": account._codeSousPoste or '00',
"devise": account.currency,
"fromDate": start_date.strftime('%Y-%m-%d'),
"toDate": end_date.strftime('%Y-%m-%d'),
"from": offset,
"size": max_length, # max length of transactions
"search": "",
"categorie": "",
}
result = self.search.go(json=call_payload)
self.page.check_error()
return result
@need_login
def iter_history(self, account, coming=False):
if account.type in (Account.TYPE_LOAN, Account.TYPE_LIFE_INSURANCE) or not account._consultable:
raise NotImplementedError()
self.move_to_universe(account._univers)
today = date.today()
end_date = date.today()
start_date = date(day=1, month=1, year=2000)
max_length = 50
successive_hist_transaction_counter = 0
self._make_api_call(
account=account, start_date=start_date,
end_date=end_date, offset=0, max_length=max_length,
)
max_transactions = self.page.get_max_transactions()
for transaction in self.page.iter_history(
account=account, account_type=account.type,
today=today, start_date=start_date, end_date=end_date,
max_length=max_length, max_transactions=max_transactions,
):
if coming == transaction._coming:
successive_hist_transaction_counter = 0
yield transaction
elif coming and not transaction._coming:
# coming transactions are at the top of history, but we make sure that there
# are no transactions in coming after few transactions that are not in coming.
# If we encounter more than 10 successive history transactions, we stop the iteration.
successive_hist_transaction_counter += 1
if successive_hist_transaction_counter > 10:
self.logger.debug('stopping coming after %s', transaction)
return
@need_login
def iter_investments(self, account):
if account.type == Account.TYPE_LIFE_INSURANCE:
for invest in account._investments:
yield invest
elif account.type in (Account.TYPE_PEA, Account.TYPE_MARKET):
if 'Portefeuille Titres' in account.label:
if account._is_in_linebourse:
self.move_to_universe(account._univers)
self.linebourse_login_for_universe(account._univers)
self.linebourse.location(
self.linebourse_urls[account._univers],
data={'SJRToken': self.linebourse_tokens[account._univers]}
)
self.linebourse.session.headers['X-XSRF-TOKEN'] = self.linebourse.session.cookies.get('XSRF-TOKEN')
for investment in self.linebourse.iter_investments(account.id.strip('0').split('.')[0]):
yield investment
else:
raise NotImplementedError()
else:
# Compte espèces
yield create_french_liquidity(account.balance)
else:
raise NotImplementedError()
@need_login
def iter_market_orders(self, account):
if account.type not in (Account.TYPE_MARKET, Account.TYPE_PEA):
return
if 'Portefeuille Titres' in account.label:
if account._is_in_linebourse:
self.move_to_universe(account._univers)
self.linebourse_login_for_universe(account._univers)
self.linebourse.location(
self.linebourse_urls[account._univers],
data={'SJRToken': self.linebourse_tokens[account._univers]}
)
self.linebourse.session.headers['X-XSRF-TOKEN'] = self.linebourse.session.cookies.get('XSRF-TOKEN')
for order in self.linebourse.iter_market_orders(account.id.strip('0').split('.')[0]):
yield order
@need_login
def get_profile(self):
self.get_universes()
self.profile.go()
return self.page.get_profile()
@need_login
def fill_account(self, account, fields):
if account.type == Account.TYPE_CHECKING and 'iban' in fields:
self.iban.go(number=account._number)
self.page.set_iban(account=account)
@need_login
def iter_transfer_recipients(self, account):
self.move_to_universe(account._univers)
self.update_headers()
try:
self.emitters_list.go(json={
'typeVirement': 'C',
})
except ClientError as e:
if e.response.status_code == 403:
msg = e.response.json().get('erreur', {}).get('libelle', '')
if msg == "Cette fonctionnalité n'est pas disponible avec votre compte.":
# Means the account cannot emit transfers
return
raise
if not self.page.can_account_emit_transfer(account.id):
return
self.update_headers()
account_id = account.id.split('.')[0]
self.recipient_list.go(json={
'numeroCompteDebiteur': account_id,
'typeVirement': 'C',
})
for obj in self.page.iter_external_recipients():
yield obj
for obj in self.page.iter_internal_recipients():
if obj.id != account.id:
yield obj
def do_strong_authent_recipient(self, recipient):
self.list_authent.go(context=self.context['contextAppli'])
self.auth_method = self.page.get_handled_auth_methods()
if not self.auth_method:
raise AuthMethodNotImplemented()
self.need_reload_state = self.auth_method != 'password'
if self.auth_method == 'password':
return self.validate_strong_authent_recipient(self.password)
elif self.auth_method == 'otp':
raise AddRecipientStep(
recipient,
Value(
'otp',
label="Veuillez générez un e-Code sur votre application BRED puis saisir cet e-Code ici",
),
)
elif self.auth_method == 'notification':
self.update_headers()
self.init_authent.go(json={
'context': self.context['contextAppli'],
'type_auth': 'NOTIFICATION',
'type_phone': 'P',
})
self.authent_id = self.page.get_authent_id()
raise AppValidation(
resource=recipient,
message='Veuillez valider la notification sur votre application mobile BRED',
)
elif self.auth_method == 'sms':
self.update_headers()
self.send_sms.go(json={
'contextAppli': self.context['contextAppli'],
'context': self.context['context'],
})
raise AddRecipientStep(
recipient,
Value('code', label='Veuillez saisir le code reçu par SMS'),
)
def validate_strong_authent_recipient(self, auth_value):
self.update_headers()
self.check_otp.go(
auth_method=self.auth_method,
json={
'contextAppli': self.context['contextAppli'],
'context': self.context['context'],
'otp': auth_value,
},
)
error = self.page.get_error()
if error:
raise AddRecipientBankError(message=error)
def new_recipient(self, recipient, **params):
if 'otp' in params:
self.validate_strong_authent_recipient(params['otp'])
elif 'code' in params:
self.validate_strong_authent_recipient(params['code'])
elif 'resume' in params:
self.handle_polling(enrol=False)
return self.init_new_recipient(recipient, **params)
@need_login
def init_new_recipient(self, recipient, **params):
for _ in range(3):
# The goal of this is to find the maximum allowed, by Bred,
# limit for transfers on new recipient. For this we do a request
# with an absurdly high limit , and the error message will tell us
# what is that maximum allowed limit by the bank.
#
# 5000€ is the default value for the recipient transfer limit
# on BRED Connect.
# TODO: What if the transfer limit is more than 5000€?
json_data = {
'nom': recipient.label,
'iban': recipient.iban,
'numCompte': '',
'plafond': str(int(self.recipient_transfer_limit or 5000)),
'typeDemande': 'A',
}
try:
self.update_headers()
self.add_recipient.go(json=json_data)
except ClientError as e:
if e.response.status_code != 449:
# Status code 449 means we need to do strong authentication
raise
self.context = e.response.json()['content']
self.do_strong_authent_recipient(recipient)
# Password authentication do not raise error, so we need
# to re-execute the request here.
if self.auth_method == 'password':
self.update_headers()
self.add_recipient.go(json=json_data)
# If the error we encounter here lists a transfer limit, we use
# this as the new recipient transfer limit for the account.
# Otherwise, it means that our current transfer limit has been
# accepted and either the recipient has been added successfully,
# or that another error has occurred.
recipient_transfer_limit = self.page.get_transfer_limit()
if recipient_transfer_limit:
self.recipient_transfer_limit = recipient_transfer_limit
continue
break
else:
raise AssertionError(
'Could not find the recipient transfer limit!',
)
error_code = self.page.get_error_code()
# '3200' is a warning, but the recipient has been added successfully.
if error_code != '3200':
error = self.page.get_error()
if error:
raise AddRecipientBankError(message=error)
return recipient
@need_login
def init_transfer(self, transfer, account, recipient, **params):
self.move_to_universe(account._univers)
account_id = account.id.split('.')[0]
poste = account.id.split('.')[1]
amount = transfer.amount.quantize(Decimal(10) ** -2)
if not amount % 1:
# Number is an integer without floating numbers.
# We need to not display the floating points in the request
# if the number is an integer or the request will not work.
amount = int(amount)
json_data = {
'compteDebite': account_id,
'posteDebite': poste,
'deviseDebite': account.currency,
'deviseCredite': recipient.currency,
'dateEcheance': transfer.exec_date.strftime('%d/%m/%Y'),
'montant': str(amount),
'motif': transfer.label,
'virementListe': True,
'plafondBeneficiaire': '',
'nomBeneficiaire': recipient.label,
'checkBeneficiaire': False,
'instantPayment': False,
}
if recipient.category == "Interne":
recipient_id_split = recipient.id.split('.')
json_data['compteCredite'] = recipient_id_split[0]
json_data['posteCredite'] = recipient_id_split[1]
else:
json_data['iban'] = recipient.iban
self.update_headers()
self.create_transfer.go(json=json_data)
error = self.page.get_error()
if error:
raise TransferBankError(message=error)
transfer.amount = self.page.get_transfer_amount()
transfer.currency = self.page.get_transfer_currency()
# The same data is needed to validate the transfer.
transfer._json_data = json_data
return transfer
@need_login
def execute_transfer(self, transfer, **params):
self.update_headers()
# This sends an email to the user to tell him that a transfer
# has been created.
self.validate_transfer.go(json=transfer._json_data)
error = self.page.get_error()
if error:
raise TransferBankError(message=error)
return transfer
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/bred/bred/pages.py 0000664 0000000 0000000 00000051257 14575653726 0030653 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2018 Célande Adrien
#
# This file is part of a woob module.
#
# This woob 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 woob 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 woob module. If not, see .
import re
from datetime import date
from decimal import Decimal
from json import loads
from woob.tools.date import parse_french_date
from woob.exceptions import (
BrowserIncorrectPassword, BrowserUnavailable, AuthMethodNotImplemented,
)
from woob.browser.pages import JsonPage, LoggedPage, HTMLPage, pagination
from woob.capabilities import NotAvailable
from woob.capabilities.bank import Account, Loan
from woob.capabilities.bank.wealth import Investment
from woob.tools.capabilities.bank.investments import is_isin_valid
from woob.capabilities.profile import Person
from woob.browser.filters.standard import (
CleanText, CleanDecimal, Coalesce, Currency, Env, Eval,
Field, Format, FromTimestamp, QueryValue, Type,
)
from woob.browser.filters.json import Dict
from woob.browser.elements import DictElement, ItemElement, method
from woob.tools.capabilities.bank.transactions import FrenchTransaction
class Transaction(FrenchTransaction):
PATTERNS = [
(re.compile(r'^.*Virement (?P.*)'), FrenchTransaction.TYPE_TRANSFER),
(re.compile(r'PRELEV SEPA (?P.*)'), FrenchTransaction.TYPE_ORDER),
(re.compile(r'.*Prélèvement.*'), FrenchTransaction.TYPE_ORDER),
(re.compile(r'^(REGL|Rgt)(?P.*)'), FrenchTransaction.TYPE_ORDER),
(
re.compile(r'^(?P.*) Carte \d+( CB\.XXXXX(\d){3})? LE (?P\d{2})/(?P\d{2})/(?P\d{2})'),
FrenchTransaction.TYPE_CARD
),
(
re.compile(r'^(?P.*) Carte \d+ CB\.\w+ LE (?P\d{2})\/(?P\d{2})\/(?P\d{2})'),
FrenchTransaction.TYPE_CARD,
),
(re.compile(r'^Débit mensuel.*'), FrenchTransaction.TYPE_CARD_SUMMARY),
(
re.compile(r"^Retrait d'espèces à un DAB (?P.*) CARTE [X\d]+ LE (?P\d{2})/(?P\d{2})/(?P\d{2})"),
FrenchTransaction.TYPE_WITHDRAWAL
),
(re.compile(r'^Paiement de chèque (?P.*)'), FrenchTransaction.TYPE_CHECK),
(re.compile(r'^(Cotisation|Intérêts) (?P.*)'), FrenchTransaction.TYPE_BANK),
(
re.compile(r'(?PRemise (de )?(C|c)hèque\(s\) \d{7})( (V|v)otre remise du : (?P\d{2})/(?P\d{2})/(?P\d{4}))?'),
FrenchTransaction.TYPE_DEPOSIT
),
(re.compile(r'^Versement (?P.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile(r'^Frais transaction carte (\d){7}'), FrenchTransaction.TYPE_BANK),
(re.compile(r'^Frais tenue de compte'), FrenchTransaction.TYPE_BANK),
(re.compile(r'^Commission virement instantané émis'), FrenchTransaction.TYPE_BANK),
(
re.compile(r'^(?P.*)LE (?P\d{2})/(?P\d{2})/(?P\d{2})\s*(?P.*)'),
FrenchTransaction.TYPE_UNKNOWN
),
]
class MyJsonPage(JsonPage):
def get_content(self):
return self.doc.get('content', {})
class HomePage(LoggedPage, HTMLPage):
pass
class LoginPage(LoggedPage, HTMLPage):
pass
class AccountsTwoFAPage(JsonPage):
pass
class InitAuthentPage(JsonPage):
def get_authent_id(self):
return Dict('content')(self.doc)
class AuthentResultPage(JsonPage):
@property
def logged(self):
return self.get_status() == 'AUTHORISED'
def get_status(self):
return Dict('content/status', default=None)(self.doc)
class SendSmsPage(JsonPage):
pass
class TrustedDevicesPage(JsonPage):
def get_error(self):
error = CleanText(Dict('erreur/libelle'))(self.doc)
if error != 'OK':
return error
class CheckOtpPage(TrustedDevicesPage):
@property
def logged(self):
return CleanText(Dict('erreur/libelle'))(self.doc) == 'OK'
class UniversePage(LoggedPage, MyJsonPage):
def get_universes(self):
universe_data = self.get_content()
universes = {}
universes[universe_data['universKey']] = universe_data['title']
for universe in universe_data.get('menus', {}):
universes[universe['universKey']] = universe['title']
return universes
class TokenPage(MyJsonPage):
pass
class MoveUniversePage(LoggedPage, HTMLPage):
pass
class SwitchPage(LoggedPage, JsonPage):
pass
class LoansPage(LoggedPage, JsonPage):
@method
class iter_loans(DictElement):
def find_elements(self):
return self.el.get('content', [])
class item(ItemElement):
klass = Loan
obj_id = Format(
'%s.%s',
CleanText(Dict('comptePrets')),
CleanText(Dict('numeroDossier')),
)
obj_label = Format(
'%s %s',
CleanText(Dict('intitule')),
CleanText(Dict('libellePrets')),
)
obj_balance = CleanDecimal(Dict('montantCapitalDu/valeur'), sign='-')
obj_currency = Currency(Dict('montantCapitalDu/monnaie/code'))
obj_next_payment_amount = CleanDecimal.SI(
Dict(
'montantProchaineEcheance/valeur',
default=NotAvailable,
),
default=NotAvailable,
)
obj_next_payment_date = FromTimestamp(
Dict(
'dateProchaineEcheance',
default=NotAvailable,
),
millis=True,
default=NotAvailable,
)
obj_last_payment_amount = CleanDecimal.SI(
Dict(
'montantEcheancePrecedent/valeur',
default=NotAvailable,
),
default=NotAvailable,
)
obj_duration = Type(
CleanText(Dict('dureePret', default=None), default=NotAvailable),
type=int,
default=NotAvailable,
)
obj_type = Account.TYPE_LOAN
obj_rate = CleanDecimal.SI(Dict('tauxNominal'))
obj_total_amount = CleanDecimal.SI(Dict('montantInitial/valeur'))
obj_maturity_date = FromTimestamp(Dict('dateFinPret'), millis=True)
obj_insurance_amount = CleanDecimal.SI(Dict('montantPartAssurance/valeur'))
obj__univers = Env('current_univers')
obj__number = Field('id')
class AccountsPage(LoggedPage, MyJsonPage):
ACCOUNT_TYPES = {
'000': Account.TYPE_CHECKING, # Compte à vue
'001': Account.TYPE_SAVINGS, # Livret Ile de France
'002': Account.TYPE_SAVINGS, # Livret Seine-et-Marne & Aisne
'003': Account.TYPE_SAVINGS, # Livret Normandie
'004': Account.TYPE_SAVINGS, # Livret Guadeloupe
'005': Account.TYPE_SAVINGS, # Livret Martinique/Guyane
'006': Account.TYPE_SAVINGS, # Livret Réunion/Mayotte
'011': Account.TYPE_CARD, # Carte bancaire
'013': Account.TYPE_LOAN, # LCR (Lettre de Change Relevé)
'020': Account.TYPE_SAVINGS, # Compte sur livret
'021': Account.TYPE_SAVINGS,
'022': Account.TYPE_SAVINGS, # Livret d'épargne populaire
'023': Account.TYPE_SAVINGS, # LDD Solidaire
'025': Account.TYPE_SAVINGS, # Livret Fidélis
'027': Account.TYPE_SAVINGS, # Livret A
'037': Account.TYPE_SAVINGS,
'070': Account.TYPE_SAVINGS, # Compte Epargne Logement
'077': Account.TYPE_SAVINGS, # Livret Bambino
'078': Account.TYPE_SAVINGS, # Livret jeunes
'080': Account.TYPE_SAVINGS, # Plan épargne logement
'081': Account.TYPE_SAVINGS,
'086': Account.TYPE_SAVINGS, # Compte épargne Moisson
'097': Account.TYPE_CHECKING, # Solde en devises
'730': Account.TYPE_DEPOSIT, # Compte à terme Optiplus
'999': Account.TYPE_MARKET, # no label, we use 'Portefeuille Titres' if needed
}
def iter_accounts(self, accnum, current_univers):
seen = set()
for content in self.get_content():
if accnum != '00000000000' and content['numero'] != accnum:
continue
for poste in content['postes']:
a = Account()
a._number = content['numeroLong']
a._nature = poste['codeNature']
a._codeSousPoste = poste['codeSousPoste'] if 'codeSousPoste' in poste else None
a._consultable = poste['consultable']
a._univers = current_univers
a._parent_number = None
a.id = '%s.%s' % (a._number, a._nature)
if content['comptePEA']:
a.type = Account.TYPE_PEA
else:
a.type = self.ACCOUNT_TYPES.get(poste['codeNature'], Account.TYPE_UNKNOWN)
if a.type == Account.TYPE_UNKNOWN:
self.logger.warning("unknown type %s" % poste['codeNature'])
if a.type != Account.TYPE_CHECKING:
a._parent_number = a._number
if 'numeroDossier' in poste and poste['numeroDossier']:
a._file_number = poste['numeroDossier']
a.id += '.%s' % a._file_number
if poste['postePortefeuille']:
a.label = '{} Portefeuille Titres'.format(content['intitule'].strip())
a.balance = Decimal(str(poste['montantTitres']['valeur']))
a.currency = poste['montantTitres']['monnaie']['code'].strip()
if not a.balance and not a.currency and 'dateTitres' not in poste:
continue
yield a
continue
if 'libelle' not in poste:
continue
a.label = ' '.join([content['intitule'].strip(), poste['libelle'].strip()])
if poste['numeroDossier']:
a.label = '{} n°{}'.format(a.label, poste['numeroDossier'])
a.balance = Decimal(str(poste['solde']['valeur']))
a.currency = poste['solde']['monnaie']['code'].strip()
# Some accounts may have balance currency
if 'Solde en devises' in a.label and a.currency != u'EUR':
a.id += str(poste['monnaie']['codeSwift'])
if a.id in seen:
# some accounts like "compte à terme fidélis" have the same _number and _nature
# but in fact are kind of closed, so worthless...
self.logger.warning('ignored account id %r (%r) because it is already used', a.id, poste.get('numeroDossier'))
continue
seen.add(a.id)
yield a
class IbanPage(LoggedPage, MyJsonPage):
def set_iban(self, account):
iban_response = self.get_content()
account.iban = CleanText(Dict('iban', default=None), default=NotAvailable)(iban_response)
class LinebourseLoginPage(LoggedPage, JsonPage):
def get_linebourse_url(self):
return Dict('content/url', default=None)(self.doc)
def get_linebourse_token(self):
return Dict('content/token', default=None)(self.doc)
class LifeInsurancesPage(LoggedPage, JsonPage):
def check_error(self):
error_code = Dict('erreur/code', default=None)(self.doc)
if error_code and int(error_code) != 0:
message = Dict('erreur/libelle', default=None)(self.doc)
if error_code in ('90000', '1000'):
raise BrowserUnavailable()
raise AssertionError(f'Unhandled error {error_code}: {message}')
@method
class iter_lifeinsurances(DictElement):
def condition(self):
return 'content' in self.el
item_xpath = 'content'
class iter_accounts(DictElement):
item_xpath = 'avoirs/contrats'
def get_owner(self):
return CleanText(Dict('titulaire'))(self)
class item(ItemElement):
klass = Account
obj_balance = CleanDecimal(Dict('valorisation'))
obj_type = Account.TYPE_LIFE_INSURANCE
obj_currency = 'EUR'
obj__univers = Env('univers')
obj__number = Field('id')
def obj_id(self):
return Eval(str, Dict('numero'))(self)
def obj_label(self):
return '%s - %s' % (CleanText(Dict('libelleProduit'))(self), self.parent.get_owner())
def obj__parent_number(self):
return CleanText(Dict('cptRattachement'))(self).rstrip('0')
# Investments are already present in this JSON,
# so we fill the lists of Investment objects now
class obj__investments(DictElement):
item_xpath = 'allocations'
class item(ItemElement):
klass = Investment
obj_label = CleanText(Dict('libelle'))
obj_valuation = CleanDecimal(Dict('montant'))
def obj_code_type(self):
if is_isin_valid(CleanText(Dict('code'))(self)):
return Investment.CODE_TYPE_ISIN
return NotAvailable
def obj_code(self):
code = CleanText(Dict('code'))(self)
if is_isin_valid(code):
return code
return NotAvailable
class SearchPage(LoggedPage, JsonPage):
def check_error(self):
if Dict('erreur/code')(self.doc) != '0':
raise BrowserUnavailable("API sent back an error code")
def get_max_transactions(self):
return Dict('content/total')(self.doc)
@pagination
@method
class iter_history(DictElement):
item_xpath = 'content/operations'
def next_page(self):
# All transaction pages show the total number of transactions.
# To obtain the next page, we simply use the _make_api_call method.
# This method makes a POST to the url of the transaction page and
# we increment the offset by 50 each time until the total number
# of transactions or a limit of 50 000 is reached.
current_offset = self.page.response.request.body
current_offset = int(loads(current_offset)['from'])
current_offset += self.env['max_length']
if current_offset < self.env['max_transactions'] and current_offset < 50000:
return self.page.browser._make_api_call(
account=self.env['account'], start_date=self.env['start_date'],
end_date=self.env['end_date'], offset=current_offset, max_length=self.env['max_length'],
)
class item(ItemElement):
klass = Transaction
obj_raw = Transaction.Raw(Format('%s %s', CleanText(Dict('libelle')), CleanText(Dict('details'))))
obj_label = CleanText(Dict('libelle'))
obj_amount = CleanDecimal(Dict('montant'))
obj_category = Dict('categorie', default=NotAvailable)
def get_date(self, timestamp):
if isinstance(timestamp, date):
return timestamp
return date.fromtimestamp(timestamp / 1000)
def obj__coming(self):
_coming = Dict('intraday')(self)
_coming |= (Field('date')(self) > self.env['today'])
return _coming
def obj_id(self):
# If op['id'] is an uuid, it will be a different one for every scrape
re_uuid = re.compile(r'[0-9a-fA-F]{8}-([0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}')
tr_id = Dict('id')(self)
if Dict('intraday')(self) or re_uuid.match(tr_id):
return ''
return tr_id
def obj_date(self):
if self.env['account_type'] == Account.TYPE_CARD:
tr_date = Coalesce(
Dict('dateDebit', default=NotAvailable),
Dict('dateOperation')
)(self)
else:
tr_date = Dict('dateOperation')(self)
return self.get_date(tr_date)
def obj_rdate(self):
if self.env['account_type'] == Account.TYPE_CARD:
rdate = self.obj.rdate or Dict('dateOperation', default=NotAvailable)(self)
elif self.env['account_type'] == Account.TYPE_CHECKING:
json_date = Coalesce(
Dict('dateValeur', default=NotAvailable),
Dict('dateOperation', default=NotAvailable),
default=NotAvailable
)(self)
rdate = self.obj.rdate or json_date
else:
rdate = Coalesce(
Dict('dateOperation', default=NotAvailable),
Dict('dateDebit', default=NotAvailable),
default=NotAvailable
)(self)
if rdate:
rdate = self.get_date(rdate)
if self.env['account_type'] == Account.TYPE_CHECKING and rdate and rdate > Field('date')(self):
rdate = NotAvailable
return rdate
def obj_vdate(self):
vdate = Coalesce(
Dict('dateValeur', default=NotAvailable),
Dict('dateDebit', default=NotAvailable),
Dict('dateOperation', default=NotAvailable)
)(self)
if vdate:
vdate = self.get_date(vdate)
return vdate
def obj_type(self):
if self.obj.type == Transaction.TYPE_CARD and self.env['account_type'] == Account.TYPE_CARD:
return Transaction.TYPE_DEFERRED_CARD
return self.obj.type
def validate(self, obj):
# If rdate and date of the transaction are too far apart we skip the transaction
if not Field('rdate')(self):
return True
return (abs(Field('rdate')(self).year - Field('date')(self).year) < 2)
class ProfilePage(LoggedPage, MyJsonPage):
def get_profile(self):
profile = Person()
content = self.get_content()
profile.name = content['prenom'] + ' ' + content['nom']
profile.address = content['adresse'] + ' ' + content['codePostal'] + ' ' + content['ville']
profile.country = content['pays']
profile.birth_date = parse_french_date(content['dateNaissance']).date()
return profile
class ErrorPage(LoggedPage, HTMLPage):
def on_load(self):
if 'gestion-des-erreurs/erreur-pwd' in self.url:
raise BrowserIncorrectPassword(CleanText('//h3')(self.doc))
if 'gestion-des-erreurs/opposition' in self.url:
# need a case to retrieve the error message
raise BrowserIncorrectPassword('Votre compte a été désactivé')
if '/pages-gestion-des-erreurs/erreur-technique' in self.url:
errmsg = CleanText('//h4')(self.doc)
raise BrowserUnavailable(errmsg)
if '/pages-gestion-des-erreurs/message-tiers-oppose' in self.url:
# need a case to retrieve the error message
raise AuthMethodNotImplemented("Impossible de se connecter au compte car l'identification en 2 étapes a été activée")
class ErrorCodePage(HTMLPage):
def get_code(self):
return QueryValue(None, 'errorCode').filter(self.url)
class ErrorMsgPage(HTMLPage):
def get_msg(self):
return CleanText('//label[contains(@class, "error")]', default=None)(self.doc)
class UnavailablePage(HTMLPage):
def is_here(self):
return CleanText('//h1[contains(text(), "Site en maintenance")]', default=None)(self.doc)
def on_load(self):
msg = CleanText('//div[contains(text(), "intervention technique est en cours")]', default=None)(self.doc)
if msg:
raise BrowserUnavailable(msg)
raise AssertionError('Ended up to this error page, message not handled yet.')
transfer_pages.py 0000664 0000000 0000000 00000012561 14575653726 0032473 0 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-bred-bred/modules/bred/bred # -*- coding: utf-8 -*-
# Copyright(C) 2020 Guillaume Risbourg
#
# This file is part of a woob module.
#
# This woob 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 woob 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 woob module. If not, see .
from datetime import date
import re
from woob.capabilities.bank import Recipient
from woob.browser.pages import LoggedPage, JsonPage
from woob.browser.elements import ItemElement, DictElement, method
from woob.browser.filters.standard import (
CleanText, Currency, Format, CleanDecimal, Regexp,
)
from woob.browser.filters.json import Dict
class ListAuthentPage(LoggedPage, JsonPage):
def get_handled_auth_methods(self):
# Order in auth_methods is important, the first method we encouter
# is the strong authentification we are going to do.
auth_methods = ('password', 'otp', 'sms', 'notification')
for auth_method in auth_methods:
if Dict('content/%s' % auth_method)(self.doc):
return auth_method
class EmittersListPage(LoggedPage, JsonPage):
def can_account_emit_transfer(self, account_id):
code = Dict('erreur/code')(self.doc)
if code == '90624':
# Not the owner of the account:
# Nous vous précisons que votre pouvoir ne vous permet pas
# d'effectuer des virements de ce type au débit du compte sélectionné.
return False
elif code == '90600':
# "Votre demande de virement ne peut être prise en compte actuellement
# The user is probably not allowed to do transfers
return False
elif code != '0':
raise AssertionError('Unhandled code %s in transfer emitter selection' % code)
for obj in Dict('content')(self.doc):
for account in Dict('postes')(obj):
_id = '%s.%s' % (
Dict('numero')(obj),
Dict('codeNature')(account),
)
if _id == account_id:
return True
return False
class RecipientListPage(LoggedPage, JsonPage):
@method
class iter_external_recipients(DictElement):
item_xpath = 'content/listeComptesCExternes'
# The id is the iban, and exceptionally there could be the same
# recipient multiple times when the bic of the recipient changed
ignore_duplicate = True
class item(ItemElement):
klass = Recipient
obj_id = CleanText(Dict('id'))
obj_iban = CleanText(Dict('iban'))
obj_bank_name = CleanText(Dict('nomBanque'))
obj_currency = Currency(Dict('monnaie/code'))
obj_enabled_at = date.today()
obj_label = CleanText(Dict('libelle'))
obj_category = 'Externe'
@method
class iter_internal_recipients(DictElement):
def find_elements(self):
for obj in Dict('content/listeComptesCInternes')(self):
number = Dict('numero')(obj)
for account in Dict('postes')(obj):
account['number'] = number
yield account
class item(ItemElement):
klass = Recipient
obj_id = Format('%s.%s', Dict('number'), Dict('codeNature'))
obj_label = CleanText(Dict('libelle'))
obj_enabled_at = date.today()
obj_currency = Currency(Dict('monnaie/code'))
obj_bank_name = 'BRED'
obj_category = 'Interne'
class ErrorJsonPage(JsonPage):
def get_error_code(self):
return CleanText(Dict('erreur/code'))(self.doc)
def get_error(self):
error = CleanText(Dict('erreur/libelle'))(self.doc)
if error != 'OK':
# The message is some partial html, the useful message
# is at the beginning, before every html tag so we just retrieve the
# first part of the message before any html tag.
# If the message begins with html tags, the regex will skip those.
m = re.search(r'^(?:<[^>]+>)*(.+?)(?=<[^>]+>)', error)
if m:
return m.group(1)
return error
class AddRecipientPage(LoggedPage, ErrorJsonPage):
def get_transfer_limit(self):
error = self.get_error()
if not error:
return None
# The message is some partial html in a json key, we can't use
# the html tags to limit the search.
text_limit = Regexp(
pattern=r"(?:plafond de virement est limité à|l'augmenter, au delà de) ([\d ,€]+)",
default='',
).filter(error)
return CleanDecimal.French(default=None).filter(text_limit)
class TransferPage(LoggedPage, ErrorJsonPage):
def get_transfer_amount(self):
return CleanDecimal(Dict('content/montant/valeur'))(self.doc)
def get_transfer_currency(self):
return Currency(Dict('content/montant/monnaie/code'))(self.doc)