Commit ca86fec9 authored by Sylvie Ye's avatar Sylvie Ye Committed by Romain Bignon

[cragr] add iter recipient and transfer to api website

parent 8b809438
......@@ -23,8 +23,8 @@ 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.capabilities.bank import Account, Transaction, AccountNotFound, RecipientNotFound
from weboob.capabilities.base import empty, NotAvailable, strict_find_object
from weboob.browser import LoginBrowser, URL, need_login
from weboob.exceptions import BrowserUnavailable, BrowserIncorrectPassword, ActionNeeded
from weboob.browser.exceptions import ServerError, BrowserHTTPNotFound
......@@ -37,6 +37,9 @@ from .pages import (
TokenPage, IbanPage, HistoryPage, CardsPage, CardHistoryPage, NetfincaRedirectionPage, PredicaRedirectionPage,
PredicaInvestmentsPage, ProfilePage, ProfileDetailsPage, ProProfileDetailsPage,
)
from .transfer_pages import (
RecipientsPage, TransferPage, TransferTokenPage,
)
from weboob.tools.capabilities.bank.investments import create_french_liquidity
......@@ -129,6 +132,18 @@ class CragrAPI(LoginBrowser):
r'agriculteur/operations/profil/infos-personnelles/controler-coordonnees.html',
r'entreprise/operations/profil/infos-personnelles/controler-coordonnees.html', ProProfileDetailsPage)
recipients = URL('(?P<space>.*)/operations/(?P<op>.*)/virement/jcr:content.accounts.json',
RecipientsPage)
transfer_token = URL('(?P<space>.*)/operations/(?P<op>.*)/virement.npcgeneratetoken.json\?tokenTypeId=1',
TransferTokenPage)
transfer = URL('(?P<space>.*)/operations/(?P<op>.*)/virement/jcr:content.check-transfer.json',
TransferPage)
transfer_recap = URL('(?P<space>.*)/operations/(?P<op>.*)/virement/jcr:content.transfer-data.json\?useSession=true',
TransferPage)
transfer_exec = URL('(?P<space>.*)/operations/(?P<op>.*)/virement/jcr:content.process-transfer.json',
TransferPage)
def __init__(self, website, *args, **kwargs):
super(CragrAPI, self).__init__(*args, **kwargs)
website = website.replace('.fr', '')
......@@ -522,16 +537,130 @@ class CragrAPI(LoginBrowser):
return profile
@need_login
def iter_transfer_recipients(self, account):
raise BrowserUnavailable()
def get_account_transfer_space_info(self, account):
self.go_to_account_space(account._contract)
space = self.session.cookies['marche']
connection_id = self.page.get_connection_id()
operations = {
'particulier': 'moyens-paiement',
'professionnel': 'paiements-encaissements',
'association': 'paiements-encaissements',
'entreprise': 'paiements-encaissements',
}
referer = self.absurl('/%s/operations/%s/virement.html.html' % (space, operations[space]))
return space, operations[space], referer, connection_id
@need_login
def iter_debit_accounts(self):
assert self.recipients.is_here()
for index, debit_accounts in enumerate(self.page.iter_debit_accounts()):
debit_accounts._index = index
yield debit_accounts
@need_login
def iter_transfer_recipients(self, account, transfer_space_info=None):
# avoid to call `get_account_transfer_space_info()` several time
if transfer_space_info:
space, operation, referer = transfer_space_info
else:
space, operation, referer, _ = self.get_account_transfer_space_info(account)
self.recipients.go(space=space, op=operation, headers={'Referer': referer})
if not self.page.is_sender_account(account.id):
return
for index, internal_rcpt in enumerate(self.page.iter_internal_recipient(account_id=account.id)):
internal_rcpt._index = index
yield internal_rcpt
# can't use 'ignore_duplicate' in DictElement because we need the 'index' to do transfer
seen = set()
for index, external_rcpt in enumerate(self.page.iter_external_recipient()):
external_rcpt._index = index
if not external_rcpt.iban in seen:
seen.add(external_rcpt.iban)
yield external_rcpt
@need_login
def init_transfer(self, transfer, **params):
raise BrowserUnavailable()
# first, get _account on account list to get recipient
_account = strict_find_object(self.get_accounts_list(), id=transfer.account_id, error=AccountNotFound)
# get information to go on transfer page
space, operation, referer, connection_id = self.get_account_transfer_space_info(account=_account)
recipient = strict_find_object(
self.iter_transfer_recipients(_account, transfer_space_info=(space, operation, referer)),
id=transfer.recipient_id,
error=RecipientNotFound
)
# Then, get account on transfer account list to get index and other information
account = strict_find_object(self.iter_debit_accounts(), id=_account.id, error=AccountNotFound)
# get token and transfer token to init transfer
token = self.token_page.go().get_token()
transfer_token = self.transfer_token.go(space=space, op=operation, headers={'Referer': referer}).get_token()
data = {
'connexionId': connection_id,
'cr': self.session.cookies['caisse-regionale'],
'creditAccountIban': recipient.iban,
'creditAccountIndex': recipient._index,
'debitAccountIndex': account._index,
'debitAccountNumber': account.number,
'externalAccount': recipient.category == 'Externe',
'recipientName': recipient.label,
'transferAmount': transfer.amount,
'transferComplementaryInformation1': transfer.label,
'transferComplementaryInformation2': '',
'transferComplementaryInformation3': '',
'transferComplementaryInformation4': '',
'transferCurrencyCode': account.currency,
'transferDate': transfer.exec_date.strftime('%d/%m/%Y'),
'transferFrequency': 'U',
'transferRef': '',
'transferType': 'UNIQUE',
'typeCompte': account.label,
}
# init transfer request
self.transfer.go(
space=space,
op=operation,
headers={'Referer': referer, 'CSRF-Token': token, 'NPC-Generated-Token': transfer_token},
json=data
)
assert self.page.check_transfer()
# get recap because it's not returned by init transfer request
self.transfer_recap.go(
space=space,
op=operation,
headers={'Referer': self.absurl('/%s/operations/%s/virement.postredirect.html' % (space, operation))}
)
# information needed to exec transfer
transfer._space = space
transfer._operation = operation
transfer._token = token
transfer._connection_id = connection_id
return self.page.handle_response(transfer)
@need_login
def execute_transfer(self, transfer, **params):
raise BrowserUnavailable()
self.transfer_exec.go(
space=transfer._space,
op=transfer._operation,
headers={
'Referer': self.absurl('/%s/operations/%s/virement.postredirect.html' % (transfer._space, transfer._operation)),
'CSRF-Token': transfer._token
},
json={'connexionId': transfer._connection_id}
)
assert self.page.check_transfer_exec()
return transfer
@need_login
def build_recipient(self, recipient):
......
......@@ -171,6 +171,13 @@ class AccountsPage(LoggedPage, JsonPage):
}
return OWNER_TYPES.get(Dict('marche')(self.doc), NotAvailable)
def get_connection_id(self):
connection_id = Regexp(
CleanText('//script[contains(text(), "NPC.utilisateur.ccptea")]'),
r"NPC.utilisateur.ccptea = '(\d+)';"
)(self.html_doc)
return connection_id
@method
class get_main_account(ItemElement):
klass = Account
......
# -*- coding: utf-8 -*-
# Copyright(C) 2019 Sylvie Ye
#
# 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 Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from datetime import date
from weboob.browser.pages import LoggedPage, JsonPage, RawPage
from weboob.browser.elements import method, ItemElement, DictElement
from weboob.capabilities.bank import (
Account, Recipient, Transfer, TransferBankError,
)
from weboob.browser.filters.standard import (
CleanDecimal, Env, Date, CleanText,
)
from weboob.browser.filters.json import Dict
class RecipientsPage(LoggedPage, JsonPage):
def is_sender_account(self, account_id):
for acc in self.doc:
if acc.get('senderOfTransfert') and account_id == acc.get('accountNumber'):
return True
@method
class iter_debit_accounts(DictElement):
class item(ItemElement):
def condition(self):
return Dict('accountNumber', default=None)(self)
klass = Account
obj_id = obj_number = Dict('accountNumber')
obj_label = Dict('accountNatureLongLabel')
obj_iban = Dict('ibanCode')
obj_currency = Dict('currencyCode')
def obj_balance(self):
balance_value = CleanDecimal(Dict('balanceValue'))(self)
if CleanText(Dict('balanceSign'))(self) == '-':
return -balance_value
return balance_value
@method
class iter_internal_recipient(DictElement):
class item(ItemElement):
def condition(self):
return Dict('recipientOfTransfert', default=None)(self) and \
Env('account_id')(self) != Dict('accountNumber', default=None)(self)
klass = Recipient
obj_id = Dict('accountNumber')
obj_label = Dict('accountNatureLongLabel')
obj_iban = Dict('ibanCode')
obj_category = 'Interne'
obj_enabled_at = date.today()
@method
class iter_external_recipient(DictElement):
class item(ItemElement):
def condition(self):
return Dict('recipientId', default=None)(self)
klass = Recipient
obj_id = obj_iban = Dict('ibanCode')
obj_label = Dict('recipientName')
obj_category = 'Externe'
obj_enabled_at = date.today()
class TransferTokenPage(LoggedPage, RawPage):
def get_token(self):
return self.doc
class TransferPage(LoggedPage, JsonPage):
def check_transfer(self):
error_msg = Dict('messageErreur')(self.doc)
if error_msg:
raise TransferBankError(message=error_msg)
return Dict('page')(self.doc) == '/recap'
def handle_response(self, transfer):
t = Transfer()
t._space = transfer._space
t._operation = transfer._operation
t._token = transfer._token
t._connection_id = transfer._connection_id
t.label = Dict('transferComplementaryInformations1')(self.doc)
t.exec_date = Date(Dict('dateVirement'), dayfirst=True)(self.doc)
t.amount = CleanDecimal(Dict('amount'))(self.doc)
t.currency = Dict('currencyCode')(self.doc)
t.account_id = Dict('currentDebitAccountNumber')(self.doc)
t.account_iban = Dict('currentDebitIbanCode')(self.doc)
t.account_label = Dict('typeCompte')(self.doc)
t.recipient_id = t.recipient_iban = Dict('currentCreditIbanCode')(self.doc)
t.recipient_label = Dict('currentCreditAccountName')(self.doc)
return t
def check_transfer_exec(self):
error_msg = Dict('messageErreur')(self.doc)
if error_msg:
raise TransferBankError(message=error_msg)
return Dict('page')(self.doc)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment