Newer
Older
# 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 <http://www.gnu.org/licenses/>.
from datetime import datetime
from dateutil.relativedelta import relativedelta
from woob.browser import URL, need_login
from woob.browser.mfa import TwoFactorBrowser
from woob.capabilities.bill import Document, DocumentTypes
from woob.exceptions import (
BrowserIncorrectPassword, ActionNeeded, ActionType, BrowserUnavailable,
AppValidation, BrowserQuestion, AppValidationError, AppValidationCancelled,
AppValidationExpired, BrowserPasswordExpired, BrowserUserBanned,
from woob.capabilities.bank import (
Account, TransferBankError, AddRecipientStep,
TransactionType, AccountOwnerType, Loan,
)
from woob.capabilities.base import NotAvailable
from woob.browser.exceptions import BrowserHTTPNotFound, ClientError
from woob.capabilities.profile import ProfileMissing
from woob.tools.value import Value, ValueBool
from woob.tools.decorators import retry
from .pages.accounts_list import (
AccountsMainPage, AccountDetailsPage, AccountsPage, LoansPage, HistoryPage,
CardHistoryPage, PeaLiquidityPage, MarketOrderPage, MarketOrderDetailPage,
AdvisorPage, HTMLProfilePage, CreditPage, CreditHistoryPage, OldHistoryPage,
MarketPage, LifeInsurance, LifeInsuranceHistory, LifeInsuranceInvest, LifeInsuranceInvest2,
LifeInsuranceAPI, LifeInsuranceInvestAPI, LifeInsuranceInvestAPI2, LifeInsuranceInvestDetailsAPI,
UnavailableServicePage, TemporaryBrowserUnavailable, RevolvingDetailsPage,
from .pages.transfer import AddRecipientPage, SignRecipientPage, TransferJson, SignTransferPage
from .pages.login import (
MainPage, LoginPage, BadLoginPage, ReinitPasswordPage,
ActionNeededPage, ErrorPage, VkImage, SkippableActionNeededPage,
)
from .pages.subscription import DocumentsPage, RibPdfPage
__all__ = ['SocieteGenerale']
class SocieteGeneraleTwoFactorBrowser(TwoFactorBrowser):
HAS_CREDENTIALS_ONLY = True
polling_transaction = None
polling_duration = 300 # default to 5 minutes
__states__ = ('polling_transaction',)
skippable_action_needed_page = URL(
r'/icd/gax/data/users/administration/out-of-remedy-security-security-zone.json',
SkippableActionNeededPage
)
def __init__(self, config, *args, **kwargs):
super(SocieteGeneraleTwoFactorBrowser, self).__init__(config, *args, **kwargs)
self.AUTHENTICATION_METHODS = {
'resume': self.handle_polling,
'code': self.handle_sms,
}
def load_state(self, state):
if state.get('polling_transaction'):
# can't start in the middle of a AppValidation process
# or we will launch another one with that URL
state.pop('url', None)
super(SocieteGeneraleTwoFactorBrowser, self).load_state(state)
def check_password(self):
if not self.password.isdigit() or len(self.password) not in (6, 7):
if not self.username.isdigit() or len(self.username) < 8:
def check_login_reason(self):
reason = self.page.get_reason()
if reason is not None:
# 'reason' doesn't bear a user-friendly message.
# The messages related to each 'reason' can be found in 'swm.main.js'
if reason == 'echec_authent':
raise BrowserIncorrectPassword()
elif reason == 'mdptmp_expire':
raise BrowserPasswordExpired()
elif reason == 'acces_bloq':
raise BrowserUserBanned(
"Suite à trois saisies erronées de vos codes, l'accès à vos comptes est bloqué jusqu'à demain pour des raisons de sécurité."
)
elif reason in ('acces_susp', 'pas_acces_bad'):
# These codes are not related to any valuable messages,
# just "Votre accès est suspendu. Vous n'êtes pas autorisé à accéder à l'application."
raise BrowserUserBanned()
elif reason in ('err_is', 'err_tech'):
# there is message "Service momentanément indisponible. Veuillez réessayer."
# in SG website in that case ...
raise BrowserUnavailable()
raise AssertionError('Unhandled error reason: %s' % reason)
def check_auth_method(self):
auth_method = self.page.get_auth_method()
if not auth_method:
self.logger.warning('No auth method available !')
raise ActionNeeded(
locale="fr-FR", message="Veuillez ajouter un numéro de téléphone sur votre banque et/ou activer votre Pass Sécurité.",
action_type=ActionType.ENABLE_MFA,
)
if auth_method['unavailability_reason'] == "ts_non_enrole":
raise ActionNeeded(
locale="fr-FR", message="Veuillez ajouter un numéro de téléphone sur votre banque.",
action_type=ActionType.ENABLE_MFA,
)
elif auth_method['unavailability_reason']:
raise AssertionError('Unknown unavailability reason "%s" found' % auth_method['unavailability_reason'])
if auth_method['type_proc'].lower() == 'auth_oob':
# notification is sent here
self.location('/sec/oob_sendooba.json', method='POST', headers={'Content-Type': 'application/x-www-form-urlencoded'})
donnees = self.page.doc['donnees']
self.polling_transaction = donnees['id-transaction']
if donnees.get('expiration_date_hh') and donnees.get('expiration_date_mm'):
now = datetime.now()
expiration_date = now.replace(
hour=int(donnees['expiration_date_hh']),
minute=int(donnees['expiration_date_mm'])
)
self.polling_duration = int((expiration_date - now).total_seconds())
message = "Veuillez valider l'opération dans votre application"
# several terminals can be associated with that user
terminals = [terminal['nom'] for terminal in auth_method['terminal'] if terminal.get('nom')]
if terminals:
message += " sur l'un de vos périphériques actifs: " + ', '.join(terminals)
raise AppValidation(message)
elif auth_method['type_proc'].lower() == 'auth_csa':
if auth_method['mode'] == "SMS":
# SMS is sent here
)
raise BrowserQuestion(
Value(
label='Entrez le Code Sécurité reçu par SMS sur le numéro ' + auth_method['ts']
)
)
self.logger.warning('Unknown CSA method "%s" found', auth_method['mod'])
else:
self.logger.warning('Unknown sign method "%s" found', auth_method['type_proc'])
raise AssertionError('Unknown auth method "%s: %s" found' % (auth_method['type_proc'], auth_method.get('mod')))
def check_skippable_action_needed(self):
if not self.login.is_here():
return
reason = self.page.get_skippable_action_needed()
if reason == 'FIABILISATION_TS':
self.skippable_action_needed_page.go(
headers={'Content-Type': 'application/json;charset=UTF-8'},
data='',
)
# Sometimes it is not possible to skip this step without SCA
if self.page.has_twofactor():
self.check_interactive()
self.check_auth_method()
def init_login(self):
self.check_password()
self.main_page.go()
try:
self.page.login(self.username[:8], self.password)
except BrowserHTTPNotFound:
raise BrowserIncorrectPassword()
assert self.login.is_here(), "An error has occurred, we should be on login page."
self.check_login_reason()
if self.page.has_twofactor():
self.check_interactive()
self.check_auth_method()
self.check_skippable_action_needed()
def check_polling_errors(self, status):
if status == "rejected":
raise AppValidationCancelled(
"L'opération dans votre application a été annulée"
)
if status == "aborted":
raise AppValidationExpired(
"L'opération dans votre application a expiré"
)
if status != "available":
raise AppValidationError()
def handle_polling(self):
assert self.polling_transaction, "polling_transaction is mandatory !"
data = {'n10_id_transaction': self.polling_transaction}
timeout = time.time() + self.polling_duration
while time.time() < timeout:
self.location('/sec/oob_pollingooba.json', data=data)
status = self.page.doc['donnees']['transaction_status']
if status != "in_progress":
break
time.sleep(3)
else:
status = "aborted"
self.check_polling_errors(status)
self.location('/sec/oob_auth.json', data=data)
if self.page.doc.get('commun', {}).get('statut').lower() == "nok":
raise BrowserUnavailable()
self.polling_transaction = None
self.check_skippable_action_needed()
# Need to end up on a LoggedPage to avoid starting back at login
# Might be caused by multiple @need_login call
self.accounts.go()
def handle_sms(self):
if len(self.code) != 6:
raise BrowserIncorrectPassword(
'Le Code Sécurité doit avoir une taille de 6 caractères'
)
data = {
'code': self.code,
'csa_op': "auth",
}
self.location('/sec/csa/check.json', data=data)
if self.page.doc.get('commun', {}).get('statut').lower() == "nok":
raise BrowserIncorrectPassword('Le Code Sécurité est invalide')
self.check_skippable_action_needed()
class SocieteGenerale(SocieteGeneraleTwoFactorBrowser):
BASEURL = 'https://particuliers.sg.fr'
STATE_DURATION = 10
# documents
documents = URL(r'/icd/cbo-edocument/data/get-all-prestations-edocument-authsec.json', DocumentsPage)
pdf_page = URL(
r'/icd/cbo-edocument/pdf/rce-authsec.pdf\?b64e200_prestationIdTechnique=(?P<id_tech>.*)&b64e200_refTechnique=(?P<ref_tech>.*)'
)
rib_pdf_page = URL(r'/com/icd-web/cbo/pdf/rib-authsec.pdf', RibPdfPage)
# Bank
accounts_main_page = URL(
r'/restitution/cns_listeprestation.html',
r'/com/icd-web/cbo/index.html',
r'/icd/cbo/index-authsec.html',
AccountsMainPage
)
account_details_page = URL(r'/restitution/cns_detailPrestation.html', AccountDetailsPage)
accounts = URL(r'/icd/cbo/data/liste-prestations-authsec.json\?n10_avecMontant=1', AccountsPage)
history = URL(r'/icd/cbo/data/liste-operations-authsec.json', HistoryPage)
loans = URL(r'/icd/espaces-thematiques/data/getLoansRecovery.json', LoansPage)
revolving_rate = URL(r'icd/cbo/data/recapitulatif-prestation-authsec.json', RevolvingDetailsPage)
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
card_history = URL(r'/restitution/cns_listeReleveCarteDd.xml', CardHistoryPage)
credit = URL(r'/restitution/cns_detailAVPAT.html', CreditPage)
credit_history = URL(r'/restitution/cns_listeEcrCav.xml', CreditHistoryPage)
old_hist_page = URL(
r'/restitution/cns_detailPep.html',
r'/restitution/cns_listeEcrPep.html',
r'/restitution/cns_detailAlterna.html',
r'/restitution/cns_listeEncoursAlterna.html',
OldHistoryPage
)
# Recipient
add_recipient = URL(
r'/personnalisation/per_cptBen_ajouterFrBic.html',
r'/lgn/url.html',
AddRecipientPage
)
json_recipient = URL(
r'/sec/getsigninfo.json',
r'/sec/csa/send.json',
r'/sec/oob_sendoob.json',
r'/sec/oob_polling.json',
SignRecipientPage
)
# Transfer
json_transfer = URL(
r'/icd/vupri/data/vupri-liste-comptes.json\?an200_isBack=false',
r'/icd/vupri/data/vupri-check.json',
TransferJson
)
sign_transfer = URL(r'/icd/vupri/data/vupri-generate-token.json', SignTransferPage)
confirm_transfer = URL(r'/icd/vupri/data/vupri-save.json', TransferJson)
# Wealth
market = URL(r'/brs/cct/comti20.html', MarketPage)
pea_liquidity = URL(r'/restitution/cns_detailPea.html', PeaLiquidityPage)
life_insurance = URL(
r'/asv/asvcns10.html',
r'/asv/AVI/asvcns10a.html',
r'/brs/fisc/fisca10a.html',
LifeInsurance
)
life_insurance_invest = URL(r'/asv/AVI/asvcns20a.html', LifeInsuranceInvest)
life_insurance_invest_2 = URL(r'/asv/PRV/asvcns10priv.html', LifeInsuranceInvest2)
auth_life_insurance_api = URL(r'/icd/avd/index-authsec.html')
life_insurance_api = URL(
r'/icd/avd/data/api/v1/prestation-assurance-vie-authsec.json\?b64e200_hashIdPrestation=(?P<id_tech>.*)',
LifeInsuranceAPI
)
life_insurance_invest_api = URL(
r'/icd/avd/data/api/v1/detail-contrat-assurance-vie-authsec.json',
LifeInsuranceInvestAPI
)
life_insurance_invest_api_2 = URL(
r'/icd/avd/data/api/v1/contrat-assurance-vie-authsec.json',
LifeInsuranceInvestAPI2
)
life_insurance_invest_details_api = URL(
r'/icd/avd/data/api/v1/performances-authsec.json\?b64e200_hashIdPrestation=(?P<id_tech>.*)',
LifeInsuranceInvestDetailsAPI
)
life_insurance_history = URL(r'/asv/AVI/asvcns2(?P<n>[0-9])c.html', LifeInsuranceHistory)
market_orders = URL(r'/brs/suo/suivor20.html', MarketOrderPage)
market_orders_details = URL(r'/brs/suo/suivor30.html', MarketOrderDetailPage)
# Profile
advisor = URL(r'/icd/pon/data/get-contacts.xml', AdvisorPage)
html_profile_page = URL(r'/com/dcr-web/dcr/dcr-coordonnees.html', HTMLProfilePage)
bad_login = URL(r'/acces/authlgn.html', r'/error403.html', BadLoginPage)
reinit = URL(
r'/acces/changecodeobligatoire.html',
r'/swm/swm-changemdpobligatoire.html',
ReinitPasswordPage
)
action_needed = URL(
r'/com/icd-web/forms/cct-index.html',
r'/com/icd-web/gdpr/gdpr-recueil-consentements.html',
r'/com/icd-web/forms/kyc-index.html',
ActionNeededPage
)
unavailable_service_page = URL(
r'/com/service-indisponible.html',
r'.*/Technical-pages/503-error-page/unavailable.html',
r'.*/Technical-pages/service-indisponible/service-indisponible.html',
r'/fonction-indisponible',
UnavailableServicePage
)
error = URL(
r'https://static.sg.fr/pri/erreur.html',
r'https://.*/pri/erreur.html',
ErrorPage
)
login = URL(
r'https://particuliers.sg.fr//sec/vk/', # yes, it works only with double slash
r'/sec/oob_sendooba.json',
r'/sec/oob_pollingooba.json',
r'/sec/oob_auth.json',
r'/sec/csa/check.json',
LoginPage
)
vk_image = URL(r'/?/sec/vkm/gen_ui', VkImage)
main_page = URL(r'https://particuliers.sg.fr', MainPage)
context = None
dup = None
id_transaction = None
def __init__(self, config, *args, **kwargs):
super(SocieteGenerale, self).__init__(config, *args, **kwargs)
self.__states__ += ('context', 'dup', 'id_transaction',)
def transfer_condition(self, state):
return state.get('dup') is not None and state.get('context') is not None
def locate_browser(self, state):
if self.transfer_condition(state):
self.location('/com/icd-web/cbo/index.html')
elif all(url in state['url'] for url in self.login.urls):
return
elif self.json_recipient.match(state['url']):
return
super(SocieteGenerale, self).locate_browser(state)
def iter_cards(self, account):
for el in account._cards:
if el['carteDebitDiffere']:
card = Account()
card.id = el['id']
card.number = el['numeroCompteFormate'].replace(' ', '')
card.label = el['labelToDisplay']
card.coming = Decimal(str(el['montantProchaineEcheance']))
card.type = Account.TYPE_CARD
card.currency = account.currency
card._internal_id = el['idTechnique']
card._prestation_id = el['id']
card.owner_type = AccountOwnerType.PRIVATE
def switch_account_to_loan(self, account):
loan = Loan()
copy_attrs = (
'id', 'number', 'label', 'type', 'ownership', 'owner_type',
'coming', '_internal_id', '_prestation_id', '_loan_type',
'_is_json_histo',
)
for attr in copy_attrs:
setattr(loan, attr, getattr(account, attr))
return loan
self.accounts_main_page.go()
self.page.is_accounts()
if self.page.is_old_website():
# go on new_website
self.location(self.absurl('/com/icd-web/cbo/index.html'))
go = retry(TemporaryBrowserUnavailable)(self.accounts.go)
go()
if not self.page.is_new_website_available():
# return in old pages to get accounts
self.accounts_main_page.go(params={'NoRedirect': True})
for acc in self.page.iter_accounts():
yield acc
return
accounts = {}
for account in self.page.iter_accounts():
account._loan_parent_id = None
account.owner_type = AccountOwnerType.PRIVATE
for card in self.iter_cards(account):
card.parent = account
card.ownership = account.ownership
card.owner_type = AccountOwnerType.PRIVATE
if account.type in (
account.TYPE_LOAN,
account.TYPE_CONSUMER_CREDIT,
account.TYPE_REVOLVING_CREDIT,
account.TYPE_MORTGAGE,
):
loan = self.switch_account_to_loan(account)
self.loans.stay_or_go()
self.page.get_loan_details(loan)
# The revolving rate is missing on this page.
# We have to go to the revolving details page for each revolving.
if loan.type == account.TYPE_REVOLVING_CREDIT:
self.revolving_rate.go(params={'b64e200_prestationIdTechnique': account._internal_id})
self.page.get_revolving_rate(loan)
# Adding parent account to LOAN account
if account._loan_parent_id:
account.parent = accounts.get(account._loan_parent_id, NotAvailable)
def fill_loan_insurance(self, loan):
if not loan.parent:
self.logger.info('Loan: %s has no parent account. Could not find insurance amount', loan)
return
for transaction in self.iter_history(loan.parent):
insurance_amount = transaction._insurance_amount
insurance_loan_id = transaction._insurance_loan_id
if insurance_loan_id and insurance_loan_id in loan.id:
if insurance_amount:
loan.insurance_amount = insurance_amount
break
else:
self.logger.info(
'A transaction related to the loan %s was found, but has no amount. transaction raw: %s',
loan,
transaction.raw,
)
break
else:
self.logger.info('No transaction related to the loan %s was found.', loan)
def next_page_retry(self, condition):
next_page = self.page.hist_pagination(condition)
if next_page:
location = retry(TemporaryBrowserUnavailable)(self.location)
location(next_page)
return True
return False
if account.type in (
account.TYPE_LOAN,
account.TYPE_MARKET,
account.TYPE_CONSUMER_CREDIT,
account.TYPE_MORTGAGE,
):
if account.type == Account.TYPE_PEA and not ('Espèces' in account.label or 'ESPECE' in account.label):
return
if not account._internal_id:
raise BrowserUnavailable()
# get history for account on old website
# request to get json is not available yet, old request to get html response
if any((
account.type in (account.TYPE_LIFE_INSURANCE, account.TYPE_PERP, account.TYPE_PER),
account.type == account.TYPE_REVOLVING_CREDIT and account._loan_type != 'PR_CONSO',
account.type in (account.TYPE_REVOLVING_CREDIT, account.TYPE_SAVINGS) and not account._is_json_histo,
go = retry(TemporaryBrowserUnavailable)(self.account_details_page.go)
go(params={'idprest': account._prestation_id})
if self.unavailable_service_page.is_here():
raise BrowserUnavailable()
history_url = self.page.get_history_url()
# history_url return NotAvailable when history page doesn't exist
# it return None when we don't know if history page exist
if history_url is None:
error_msg = self.page.get_error_msg()
assert error_msg, 'There should have error or history url'
raise BrowserUnavailable(error_msg)
elif history_url:
self.location(self.absurl(history_url))
for tr in self.page.iter_history():
if account.type == account.TYPE_CARD:
go = retry(TemporaryBrowserUnavailable)(self.history.go)
go(params={'b64e200_prestationIdTechnique': account.parent._internal_id})
next_page = True
while next_page:
for summary_card_tr in self.page.iter_card_transactions(card_number=account.number):
yield summary_card_tr
for card_tr in summary_card_tr._card_transactions:
card_tr.date = summary_card_tr.date
# We use the Raw pattern to set the rdate automatically, but that make
# the transaction type to "CARD", so we have to correct it in the browser.
card_tr.type = TransactionType.DEFERRED_CARD
yield card_tr
next_page = self.next_page_retry('history')
go = retry(TemporaryBrowserUnavailable)(self.history.go)
go(params={'b64e200_prestationIdTechnique': account._internal_id})
next_page = True
while next_page:
for transaction in self.page.iter_history():
yield transaction
next_page = self.next_page_retry('history')
@need_login
def iter_coming(self, account):
skipped_types = (
Account.TYPE_LOAN,
Account.TYPE_MARKET,
Account.TYPE_PEA,
Account.TYPE_LIFE_INSURANCE,
Account.TYPE_REVOLVING_CREDIT,
Account.TYPE_CONSUMER_CREDIT,
Account.TYPE_PERP,
)
if account.type in skipped_types:
if not account._internal_id:
raise BrowserUnavailable()
if account.type == account.TYPE_SAVINGS and not account._is_json_histo:
# Waiting for account with transactions
return
internal_id = account._internal_id
if account.type == account.TYPE_CARD:
internal_id = account.parent._internal_id
go = retry(TemporaryBrowserUnavailable)(self.history.go)
go(params={'b64e200_prestationIdTechnique': internal_id})
if account.type == account.TYPE_CARD:
next_page = True
while next_page:
for transaction in self.page.iter_future_transactions(acc_prestation_id=account._prestation_id):
# coming transactions on this page are not included in coming balance
# use it only to retrive deferred card coming transactions
if transaction._card_coming:
for card_coming in transaction._card_coming:
card_coming.date = transaction.date
# We use the Raw pattern to set the rdate automatically, but that makes
# the transaction type to "CARD", so we have to correct it in the browser.
card_coming.type = TransactionType.DEFERRED_CARD
yield card_coming
next_page = self.next_page_retry('future')
next_page = True
while next_page:
for intraday_tr in self.page.iter_intraday_comings():
yield intraday_tr
next_page = self.next_page_retry('intraday')
def iter_investment(self, account):
if account.type not in (
Christophe François
committed
Account.TYPE_MARKET, Account.TYPE_LIFE_INSURANCE,
Account.TYPE_PEA, Account.TYPE_PERP, Account.TYPE_PER,
self.logger.debug('This account is not supported')
# request to get json is not available yet, old request to get html response
self.account_details_page.go(params={'idprest': account._prestation_id})
if account.type in (Account.TYPE_PEA, Account.TYPE_MARKET):
for invest in self.page.iter_investments(account=account):
yield invest
Quentin Defenouillere
committed
if account.type in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_PERP, Account.TYPE_PER):
self.auth_life_insurance_api.go()
self.life_insurance_api.go(id_tech=account._internal_id)
# Case 1: Life Insurance investment are available on the API.
if self.page.check_availability():
self.life_insurance_invest_api.go()
# Case 2: We need to query a different life insurance space API.
if not self.page.check_availability():
self.life_insurance_invest_api_2.go()
investments = self.page.iter_investment()
self.life_insurance_invest_details_api.go(id_tech=account._internal_id)
for inv in investments:
self.page.fill_life_insurance_investment(obj=inv)
yield inv
return
# Case 3: Life insurance investments can be parsed on the website.
else:
self.account_details_page.go(params={'idprest': account._prestation_id})
if self.page.has_link():
self.life_insurance_invest.go()
for invest in self.page.iter_investment():
yield invest
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
@need_login
def access_market_orders(self, account):
account_dropdown_id = self.page.get_dropdown_menu()
link = self.page.get_market_order_link()
if not link:
self.logger.warning('Could not find Market Order link for account %s.', account.label)
return
self.location(link)
# Once we reached the Market Orders page, we must select the right market account:
params = {
'action': '10',
'numPage': '1',
'idCptSelect': account_dropdown_id,
}
self.market_orders.go(params=params)
@need_login
def iter_market_orders(self, account):
if account.type not in (Account.TYPE_MARKET, Account.TYPE_PEA):
return
# Market Orders page sometimes bugs so we try accessing them twice
for trial in range(2):
self.account_details_page.go(params={'idprest': account._prestation_id})
if self.pea_liquidity.is_here():
self.logger.debug('Liquidity PEA have no market orders')
return
self.access_market_orders(account)
if not self.market_orders.is_here():
self.logger.warning(
'Landed on unknown page when trying to fetch market orders for account %s',
account.label
)
return
if self.page.orders_unavailable():
if trial == 0:
self.logger.warning(
'Market Orders page is unavailable for account %s, retrying now.',
account.label
)
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
continue
self.logger.warning('Market Orders are unavailable for account %s.', account.label)
return
if self.page.has_no_market_order():
self.logger.debug('Account %s has no market orders.', account.label)
return
# Handle pagination
total_pages = self.page.get_pages()
account_dropdown_id = self.page.get_dropdown_menu()
for page in range(1, total_pages + 1):
if page > 1:
# Select the right page
params = {
'action': '12',
'numPage': page,
'idCptSelect': account_dropdown_id,
}
self.market_orders.go(params=params)
for order in self.page.iter_market_orders():
if order.url:
self.location(order.url)
if self.market_orders_details.is_here():
self.page.fill_market_order(obj=order)
else:
self.logger.warning('Landed on unknown Market Order detail page for order %s', order.label)
yield order
def iter_recipients(self, account, ignore_errors=True):
self.json_transfer.go()
if ignore_errors:
return []
raise
if not self.page.is_able_to_transfer(account):
return []
return self.page.iter_recipients(account_id=account.id)
@need_login
def init_transfer(self, account, recipient, transfer):
self.json_transfer.go()
Sylvie Ye
committed
first_transfer_date = self.page.get_first_available_transfer_date()
if transfer.exec_date and transfer.exec_date < first_transfer_date:
transfer.exec_date = first_transfer_date
self.page.init_transfer(account, recipient, transfer)
return self.page.handle_response(recipient)
@need_login
def execute_transfer(self, transfer):
assert transfer.id, 'Transfer token is missing'
data = {
}
# get token and virtual keyboard
self.sign_transfer.go(params=data)
data.update(self.page.get_confirm_transfer_data(self.password))
# execute transfer
headers = {'Referer': self.absurl('/com/icd-web/vupri/virement.html')}
self.confirm_transfer.go(data=data, headers=headers)
assert self.page.is_transfer_validated(), 'Something went wrong, transfer is not executed'
# return on main page to avoid reload on transfer confirmation page
self.accounts_main_page.go()
def end_sms_recipient(self, recipient, **params):
"""End adding recipient with OTP SMS authentication"""
data = [
('context', [self.context, self.context]),
('dup', self.dup),
('code', params['code']),
]
# needed to confirm recipient validation
add_recipient_url = self.absurl('/lgn/url.html', base=True)
self.location(add_recipient_url, data=data, headers={'Referer': add_recipient_url})
return self.page.get_recipient_object(recipient)
def end_oob_recipient(self, recipient, **params):
"""End adding recipient with 'pass sécurité' authentication"""
r = self.open(
self.absurl('/sec/oob_polling.json'),
data={'n10_id_transaction': self.id_transaction}
)
assert self.id_transaction, "Transaction id is missing, can't sign new recipient."
r.page.check_recipient_status()
data = [
('context', self.context),
('b64_jeton_transaction', self.context),
('dup', self.dup),
('n10_id_transaction', self.id_transaction),
# needed to confirm recipient validation
add_recipient_url = self.absurl('/lgn/url.html', base=True)
self.location(add_recipient_url, data=data, headers={'Referer': add_recipient_url})
return self.page.get_recipient_object(recipient)
def send_sms_to_user(self, recipient):
"""Add recipient with OTP SMS authentication"""
data = {}
data['csa_op'] = 'sign'
data['context'] = self.context
self.open(self.absurl('/sec/csa/send.json'), data=data)
raise AddRecipientStep(
recipient,
Value('code', label='Cette opération doit être validée par un Code Sécurité.')
)
def send_notif_to_user(self, recipient):
"""Add recipient with 'pass sécurité' authentication"""
data = {}
data['b64_jeton_transaction'] = self.context
r = self.open(self.absurl('/sec/oob_sendoob.json'), data=data)
self.id_transaction = r.page.get_transaction_id()
raise AddRecipientStep(recipient, ValueBool('pass', label='Valider cette opération sur votre applicaton société générale'))
@retry(BrowserUnavailable)
def get_sign_method(self, data):
r = self.open(self.absurl('/sec/getsigninfo.json'), data=data)
return r.page.get_sign_method()
@need_login
def new_recipient(self, recipient, **params):
if 'code' in params:
return self.end_sms_recipient(recipient, **params)
if 'pass' in params:
return self.end_oob_recipient(recipient, **params)
self.add_recipient.go()
if self.main_page.is_here():
self.page.handle_error()
raise AssertionError('Should not be on this page.')
self.page.post_iban(recipient)
self.page.post_label(recipient)
recipient = self.page.get_recipient_object(recipient, get_info=True)
self.page.update_browser_recipient_state()
data = self.page.get_signinfo_data()
sign_method = self.get_sign_method(data)
# WARNING: this send validation request to user
if sign_method == 'CSA':
return self.send_sms_to_user(recipient)
elif sign_method == 'OOB':
return self.send_notif_to_user(recipient)
raise AssertionError('Sign process unknown: %s' % sign_method)
@need_login
def get_advisor(self):
return self.advisor.go().get_advisor()
@need_login
def get_profile(self):
self.html_profile_page.go()
return self.page.get_profile()
@need_login
def iter_subscription(self):
self.accounts_main_page.go()
try:
profile = self.get_profile()
subscriber = profile.name
Martin Morlot
committed
except (ProfileMissing, BrowserUnavailable):
subscriber = NotAvailable
self.accounts.go()
return self.page.iter_subscription(subscriber=subscriber)
def _fetch_rib_document(self, subscription):
d = Document()
d.id = subscription.id + '_RIB'
d.url = self.rib_pdf_page.build(params={'b64e200_prestationIdTechnique': subscription._internal_id})
d.type = DocumentTypes.RIB
d.format = 'pdf'
d.label = 'RIB'
return d
def _iter_statements(self, subscription):
# we need _rad_button_id for post_form function
# if not present it means this subscription doesn't have any bank statement
end_date = datetime.today()
begin_date = (end_date - relativedelta(months=+2)).replace(day=1)
empty_page = 0
stop_after_empty_limit = 4
for _ in range(60):
is_empty = True
params = {
'b64e200_prestationIdTechnique': subscription._internal_id,
'dt10_dateDebut': begin_date.strftime('%d/%m/%Y'),
'dt10_dateFin': end_date.strftime('%d/%m/%Y'),
}
self.documents.go(params=params)
for d in self.page.iter_documents(subid=subscription.id):
is_empty = False
yield d
self.logger.debug('no documents on %s', end_date)
if empty_page >= stop_after_empty_limit:
# No more documents
return
end_date = begin_date - relativedelta(day=1)
begin_date = end_date - relativedelta(months=3)
@need_login
def iter_documents(self, subscription):
yield self._fetch_rib_document(subscription)
for doc in self._iter_statements(subscription):
yield doc
@need_login
def iter_documents_by_types(self, subscription, accepted_types):
if DocumentTypes.RIB in accepted_types:
yield self._fetch_rib_document(subscription)
if DocumentTypes.STATEMENT not in accepted_types:
return
for doc in self._iter_statements(subscription):
yield doc