# -*- coding: utf-8 -*-
# Copyright(C) 2012-2019 Romain Bignon
#
# 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 .
from __future__ import unicode_literals
from decimal import Decimal
import re
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, ActionNeeded
from weboob.browser.exceptions import ServerError
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, CardsPage, CardHistoryPage, ProfilePage,
)
__all__ = ['CragrAPI']
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)
logged_out = URL(r'.*', LoggedOutPage)
contracts_page = URL(r'particulier/operations/.rechargement.contexte.html\?idBamIndex=(?P)',
r'association/operations/.rechargement.contexte.html\?idBamIndex=(?P)',
r'professionnel/operations/.rechargement.contexte.html\?idBamIndex=(?P)', ContractsPage)
accounts_page = URL(r'particulier/operations/synthese.html',
r'association/operations/synthese.html',
r'professionnel/operations/synthese.html', AccountsPage)
account_details = URL(r'particulier/operations/synthese/jcr:content.produits-valorisation.json/(?P)',
r'association/operations/synthese/jcr:content.produits-valorisation.json/(?P)',
r'professionnel/operations/synthese/jcr:content.produits-valorisation.json/(?P)', 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)
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', 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', 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', CardHistoryPage)
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)
def __init__(self, website, *args, **kwargs):
super(CragrAPI, self).__init__(*args, **kwargs)
website = website.replace('.fr', '')
self.region = re.sub('^m\.', 'www.credit-agricole.fr/', website)
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'
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)
try:
self.security_check.go(data=form)
except ServerError as exc:
# Wrongpass returns a 500 server error...
error = exc.response.json().get('error')
if error:
message = error.get('message', '')
if 'Votre identification est incorrecte' in message:
raise BrowserIncorrectPassword()
if 'obtenir un nouveau code' in message:
raise ActionNeeded(message)
elif 'Un incident technique' in message:
# If it is a technical error, we try login again
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)
assert False, 'Unhandled Server Error encountered: %s' % error.get('message', '')
# accounts_url may contain '/particulier', '/professionnel' 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!'
# Once the security check is passed, we are logged in.
@need_login
def get_accounts_list(self):
# Determine how many spaces are present on the connection:
self.location(self.accounts_url)
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)
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)
except ServerError:
self.logger.warning('Server returned error 500 twice when trying to access space %s, this space will be skipped' % contract)
continue
# The main account is not located at the same place in the JSON.
main_account = self.page.get_main_account()
main_account.owner_type = self.page.get_owner_type()
main_account._contract = contract
accounts_list = list(self.page.iter_accounts())
for account in accounts_list:
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_element_contrat, balance) pairs in the account_details JSON:
categories = {int(account._category) for account in accounts_list if account._category != None}
account_balances = {}
loan_ids = {}
for category in categories:
self.account_details.go(category=category)
account_balances.update(self.page.get_account_balances())
loan_ids.update(self.page.get_loan_ids())
if main_account.type == Account.TYPE_CHECKING:
params = {
'compteIdx': int(main_account._index),
'grandeFamilleCode': 1,
}
self.account_iban.go(params=params)
iban = self.page.get_iban()
if is_iban_valid(iban):
main_account.iban = iban
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_element_contrat, NotAvailable)
if account.type == Account.TYPE_CHECKING:
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 account.type == Account.TYPE_LOAN:
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 = 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()
@need_login
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()
@need_login
def iter_advisor(self):
raise BrowserUnavailable()
@need_login
def get_profile(self):
#self.profile.go()
raise BrowserUnavailable()
@need_login
def iter_transfer_recipients(self, account):
raise BrowserUnavailable()
@need_login
def init_transfer(self, transfer, **params):
raise BrowserUnavailable()
@need_login
def execute_transfer(self, transfer, **params):
raise BrowserUnavailable()
@need_login
def build_recipient(self, recipient):
raise BrowserUnavailable()
@need_login
def new_recipient(self, recipient, **params):
raise BrowserUnavailable()