browser.py

# -*- coding: utf-8 -*-

# Copyright(C) 2010-2011 Jocelyn Jaubert
#
# 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 <http://www.gnu.org/licenses/>.

from __future__ import unicode_literals

from datetime import datetime
from decimal import Decimal
from dateutil.relativedelta import relativedelta

from weboob.browser import LoginBrowser, URL, need_login, StatesMixin
from weboob.exceptions import BrowserIncorrectPassword, ActionNeeded, BrowserUnavailable
from weboob.capabilities.bank import Account, TransferBankError, AddRecipientStep
from weboob.capabilities.base import find_object, NotAvailable
from weboob.browser.exceptions import BrowserHTTPNotFound, ClientError
from weboob.capabilities.profile import ProfileMissing
from weboob.tools.value import Value, ValueBool

from .pages.accounts_list import (
    AccountsMainPage, AccountDetailsPage, AccountsPage, LoansPage, HistoryPage,
    CardHistoryPage, PeaLiquidityPage, AccountsSynthesesPage,
    AdvisorPage, HTMLProfilePage, XMLProfilePage, CreditPage, CreditHistoryPage,
    MarketPage, LifeInsurance, LifeInsuranceHistory, LifeInsuranceInvest, LifeInsuranceInvest2,
    UnavailableServicePage,
)
from .pages.transfer import AddRecipientPage, SignRecipientPage, TransferJson, SignTransferPage
from .pages.login import MainPage, LoginPage, BadLoginPage, ReinitPasswordPage, ActionNeededPage, ErrorPage
from .pages.subscription import BankStatementPage


__all__ = ['SocieteGenerale']


