From a53dc875acf2d83f27ec6db43beb66de6336540f Mon Sep 17 00:00:00 2001 From: Vincent Ardisson Date: Sat, 3 Apr 2021 09:12:55 +0200 Subject: [PATCH] [swisslife] new CapBankWealth module --- modules/swisslife/__init__.py | 24 ++ modules/swisslife/browser.py | 305 +++++++++++++++++++ modules/swisslife/module.py | 69 +++++ modules/swisslife/pages.py | 549 ++++++++++++++++++++++++++++++++++ 4 files changed, 947 insertions(+) create mode 100644 modules/swisslife/__init__.py create mode 100644 modules/swisslife/browser.py create mode 100644 modules/swisslife/module.py create mode 100644 modules/swisslife/pages.py diff --git a/modules/swisslife/__init__.py b/modules/swisslife/__init__.py new file mode 100644 index 0000000000..1517252294 --- /dev/null +++ b/modules/swisslife/__init__.py @@ -0,0 +1,24 @@ +# -*- coding: utf-8 -*- + +# Copyright(C) 2012-2021 Budget Insight +# +# 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 .module import SwisslifeModule + + +__all__ = ['SwisslifeModule'] diff --git a/modules/swisslife/browser.py b/modules/swisslife/browser.py new file mode 100644 index 0000000000..a56af3852b --- /dev/null +++ b/modules/swisslife/browser.py @@ -0,0 +1,305 @@ +# -*- coding: utf-8 -*- + +# Copyright(C) 2012-2021 Budget Insight +# +# 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 __future__ import unicode_literals + +from requests.exceptions import ConnectionError + +from weboob.browser import LoginBrowser, URL, need_login +from weboob.browser.exceptions import ClientError +from weboob.exceptions import BrowserIncorrectPassword, BrowserHTTPError, BrowserUnavailable, BrowserHTTPNotFound +from weboob.browser.exceptions import ServerError +from weboob.capabilities.bank import Account +from weboob.capabilities.wealth import Per, PerVersion, Investment, Pocket +from weboob.capabilities.base import NotAvailable, empty +from weboob.tools.capabilities.bank.transactions import sorted_transactions + +from .pages import ( + ProfilePage, AccountsPage, AccountDetailPage, AccountVieEuroPage, InvestmentPage, + AccountVieUCCOPage, AccountVieUCCODetailPage, AccountVieUCPage, BankAccountDetailPage, + BankAccountTransactionsPage, MaintenancePage, +) + + +class SwisslifeBrowser(LoginBrowser): + profile = URL(r'/api/v3/personne', ProfilePage) + accounts = URL(r'/api/v3/contrat/home', AccountsPage) + investment = URL(r'/api/v3/contrat/.*/encours.*', InvestmentPage) + bank_account_detail = URL(r'/api/v3/contrat/detail/(?P.*)', BankAccountDetailPage) + bank_account_transactions = URL( + r'/api/v3/contrat/operationListe/(?P.+)/(?P\d+)/(?P\d+)$', + BankAccountTransactionsPage + ) + account_vie_ucco_detail = URL(r'/api/v3/contrat/.*/operations.*', AccountVieUCCODetailPage) + account_vie_ucco = URL(r'/api/v3/contrat/(?P.*)\?typeContrat=ADHERENT', AccountVieUCCOPage) + account_detail = URL( + r'/api/v3/contrat/(?P.*)', + r'/api/v3/contrat/(?P.*)/encours\?codeProfil=(?P.*)', + AccountDetailPage + ) + account_vie_euro = URL(r'/api/v3/contratVieEuro/(?P.*)', AccountVieEuroPage) + account_vie_uc = URL(r'/api/v3/contratVieucEntreprise/.*', AccountVieUCPage) + + maintenance = URL(r'/api/v3/authenticate', MaintenancePage) + + def __init__(self, domain, *args, **kwargs): + super(SwisslifeBrowser, self).__init__(*args, **kwargs) + + self.BASEURL = 'https://%s' % (domain if '.' in domain else 'myswisslife.fr') + self.session.headers['X-Requested-With'] = 'XMLHttpRequest' + self.session.headers['Accept'] = 'application/json, text/javascript, */*; q=0.01' + + def do_login(self): + try: + self.location('/api/v3/authenticate', data={'username': self.username, 'password': self.password, 'media': 'web'}) + except ClientError: + raise BrowserIncorrectPassword("Votre identifiant utilisateur est inconnu ou votre mot de passe est incorrect.") + + if self.maintenance.is_here(): + # If the website is in maintenance, we are redirected to a HTML page + raise BrowserUnavailable(self.page.get_error_message()) + + @need_login + def go_to_account(self, account): + # For some accounts, we get a 500 error even on the website... + # These accounts have no balance so we do not fetch them. + try: + self.location(account.url) + except ServerError as e: + # Some accounts sometimes return 503, we can try again + if e.response.status_code == 503: + try: + self.accounts.go() + self.location(account.url) + except ServerError as e: + if e.response.status_code == 503: + raise BrowserUnavailable() + else: + return True + self.logger.warning('Server Error: could not fetch the details for account %s.', account.label) + except ClientError as e: + # Some accounts return 403 forbidden and don't appear on the website + if e.response.status_code == 403: + self.logger.warning('Client Error: could not fetch the details for account %s.', account.label) + raise + except BrowserHTTPNotFound: + # Some accounts return 404 with an error message on the website + self.logger.warning('404 Error: could not fetch the details for account %s.', account.label) + else: + return True + + @need_login + def iter_accounts(self): + try: + self.accounts.stay_or_go() + except BrowserHTTPError: + raise BrowserUnavailable() + + if not self.page.has_accounts(): + self.logger.warning('Could not find any account') + return + + bank_accounts = self.page.iter_bank_accounts() + wealth_accounts = self.page.iter_wealth_accounts() + + for account in bank_accounts: + self.go_to_account(account) + account._is_market_account = self.page.is_market_account() + yield account + + for account in wealth_accounts: + # The new API with account details is only available for bank accounts. + # We still use the old version until savings accounts are migrated to the new API. + if not self.go_to_account(account): + # For some accounts, the details URL systematically leads to an error. + # We do not fetch them. + continue + if self.account_vie_euro.is_here() and self.page.is_error(): + # Sometimes the account URL works but it lands on a page with a status and error message. + # The account has no balance and no info on the website, we do not fetch it. + continue + if any(( + self.account_detail.is_here(), + self.account_vie_euro.is_here(), + self.account_vie_ucco.is_here(), + self.account_vie_uc.is_here(), + )): + self.page.fill_account(obj=account) + if account.type == Account.TYPE_UNKNOWN: + if not empty(account._fiscality_type): + # Type account using fiscality if the label could not type the account properly + account.type = account._fiscality_type + else: + self.logger.warning('Could not type account "%s"', account.label) + + if account.type == Account.TYPE_PER: + # Transform account into PER and set PER version + per = Per.from_dict(account.to_dict()) + + per._fiscality_type = account._fiscality_type + per._profile_types = account._profile_types + per._history_urls = account._history_urls + per._is_bank_account = account._is_bank_account + per._is_market_account = account._is_market_account + + if 'PER INDIVIDUEL' in per.label.upper(): + per.version = PerVersion.PERIN + elif 'PER ENTREPRISE' in per.label.upper(): + per.version = PerVersion.PERCOL + + # No information concerning the PER provider_type + per.provider_type = NotAvailable + yield per + + else: + yield account + + @need_login + def get_profile(self): + self.profile.stay_or_go() + return self.page.get_profile() + + def create_euro_fund(self, account): + inv = Investment() + inv.label = 'FONDS EN EUROS' + inv.valuation = account.balance + inv.code = NotAvailable + inv.code_type = NotAvailable + return inv + + @need_login + def iter_investment(self, account): + if account.balance == 0: + return + + if self.bank_account_detail.match(account.url): + self.location(account.url) + if account._is_market_account: + for inv in self.page.iter_investment(): + yield inv + elif not account._profile_types: + if not account.url: + raise NotImplementedError() + if self.account_vie_euro.match(account.url): + yield self.create_euro_fund(account) + else: + try: + self.location(account.url) + for inv in self.page.iter_investment(): + yield inv + except BrowserHTTPError: + yield self.create_euro_fund(account) + # No invest on this account + except BrowserHTTPNotFound as e: + self.logger.warning(e) + else: + for profile_type in account._profile_types: + try: + self.account_detail.go(id=account.number, profile_type=profile_type) + except ClientError as e: + # Some accounts return 403 forbidden and don't appear on the website + if e.response.status_code == 403: + self.logger.warning('Client Error: could not fetch investments for account %s.', account.label) + continue + else: + raise + for inv in self.page.iter_investment(): + inv._profile_type = profile_type + yield inv + + @need_login + def iter_pocket(self, account): + if not account._profile_types: + raise NotImplementedError() + + # not the best way but those names are generated with js + natures = { + 'UC': 'Unités de compte', + 'DV': 'Fonds en Euros', + } + profiles = { + 'AP': 'sous allocation pilotée', + 'LIBRE': 'sous allocation libre' + } + + pockets = [] + # for now, we create a pocket for each investment + for inv in self.iter_investment(account): + pocket = Pocket() + nature = natures.get(inv._nature) + if nature: + pocket.label = ('%s %s' % (nature, profiles.get(inv._profile_type, ""))).strip() + pocket.amount = inv.valuation + pocket.quantity = inv.quantity + pocket.availability_date = NotAvailable + pocket.condition = Pocket.CONDITION_AVAILABLE + pocket.investment = inv + pockets.append(pocket) + return pockets + + @need_login + def iter_history(self, account): + if account._is_bank_account: + # We must do the pagination manually with the URL. + # There is no indication that we are on the last page except if there are less transactions than the page size. + # If the total number of transactions is a multiple of the size, we arrive at a page with no transaction. + # This last page format is different (no operations list at all). + index = 0 + size = 50 + has_next_page = True + iteration = 0 + while has_next_page and iteration < 100: + iteration += 1 + self.bank_account_transactions.go(account_id=account.id, index=index, size=size) + has_next_page = self.page.has_next_page(size) + index += size + if self.page.has_operations(): + for tr in self.page.iter_history(): + yield tr + elif account._history_urls: + for urls in account._history_urls: + try: + self.location(urls) + except (ConnectionError, ServerError) as e: + # Error on swisslife website. + self.logger.error(e) + for tr in self.page.iter_history(): + yield tr + elif not account.url: + raise NotImplementedError() + # Article 39 accounts history + elif 'contratVieucEntreprise' in account.url: + # This key param seems to be hardcoded for this type of contract + params = {'natureCodes': 'A02A,A02B,A02D,A02T,B03A,B03C,B03I,B03R,B03S,B03T,C06A,C06J,C06L,C06M,C06S,C06P,C06B'} + self.location('/api/v3/contratVieucEntreprise/operations/%s' % account.id, params=params) + for tr in sorted_transactions(self.page.iter_history()): + yield tr + elif 'contratVieEuro' in account.url: + # If there are no transactions, the request will fail + try: + self.location(account.url + '/primesPayees') + except (BrowserHTTPError, BrowserHTTPNotFound): + self.logger.warning('Could not access history for account %s', account.id) + else: + for tr in sorted_transactions(self.page.iter_history()): + yield tr + else: + self.location(account.url) + for tr in sorted_transactions(self.page.iter_history()): + yield tr diff --git a/modules/swisslife/module.py b/modules/swisslife/module.py new file mode 100644 index 0000000000..bebdad58c5 --- /dev/null +++ b/modules/swisslife/module.py @@ -0,0 +1,69 @@ +# -*- coding: utf-8 -*- + +# Copyright(C) 2012-2021 Budget Insight +# +# 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 __future__ import unicode_literals + +from weboob.capabilities.profile import CapProfile +from weboob.capabilities.wealth import CapBankWealth +from weboob.tools.backend import Module, BackendConfig +from weboob.tools.value import ValueBackendPassword, Value + +from .browser import SwisslifeBrowser + + +__all__ = ['SwisslifeModule'] + + +class SwisslifeModule(Module, CapBankWealth, CapProfile): + MODULE = 'swisslife' + NAME = 'swisslife' + DESCRIPTION = 'SwissLife' + MAINTAINER = 'Christophe François' + EMAIL = 'christophe.francois@budget-insight.com' + LICENSE = 'LGPLv3+' + VERSION = '2.1' + CONFIG = BackendConfig( + ValueBackendPassword('login', label='Identifiant personnel', masked=False), + ValueBackendPassword('password', label='Mot de passe'), + Value('domain', label='Domain', default='myswisslife.fr'), + ) + + BROWSER = SwisslifeBrowser + + def create_default_browser(self): + return self.create_browser( + self.config['domain'].get(), + self.config['login'].get(), + self.config['password'].get(), + ) + + def iter_accounts(self): + return self.browser.iter_accounts() + + def iter_history(self, account): + return self.browser.iter_history(account) + + def iter_investment(self, account): + return self.browser.iter_investment(account) + + def get_profile(self): + return self.browser.get_profile() + + def iter_pocket(self, account): + return self.browser.iter_pocket(account) diff --git a/modules/swisslife/pages.py b/modules/swisslife/pages.py new file mode 100644 index 0000000000..3ebc7e0687 --- /dev/null +++ b/modules/swisslife/pages.py @@ -0,0 +1,549 @@ +# -*- coding: utf-8 -*- + +# Copyright(C) 2012-2021 Budget Insight +# +# 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 __future__ import unicode_literals + +import datetime + +from weboob.browser.elements import method, ListElement, ItemElement, DictElement +from weboob.browser.filters.json import Dict +from weboob.browser.filters.standard import ( + CleanText, CleanDecimal, Eval, Field, Map, Currency, Regexp, + Env, Date, BrowserURL, Coalesce, MultiJoin, MapIn, Lower, +) +from weboob.browser.pages import LoggedPage, JsonPage, HTMLPage +from weboob.capabilities.bank import Account, Transaction +from weboob.capabilities.wealth import Investment +from weboob.capabilities.profile import Person +from weboob.capabilities.base import NotAvailable, empty +from weboob.tools.compat import urlparse +from weboob.tools.capabilities.bank.transactions import FrenchTransaction +from weboob.tools.capabilities.bank.investments import IsinCode, IsinType +from weboob.tools.date import parse_french_date + + +def date_from_timestamp(date): + # 'date' may be None or NotAvailable + if empty(date): + return NotAvailable + return datetime.datetime.fromtimestamp(date/1000) + + +class MaintenancePage(HTMLPage): + def is_here(self): + return bool(CleanText('//h1[contains(text(), "Opération technique exceptionnelle")]')(self.doc)) + + def get_error_message(self): + return CleanText('//h1')(self.doc) + + +class ProfilePage(LoggedPage, JsonPage): + @method + class get_profile(ItemElement): + klass = Person + + obj_name = MultiJoin(Dict('prenom'), Dict('nom'), pattern=' ') + obj_email = Dict('emailPrive') + obj_job = Dict('professionLibelle', default=NotAvailable) + obj_address = CleanText(MultiJoin( + Dict('adresse/mentionSupplementaire', default=''), + Dict('adresse/voie', default=''), + Dict('adresse/codePostal', default=''), + Dict('adresse/ville', default=''), + pattern=' ' + )) + + +class AccountsPage(LoggedPage, JsonPage): + def has_accounts(self): + return Dict('epargne/contrats', default=None)(self.doc) is not None + + @method + class iter_bank_accounts(DictElement): + item_xpath = 'epargne/contrats' + + class item(ItemElement): + klass = Account + + def condition(self): + return Dict('estBanque')(self) + + ACCOUNT_TYPES = { + 'Espèces': Account.TYPE_DEPOSIT, + 'Titres': Account.TYPE_MARKET, + 'PEA': Account.TYPE_PEA, + } + + obj_id = obj_number = CleanText(Dict('compteNumero')) + obj_label = CleanText(Dict('compteLibelle')) + + def obj_type(self): + account_type = MapIn(CleanText(Dict('compteType')), self.ACCOUNT_TYPES, Account.TYPE_UNKNOWN)(self) + if account_type == Account.TYPE_UNKNOWN: + self.logger.warning('Could not type account "%s"', Field('label')(self)) + return account_type + + def obj_opening_date(self): + return parse_french_date(Dict('ouvertureDate')(self)) + + obj_balance = CleanDecimal.SI(Dict('soldeDeviseCompteMontant')) + obj_currency = Currency(Dict('devise')) + + obj__profile_types = NotAvailable + obj__fiscality_type = NotAvailable + obj__history_urls = NotAvailable + + obj__is_bank_account = True + + obj_url = BrowserURL('bank_account_detail', id=Field('id')) + + @method + class iter_wealth_accounts(DictElement): + item_xpath = 'epargne/contrats' + + class item(ItemElement): + klass = Account + + def condition(self): + if Dict('estBanque')(self) or Dict('estResilie')(self): + return False + # Only these accounts have a details URL and a balance. + return any(( + Dict('estVieArt39IFC')(self), + Dict('estVieUC')(self), + Dict('estContratBull')(self), + Dict('estVieEuro')(self), + )) + + ACCOUNT_TYPES = { + 'EPARGNE RETRAITE': Account.TYPE_PERP, + 'ARTICLE 83': Account.TYPE_ARTICLE_83, + 'SWISSLIFE RETRAITE ENTREPRISES': Account.TYPE_ARTICLE_83, + 'SWISSLIFE RETRAITE ARTICLE 83': Account.TYPE_ARTICLE_83, + 'SWISSLIFE PER ENTREPRISES': Account.TYPE_PER, + 'SWISS RETRAITE ADDITIVE': Account.TYPE_ARTICLE_83, + 'GARANTIE RETRAITE': Account.TYPE_LIFE_INSURANCE, + 'GARANTIE RETRAITE ENTREPRISES 2000': Account.TYPE_LIFE_INSURANCE, + 'GARANTIE RETRAITE INDEPENDANTS 2000': Account.TYPE_LIFE_INSURANCE, + 'RETRAITE ART 83 CGI MULTISUPPORTS': Account.TYPE_ARTICLE_83, + 'ERES RETRAITE 83 V2': Account.TYPE_ARTICLE_83, + 'SWISSLIFE INDEM FIN DE CARRIERES': Account.TYPE_PER, + 'SWISSLIFE PER INDIVIDUEL': Account.TYPE_PER, + } + + obj_id = obj_number = CleanText(Dict('numContrat')) + obj__contract_id = CleanText(Dict('contratId')) + obj_label = CleanText(Dict('libelleProduit')) + + obj_currency = 'EUR' + + obj__is_bank_account = False + obj__is_market_account = True + + def obj_opening_date(self): + return date_from_timestamp(Dict('dateEffet')(self)) + + def obj_type(self): + account_type = Map(Field('label'), self.ACCOUNT_TYPES, Account.TYPE_UNKNOWN)(self) + return account_type + + def obj_url(self): + # Article 39 accounts have a specific JSON that only contains the balance + if Dict('estVieArt39IFC')(self): + return self.page.browser.absurl('/api/v3/contratVieucEntreprise/encours/%s' % Field('id')(self)) + # Regular accounts + elif Dict('estVieUC')(self) and not Dict('estResilie')(self): + if Dict('typeContrat')(self) == 'ADHERENT': + return BrowserURL('account_vie_ucco', id=Field('id'))(self) + return BrowserURL('account_detail', id=Field('id'))(self) + # Life insurances and retirement plans + elif Dict('estVieEuro')(self) and not Dict('estResilie')(self): + if Dict('estEpargne')(self): + return BrowserURL('account_vie_euro', id=Field('_contract_id'))(self) + return BrowserURL('account_vie_euro', id=Field('_contract_id'))(self) + # TODO: The following assert is replaced temporarily by a warning. + # It should be reverted after the release when we find the correct url for these accounts. + # assert False, 'Could not find details URL for account %s' % Field('id')(self) + self.logger.warning('Could not find details URL for account %s', Field('id')(self)) + + +class IterInvestment(ListElement): + class item(ItemElement): + klass = Investment + + obj_label = Dict('nomSupport') + obj_quantity = CleanDecimal.SI(Dict('nbPart', default=''), default=NotAvailable) + obj_unitvalue = CleanDecimal.SI(Dict('valPart', default=''), default=NotAvailable) + obj_valuation = CleanDecimal.SI(Dict('montantNet')) + obj_diff = CleanDecimal.SI(Dict('evolution', default=''), default=NotAvailable) + obj__nature = Dict('codeNature') + obj_code = IsinCode(CleanText(Dict('codeIsin')), default=NotAvailable) + obj_code_type = IsinType(CleanText(Dict('codeIsin')), default=NotAvailable) + + def obj_unitprice(self): + unitprice = CleanDecimal.SI(Dict('prixMoyenAchat', default=''), default=NotAvailable)(self) + if unitprice == 0: + return NotAvailable + return unitprice + + def obj_vdate(self): + return date_from_timestamp(Dict('dateValeur', default=NotAvailable)(self)) + + def find_elements(self): + for el in self.el: + yield el + + +class BankAccountDetailPage(LoggedPage, JsonPage): + def is_market_account(self): + return Dict('operationsListe/estTitre')(self.doc) + + @method + class iter_investment(DictElement): + def find_elements(self): + # All investments are not on the same depth. + # Investments with stocks are grouped in a single investment so we must skip it and get them instead. + for obj in Dict('positions/data')(self): + if Dict('actions', default=None)(obj) is None: + yield obj + else: + for sub_obj in Dict('actions')(obj): + yield sub_obj + + class item(ItemElement): + klass = Investment + + # Investment characteristics are stored in a field data (when it exists) that looks like this: + # "data": [ + # {"key": "XXX", "value": "XXX"}, + # {"key": "XXX", "value": "XXX"}, + # ... + # ] + def parse(self, el): + for obj in Dict('caracteristiques/data', default={})(el): + key = CleanText(Dict('key', default=NotAvailable), default=NotAvailable)(obj) + value = Dict('value', default=NotAvailable)(obj) + + if empty(key) or empty(value): + continue + + if key == 'Code ISIN': + self.env['code'] = IsinCode(default=NotAvailable).filter( + CleanText(default=NotAvailable).filter(value) + ) + self.env['code_type'] = IsinType(default=NotAvailable).filter( + CleanText(default=NotAvailable).filter(value) + ) + elif key == '+/- value latente': + self.env['diff'] = CleanDecimal.French(default=NotAvailable).filter(value) + elif key == 'Quantité': + self.env['quantity'] = CleanDecimal.SI(default=NotAvailable).filter(value) + elif key == 'Prix de revient unitaire': + self.env['unitprice'] = CleanDecimal.French(default=NotAvailable).filter(value) + elif key == 'Valorisation du titre': + self.env['unitvalue'] = CleanDecimal.French(default=NotAvailable).filter(value) + elif key == 'Date de valorisation': + self.env['vdate'] = Date(default=NotAvailable, parse_func=parse_french_date).filter( + CleanText(default=NotAvailable).filter(value) + ) + + obj_valuation = CleanDecimal.French(Dict('montant')) + obj_code = Env('code', default=NotAvailable) + obj_code_type = Env('code_type', default=NotAvailable) + obj_diff = Env('diff', default=NotAvailable) + obj_quantity = Env('quantity', default=NotAvailable) + obj_unitprice = Env('unitprice', default=NotAvailable) + obj_unitvalue = Env('unitvalue', default=NotAvailable) + obj_vdate = Env('vdate', default=NotAvailable) + obj_label = Coalesce( + Regexp(CleanText(Dict('libelle')), r'(.*) \(', default=None), + CleanText(Dict('libelle')) + ) + + def obj_portfolio_share(self): + percentage = CleanDecimal.SI(Dict('pourcentage'))(self) + if percentage: + return percentage / 100 + return NotAvailable + + +class BankAccountTransactionsPage(LoggedPage, JsonPage): + def has_operations(self): + return Dict('operations', default=None)(self.doc) is not None + + def has_next_page(self, size): + return self.has_operations() and len(Dict('operations', default={})(self.doc)) == size + + @method + class iter_history(DictElement): + item_xpath = 'operations' + + class item(ItemElement): + klass = Transaction + + # Transaction characteristics are stored in a field detailData that looks like this: + # "data": [ + # {"key": "XXX", "value": "XXX"}, + # {"key": "XXX", "value": "XXX"}, + # ... + # ] + def parse(self, el): + for obj in Dict('detailData')(el): + key = CleanText(Dict('key', default=NotAvailable), default=NotAvailable)(obj) + value = Dict('value', default=NotAvailable)(obj) + + if empty(key) or empty(value): + continue + + if key == "Date de valeur de l'opération" or key == "Date de valeur": + self.env['vdate'] = Date(default=NotAvailable, parse_func=parse_french_date).filter( + CleanText(default=NotAvailable).filter(value) + ) + elif key == "Date d'exécution": + self.env['rdate'] = Date(default=NotAvailable, parse_func=parse_french_date).filter( + Regexp(pattern=r'(.*) \d+h\d+', default=NotAvailable).filter( + CleanText(default=NotAvailable).filter(value) + ) + ) + elif key == 'Quantité': + self.env['quantity'] = CleanDecimal.SI(default=NotAvailable).filter(value) + elif key == "Prix unitaire moyen": + self.env['unitprice'] = CleanDecimal.French(default=NotAvailable).filter(value) + + # TODO This list of labels is from the API test environment, it should be revised. + TRANSACTION_TYPES = { + 'frais de tenue': Transaction.TYPE_BANK, + "frais d'administration": Transaction.TYPE_BANK, + 'envoi chequier': Transaction.TYPE_BANK, + 'operation': Transaction.TYPE_BANK, + 'saisie de valeurs': Transaction.TYPE_BANK, + 'blocage de fonds': Transaction.TYPE_BANK, + 'droits de garde': Transaction.TYPE_BANK, + 'dividende': Transaction.TYPE_BANK, + 'conversion': Transaction.TYPE_BANK, + 'interets': Transaction.TYPE_BANK, + 'apport': Transaction.TYPE_TRANSFER, + 'versement': Transaction.TYPE_TRANSFER, + 'virement': Transaction.TYPE_TRANSFER, + 'cheque': Transaction.TYPE_CHECK, + 'paiement en faveur': Transaction.TYPE_ORDER, + } + + obj_amount = CleanDecimal.French(Dict('deviseMontant')) + obj_label = CleanText(Dict('operationLibelle')) + obj_date = Date(CleanText(Dict('operationDate')), parse_func=parse_french_date) + obj_vdate = Env('vdate', default=NotAvailable) + obj_rdate = Env('rdate', default=NotAvailable) + + def obj_type(self): + if Dict('instrumentFinancier', default=None)(self): + return Transaction.TYPE_BANK + account_type = MapIn(Lower(Field('label'), transliterate=True), self.TRANSACTION_TYPES, default=Transaction.TYPE_UNKNOWN)(self) + if account_type == Transaction.TYPE_UNKNOWN: + self.logger.warning('Could not type transaction "%s"', Field('label')(self)) + return account_type + + def obj_investments(self): + if Dict('instrumentFinancier', default=None)(self) is None: + return NotAvailable + inv = Investment() + inv.valuation = Eval(lambda x: abs(x), Field('amount'))(self) + inv.label = Regexp(CleanText(Dict('instrumentFinancier')), r'(.*) \(', default=None)(self) \ + or CleanText(Dict('instrumentFinancier'))(self) + inv.code = IsinCode(Regexp(CleanText(Dict('instrumentFinancier')), r'.* \((.*)\)', default=''), default=NotAvailable)(self) + inv.code_type = IsinType(Regexp(CleanText(Dict('instrumentFinancier')), r'.* \((.*)\)', default=''), default=NotAvailable)(self) + inv.quantity = Env('quantity')(self) + inv.unitprice = Env('unitprice')(self) + return [inv] + + +class AccountDetailPage(LoggedPage, JsonPage): + def is_error(self): + # OK status can be null or 200 + return CleanText(Dict('status', default=''))(self.doc) not in ('', '200') + + TYPES = { + 'MADELIN': Account.TYPE_MADELIN, + 'Article 83': Account.TYPE_ARTICLE_83, + 'PERP': Account.TYPE_PERP, + } + + @method + class fill_account(ItemElement): + obj_currency = 'EUR' + obj_balance = CleanDecimal(Dict('encours/montantNetDeFrais', default='0')) + obj_valuation_diff = CleanDecimal(Dict('encours/plusValue', default=''), default=NotAvailable) + obj__history_urls = NotAvailable + + def obj__profile_types(self): + profile_types = [] + for v in Dict('profils', default=NotAvailable)(self).values(): + if isinstance(v, dict) and "typeProfil" in v: + profile_types.append(v['typeProfil']) + return profile_types + + def obj__fiscality_type(self): + return Map(Dict('fiscalite', default=''), self.page.TYPES, Account.TYPE_LIFE_INSURANCE)(self) + + @method + class iter_history(ListElement): + class item(ItemElement): + klass = FrenchTransaction + + obj_raw = FrenchTransaction.Raw(Dict('nature')) + obj_amount = CleanDecimal(Dict('montantBrut')) + + def obj_date(self): + return date_from_timestamp(Dict('dateEffet', default=NotAvailable)(self)) + + def obj_investments(self): + l = Dict('supports')(self) + m = IterInvestment(self.page, el=l) + return list(m()) + + def find_elements(self): + if 'mouvements' in self.el: + for el in self.el.get('mouvements', ()): + yield el + + +class AccountVieEuroPage(AccountDetailPage): + @method + class fill_account(ItemElement): + obj_balance = Dict('rachatValeurMontant') & CleanDecimal + obj_valuation_diff = CleanDecimal(Dict('plusValueMontant', default=''), default=NotAvailable) + obj__profile_types = NotAvailable + obj__history_urls = NotAvailable + obj__fiscality_type = NotAvailable + + @method + class iter_history(ListElement): + class item(ItemElement): + klass = FrenchTransaction + + obj_raw = Eval(lambda t: 'Primes' if t == 'PRI' else 'Règlement', Dict('operation/natureCode')) + obj_amount = Eval(lambda t: t/100, CleanDecimal.SI(Dict('montantBrut', default='0'))) + + def obj_date(self): + return date_from_timestamp(Dict('effetDate', default=NotAvailable)(self)) + + # "impaye" returns timestamp -2211757200000 which raises an error in backend + def validate(self, obj): + return obj.date.year > 1969 + + def find_elements(self): + for el in self.el: + yield el + + +class AccountVieUCCOPage(AccountDetailPage): + @method + class fill_account(ItemElement): + obj_balance = CleanDecimal.SI(Dict('encours/montantEpargne', default='0')) + # Currency not available in the JSON. hardcoded until someone get a life-insurance != EUR + obj_currency = "EUR" + obj__profile_types = NotAvailable + obj__fiscality_type = NotAvailable + + def obj__history_urls(self): + parsed_url = urlparse(self.obj.url) + history_url = 'https://' + parsed_url.netloc + parsed_url.path + '/operations?' + parsed_url.query + return [history_url] + + @method + class iter_investment(IterInvestment): + def find_elements(self): + for el in self.el.get('encoursListe', ()): + yield el + + +class AccountVieUCPage(AccountDetailPage): + @method + class fill_account(ItemElement): + obj_balance = CleanDecimal.SI(Dict('rachatValeurMontant', default='0')) + obj__history_urls = NotAvailable + obj__profile_types = NotAvailable + obj__fiscality_type = NotAvailable + + @method + class iter_investment(DictElement): + + class item(ItemElement): + klass = Investment + + obj_label = CleanText(Dict('nomSupport')) + obj_valuation = CleanDecimal.SI(Dict('montantNet')) + obj_quantity = CleanDecimal.SI(Dict('nbPart')) + obj_unitvalue = CleanDecimal.SI(Dict('valPart')) + obj_code = IsinCode(Dict('codeIsin'), default=NotAvailable) + obj_code_type = IsinType(Dict('codeIsin'), default=NotAvailable) + + def obj_unitprice(self): + unitprice = CleanDecimal.SI(Dict('prixMoyenAchat'))(self) + if unitprice == 0: + return NotAvailable + return unitprice + + def obj_vdate(self): + return date_from_timestamp(Dict('dateValeur', default=NotAvailable)(self)) + + @method + class iter_history(DictElement): + item_xpath = 'operationItems' + + class item(ItemElement): + klass = FrenchTransaction + + obj_id = Dict('operationId') + obj_label = CleanText(Dict('natureLibelle')) + obj_amount = CleanDecimal.SI(Dict('montantBrut')) + obj_type = FrenchTransaction.TYPE_BANK + + def obj_date(self): + return date_from_timestamp(Dict('effetDate', default=NotAvailable)(self)) + + def obj_vdate(self): + return date_from_timestamp(Dict('effetDate', default=NotAvailable)(self)) + + def validate(self, obj): + return CleanText(Dict('etatLibelle'))(self) == 'Validé' + + +class AccountVieUCCODetailPage(LoggedPage, JsonPage): + @method + class iter_history(ListElement): + class item(ItemElement): + klass = FrenchTransaction + + obj_raw = FrenchTransaction.Raw(Dict('nature')) + obj_amount = CleanDecimal.SI(Dict('montantBrut')) + obj__fiscality_type = NotAvailable + + def obj_date(self): + return date_from_timestamp(Dict('dateEffet', default=NotAvailable)(self)) + + def find_elements(self): + for el in self.el: + yield el + + +class InvestmentPage(LoggedPage, JsonPage): + @method + class iter_investment(IterInvestment): + pass -- GitLab