class SocieteGenerale(LoginBrowser, StatesMixin):
    BASEURL = 'https://particuliers.societegenerale.fr'
    STATE_DURATION = 5

    # Bank
    accounts_main_page = URL(r'/restitution/cns_listeprestation.html',
                             r'/com/icd-web/cbo/index.html', AccountsMainPage)
    account_details_page = URL(r'/restitution/cns_detailPrestation.html', AccountDetailsPage)
    accounts = URL(r'/icd/cbo/data/liste-prestations-navigation-authsec.json', AccountsPage)
    accounts_syntheses = URL(r'/icd/cbo/data/liste-prestations-authsec.json\?n10_avecMontant=1', AccountsSynthesesPage)
    history = URL(r'/icd/cbo/data/liste-operations-authsec.json', HistoryPage)
    card_history = URL(r'/restitution/cns_listeReleveCarteDd.xml', CardHistoryPage)
    loans = URL(r'/abm/restit/listeRestitutionPretsNET.json\?a100_isPretConso=(?P<conso>\w+)', LoansPage)
    credit = URL(r'/restitution/cns_detailAVPAT.html', CreditPage)
    credit_history = URL(r'/restitution/cns_listeEcrCav.xml', CreditHistoryPage)

    # 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',
                        r'/lgn/url.html', 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)
    life_insurance_history = URL(r'/asv/AVI/asvcns2(?P<n>[0-9])c.html', LifeInsuranceHistory)

    # Profile
    advisor = URL(r'/icd/pon/data/get-contacts.xml', AdvisorPage)
    html_profile_page = URL(r'/com/dcr-web/dcr/dcr-coordonnees.html', HTMLProfilePage)
    xml_profile_page = URL(r'/gms/gmsRestituerAdresseNotificationServlet.xml', XMLProfilePage)

    # Document
    bank_statement = URL(r'/restitution/rce_derniers_releves.html', BankStatementPage)
    bank_statement_search = URL(r'/restitution/rce_recherche.html\?noRedirect=1',
                                r'/restitution/rce_recherche_resultat.html', BankStatementPage)

    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', UnavailableServicePage)
    error = URL(r'https://static.societegenerale.fr/pri/erreur.html', ErrorPage)
    login = URL(r'/sec/vk', LoginPage)
    main_page = URL(r'https://particuliers.societegenerale.fr', MainPage)

    context = None
    dup = None
    id_transaction = None

    __states__ = ('context', 'dup', 'id_transaction')

    def locate_browser(self, state):
        self.location('/com/icd-web/cbo/index.html')

    def load_state(self, state):
        if state.get('dup') is not None and state.get('context') is not None:
            super(SocieteGenerale, self).load_state(state)

    def do_login(self):
        if not self.password.isdigit() or len(self.password) not in (6, 7):
            raise BrowserIncorrectPassword()
        if not self.username.isdigit() or len(self.username) < 8:
            raise BrowserIncorrectPassword()
        self.username = self.username[:8]

        self.main_page.go()
        try:
            self.page.login(self.username, self.password)
        except BrowserHTTPNotFound:
            raise BrowserIncorrectPassword()

        assert self.login.is_here()
        reason, action = self.page.get_error()
        if reason == 'echec_authent':
            raise BrowserIncorrectPassword()
        elif reason in ('acces_bloq', 'acces_susp', 'pas_acces_bad', ):
            raise ActionNeeded()
        elif reason == 'err_tech':
            # there is message "Service momentanément indisponible. Veuillez réessayer."
            # in SG website in that case ...
            raise BrowserUnavailable()

    def iter_cards(self, account):
        for el in account._cards:
            if el['carteDebitDiffere']:
                card = Account()
                card.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']
                yield card

    @need_login
    def get_accounts_list(self):
        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'))

        # get account iban on transfer page
        account_ibans = {}
        try:
            self.json_transfer.go()
        except (TransferBankError, ClientError, BrowserUnavailable):
            # some user can't access this page
            pass
        else:
            account_ibans = self.page.get_account_ibans_dict()

        # get accounts coming
        self.accounts_syntheses.go()
        account_comings = self.page.get_account_comings()

        self.accounts.go()
        for account in self.page.iter_accounts():
            for card in self.iter_cards(account):
                card.parent = account
                yield card

            if account._prestation_id in account_ibans:
                account.iban = account_ibans[account._prestation_id]

            if account._prestation_id in account_comings:
                account.coming = account_comings[account._prestation_id]

            if account.type in (account.TYPE_LOAN, account.TYPE_CONSUMER_CREDIT, ):
                self.loans.stay_or_go(conso=(account._loan_type == 'PR_CONSO'))
                account = self.page.get_loan_account(account)

            if account.type == account.TYPE_REVOLVING_CREDIT:
                self.loans.stay_or_go(conso=(account._loan_type == 'PR_CONSO'))
                account = self.page.get_revolving_account(account)

            yield account

    @need_login
    def iter_history(self, account):
        if account.type in (account.TYPE_LOAN, account.TYPE_MARKET, account.TYPE_CONSUMER_CREDIT, ):
            return

        if account.type == Account.TYPE_PEA and not ('Espèces' in account.label or 'ESPECE' in account.label):
            return

        if account.type in (account.TYPE_LIFE_INSURANCE, account.TYPE_PERP, ):
            # request to get json is not available yet, old request to get html response
            self.account_details_page.go(params={'idprest': account._prestation_id})
            link = self.page.get_history_link()
            if link:
                self.location(self.absurl(link))
                for tr in self.page.iter_li_history():
                    yield tr
            return

        if account.type == account.TYPE_REVOLVING_CREDIT and account._loan_type != 'PR_CONSO':
            # request to get json is not available yet, old request to get html response
            self.account_details_page.go(params={'idprest': account._prestation_id})
            self.page.go_history_page()
            for tr in self.page.iter_credit_history():
                yield tr
            return

        if account.type == account.TYPE_REVOLVING_CREDIT and not account._is_json_histo:
            # Waiting for account with transactions
            return

        if account.type == account.TYPE_CARD:
            self.history.go(params={'b64e200_prestationIdTechnique': account.parent._internal_id})
            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
                    yield card_tr
            return

        self.history.go(params={'b64e200_prestationIdTechnique': account._internal_id})
        for transaction in self.page.iter_history():
            yield transaction

    @need_login
    def iter_coming(self, account):
        if account.type in (account.TYPE_LOAN, account.TYPE_MARKET, account.TYPE_PEA,
                            account.TYPE_LIFE_INSURANCE, account.TYPE_REVOLVING_CREDIT,
                            account.TYPE_CONSUMER_CREDIT, Account.TYPE_PERP, ):
            return

        internal_id = account._internal_id
        if account.type == account.TYPE_CARD:
            internal_id = account.parent._internal_id
        self.history.go(params={'b64e200_prestationIdTechnique': internal_id})

        if account.type == account.TYPE_CARD:
            for transaction in self.page.iter_future_transactions(acc_prestation_id=account._prestation_id):
                # coming transactions on this page are not in coming balance
                # except for defered card coming transaction, use only for it for the moment
                if transaction._card_coming:
                    for card_coming in transaction._card_coming:
                        card_coming.date = transaction.date
                        yield card_coming
                    return

        for intraday_tr in self.page.iter_intraday_comings():
            yield intraday_tr

    @need_login
    def iter_investment(self, account):
        if account.type not in (Account.TYPE_MARKET, Account.TYPE_LIFE_INSURANCE,
                                Account.TYPE_PEA, Account.TYPE_PERP, ):
            self.logger.debug('This account is not supported')
            return

        # 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

        if account.type in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_PERP, ):
            if self.page.has_link():
                self.life_insurance_invest.go()

            for invest in self.page.iter_investment():
                yield invest

    @need_login
    def iter_recipients(self, account):
        try:
            self.json_transfer.go()
        except TransferBankError:
            return []
        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()

        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 = {
            'b64e200_idVirement': transfer.id
        }
        # 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 transfer

    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']),
            ('csa_op', 'sign')
        ]
        # 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),
            ('oob_op', 'sign')
        ]
        # 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'))

    @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()
        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()

        r = self.open(self.absurl('/sec/getsigninfo.json'), data=data)
        sign_method = r.page.get_sign_method()

        # 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)
        assert False, '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()
        profile = self.page.get_profile()
        self.xml_profile_page.go()
        profile.email = self.page.get_email()
        return profile

    @need_login
    def iter_subscription(self):
        try:
            profile = self.get_profile()
            subscriber = profile.name
        except ProfileMissing:
            subscriber = NotAvailable

        # subscriptions which have statements are present on the last statement page
        self.bank_statement.go()
        subscriptions_list = list(self.page.iter_subscription())

        # this way the no statement accounts are excluded
        # and the one keeped have all the data and parameters needed
        self.bank_statement_search.go()
        for sub in self.page.iter_searchable_subscription(subscriber=subscriber):
            found_sub = find_object(subscriptions_list, id=sub.id)
            if found_sub:
                yield sub

    @need_login
    def iter_documents(self, subscribtion):
        end_date = datetime.today()

        # 5 years since it goes with a 2 months step
        security_limit = 30
        i = 0
        while i < security_limit:
            self.bank_statement_search.go()
            self.page.post_form(subscribtion, end_date)

            # No more documents
            if self.page.has_error_msg():
                break

            for d in self.page.iter_documents(subscribtion):
                yield d

            # 3 months step because the documents list is inclusive
            # from the 08 to the 06, the 06 statement is included
            end_date = end_date - relativedelta(months=+3)
            i += 1