Skip to content
Commits on Source (110)
# -*- coding: utf-8 -*-
# Copyright(C) 2015 James GALT
# Copyright(C) 2012-2019 Budget Insight
#
# This file is part of a weboob module.
#
......@@ -19,62 +19,15 @@
from __future__ import unicode_literals
from random import randint
from weboob.browser import URL, LoginBrowser, need_login
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable, BrowserPasswordExpired
from weboob.tools.compat import basestring
from .pages import (
LoginPage, IndexPage, WrongPasswordPage, WrongWebsitePage,
AccountDetailPage, AccountHistoryPage, MigrationPage,
)
from weboob.browser import AbstractBrowser
class AferBrowser(LoginBrowser):
class AferBrowser(AbstractBrowser):
PARENT = 'aviva'
PARENT_ATTR = 'package.browser.AvivaBrowser'
BASEURL = 'https://adherent.gie-afer.fr'
login = URL(r'/espaceadherent/MonCompte/Connexion$', LoginPage)
wrong_password = URL(r'/espaceadherent/MonCompte/Connexion\?err=6001', WrongPasswordPage)
wrong_website = URL(r'/espaceadherent/MonCompte/Connexion\?err=6008', WrongWebsitePage)
migration = URL(r'/espaceadherent/MonCompte/Migration', MigrationPage)
index = URL('/web/ega.nsf/listeAdhesions\?OpenForm', IndexPage)
account_detail = URL('/web/ega.nsf/soldeEpargne\?openForm', AccountDetailPage)
account_history = URL('/web/ega.nsf/generationSearchModule\?OpenAgent', AccountHistoryPage)
history_detail = URL('/web/ega.nsf/WOpendetailOperation\?OpenAgent', AccountHistoryPage)
def do_login(self):
assert isinstance(self.username, basestring)
assert isinstance(self.password, basestring)
self.login.go()
try:
self.page.login(self.username, self.password)
except BrowserUnavailable:
raise BrowserIncorrectPassword()
if self.migration.is_here():
raise BrowserPasswordExpired(self.page.get_error())
if self.wrong_password.is_here():
error = self.page.get_error()
if error:
raise BrowserIncorrectPassword(error)
assert False, 'We landed on WrongPasswordPage but no error message was fetched.'
@need_login
def iter_accounts(self):
self.index.stay_or_go()
return self.page.iter_accounts()
@need_login
def iter_investment(self, account):
self.account_detail.go(params={'nads': account.id})
return self.page.iter_investment()
@need_login
def iter_history(self, account):
al = randint(0, 1000)
data = {'cdeAdh': account.id, 'al': al, 'page': 1, 'form': 'F'}
self.account_history.go(data={'cdeAdh': account.id, 'al': al, 'page': 1, 'form': 'F'})
return self.page.iter_history(data=data)
def __init__(self, *args, **kwargs):
self.subsite = 'espaceadherent'
super(AferBrowser, self).__init__(*args, **kwargs)
......@@ -19,10 +19,8 @@
from __future__ import unicode_literals
from weboob.capabilities.base import find_object
from weboob.capabilities.bank import CapBankWealth, AccountNotFound
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword
from weboob.capabilities.bank import CapBankWealth
from weboob.tools.backend import AbstractModule
from .browser import AferBrowser
......@@ -30,38 +28,20 @@
__all__ = ['AferModule']
class AferModule(Module, CapBankWealth):
class AferModule(AbstractModule, CapBankWealth):
NAME = 'afer'
DESCRIPTION = u'Association française d\'épargne et de retraite'
MAINTAINER = u'James GALT'
EMAIL = 'jgalt@budget-insight.com'
DESCRIPTION = "Association française d'épargne et de retraite"
MAINTAINER = 'Quentin Defenouillère'
EMAIL = 'quentin.defenouillere@budget-insight.com'
LICENSE = 'LGPLv3+'
VERSION = '1.6'
PARENT = 'aviva'
BROWSER = AferBrowser
CONFIG = BackendConfig(
ValueBackendPassword('login', label='Identifiant', regexp=r'.+', masked=False),
ValueBackendPassword('password', label="Mot de passe", regexp=r'\d{1,8}|[a-zA-Z0-9]{7,30}')
# TODO lose previous regex (and in backend) once users credentials migration is complete
)
def create_default_browser(self):
return self.create_browser(self.config['login'].get(),
self.config['password'].get())
def get_account(self, id):
return find_object(self.iter_accounts(), id=id, error=AccountNotFound)
def iter_accounts(self):
return self.browser.iter_accounts()
def iter_coming(self, account):
raise NotImplementedError()
def iter_history(self, account):
return self.browser.iter_history(account)
def iter_investment(self, account):
return self.browser.iter_investment(account)
return self.create_browser(
self.config['login'].get(),
self.config['password'].get(),
weboob=self.weboob
)
# -*- coding: utf-8 -*-
# Copyright(C) 2015 James GALT
#
# 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 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 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from random import randint
import requests
from weboob.browser.elements import ListElement, ItemElement, method
from weboob.browser.filters.standard import CleanText, Regexp, CleanDecimal, Date, Async, BrowserURL
from weboob.browser.pages import HTMLPage, LoggedPage, pagination
from weboob.capabilities.bank import Account, Investment, Transaction
from weboob.capabilities.base import NotAvailable
from weboob.exceptions import BrowserUnavailable, ActionNeeded
class LoginPage(HTMLPage):
def login(self, login, passwd):
form = self.get_form(id='loginForm')
form['username'] = login
form['password'] = passwd
form.submit()
class WrongPasswordPage(HTMLPage):
def get_error(self):
return CleanText('//p[contains(text(), "Votre saisie est erronée")]')(self.doc)
class WrongWebsitePage(HTMLPage):
# We land on this page when the website indicates that
# an account is already created on the 'Aviva et moi' space,
# So we check the message and raise ActionNeeded with it
def on_load(self):
message = CleanText('//p[contains(text(), "Vous êtes déjà inscrit")]')(self.doc)
if message:
raise ActionNeeded(message)
assert False, 'We landed on WrongWebsitePage but no message was fetched.'
class MigrationPage(HTMLPage):
def get_error(self):
return CleanText('//h1[contains(text(), "Votre nouvel identifiant et mot de passe")]')(self.doc)
class IndexPage(LoggedPage, HTMLPage):
def on_load(self):
HTMLPage.on_load(self)
msg = CleanText('//div[has-class("form-input-label")]', default='')(self.doc)
if "prendre connaissance des nouvelles conditions" in msg:
raise ActionNeeded(msg)
msg = CleanText('//span[@id="txtErrorAccesBase"]')(self.doc)
if 'Merci de nous envoyer' in msg:
raise ActionNeeded(msg)
# website sometime crash
if self.doc.xpath(u'//div[@id="divError"]/span[contains(text(),"Une erreur est survenue")]'):
raise BrowserUnavailable()
def is_here(self):
return bool(self.doc.xpath('//img[contains(@src, "deconnexion.jpg")]'))
@method
class iter_accounts(ListElement):
item_xpath = '//div[@id="adhesions"]/table//tr[td//a]'
class item(ItemElement):
klass = Account
obj_id = CleanText('.//a')
obj_label = CleanText('.//td[3]')
obj_currency = u'EUR'
def obj_balance(self):
if not '%' in CleanText('.//td[last()-2]')(self):
return CleanDecimal('.//td[last()-2]', replace_dots=True)(self)
elif not '%' in CleanText('.//td[last()-3]')(self):
return CleanDecimal('.//td[last()-3]', replace_dots=True)(self)
else:
return NotAvailable
obj_type = Account.TYPE_LIFE_INSURANCE
class AccountDetailPage(LoggedPage, HTMLPage):
def is_here(self):
return bool(self.doc.xpath('//*[@id="linkadhesion"]/a'))
@method
class iter_investment(ListElement):
item_xpath = '//div[@id="savingBalance"]/table[1]//tr'
class item(ItemElement):
def condition(self):
return self.el.xpath('./td[contains(@class,"dateUC")]')
klass = Investment
obj_label = CleanText('.//td[1]')
obj_code = NotAvailable
obj_description = NotAvailable
obj_quantity = CleanDecimal('.//td[7]', replace_dots=True, default=NotAvailable)
obj_unitvalue = CleanDecimal('.//td[5]', replace_dots=True, default=NotAvailable)
obj_valuation = CleanDecimal('.//td[3]', replace_dots=True)
obj_vdate = Date(Regexp(CleanText('.//td[2]'), '(((0[1-9]|[12][0-9]|3[01])[- /.]'
'(0[13578]|1[02])|(0[1-9]|[12][0-9]|30)[- /.](0[469]|11)|'
'(0[1-9]|1\d|2[0-8])[- /.]02)[- /.]\d{4}|29[- /.]02[- /.]'
'(\d{2}(0[48]|[2468][048]|[13579][26])|([02468][048]|'
'[1359][26])00))$', nth=0), dayfirst=True)
def obj_unitprice(self):
try:
return CleanDecimal(replace_dots=True, default=NotAvailable).filter(
self.el.xpath('.//td[6]')[0].text) / \
CleanDecimal(replace_dots=True, default=NotAvailable).filter(
self.el.xpath('.//td[7]')[0].text)
except TypeError:
return NotAvailable
def obj_diff(self):
try:
return self.obj.valuation - (self.obj.unitprice * self.obj.quantity)
except TypeError:
return NotAvailable
class AccountHistoryPage(LoggedPage, HTMLPage):
@pagination
@method
class iter_history(ListElement):
item_xpath = '//table[@class="confirmation-table"]//tr'
def next_page(self):
if self.page.doc.xpath("//table[@id='tabpage']//td"):
array_page = self.page.doc.xpath("//table[@id='tabpage']//td")[0][3].text
curr_page, max_page = array_page.split(' ')[1::2]
if int(curr_page) < int(max_page):
data = self.env['data']
data['page'] += 1
data['al'] = randint(1, 1000)
return requests.Request("POST", self.page.url, data=data)
return
class item(ItemElement):
condition = lambda self: len(self.el.xpath('./td')) >= 3
def load_details(self):
a = self.el.xpath(".//img[@src='../../images/commun/loupe.png']")
if len(a) > 0:
values = a[0].get('onclick').replace('OpenDetailOperation(', '') \
.replace(')', '').replace(' ', '').replace("'", '').split(',')
keys = ["nummvt", "&numads", "dtmvt", "typmvt"]
data = dict(zip(keys, values))
url = BrowserURL('history_detail')(self)
r = self.page.browser.async_open(url=url, data=data)
return r
return None
klass = Transaction
obj_date = obj_rdate = obj_vdate = Date(CleanText('.//td[3]'), dayfirst=True)
obj_label = CleanText('.//td[1]')
def obj_amount(self):
am = CleanDecimal('.//td[2]', replace_dots=True, default=NotAvailable)(self)
if am is not NotAvailable:
return am
return (Async('details') & CleanDecimal('//div//tr[2]/td[2]', replace_dots=True, default=NotAvailable))(
self)
......@@ -18,8 +18,7 @@
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from weboob.capabilities.bank import CapBank, AccountNotFound
from weboob.capabilities.base import find_object
from weboob.capabilities.bank import CapBank
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword
......@@ -47,9 +46,6 @@ def create_default_browser(self):
def iter_accounts(self):
return self.browser.get_accounts_list()
def get_account(self, _id):
return find_object(self.browser.get_accounts_list(), id=_id, error=AccountNotFound)
def iter_history(self, account):
return self.browser.iter_history(account)
......
......@@ -17,11 +17,19 @@
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from .pages import LoginPage, AccountsPage, AccountHistoryPage
from __future__ import unicode_literals
from weboob.browser import URL, LoginBrowser, need_login
from weboob.exceptions import BrowserIncorrectPassword
from weboob.browser.exceptions import ClientError
from weboob.capabilities.base import empty
from weboob.browser.exceptions import ClientError, ServerError
from weboob.capabilities.base import empty, NotAvailable
from .pages import (
LoginPage, AccountsPage, AccountHistoryPage, AmundiInvestmentsPage, AllianzInvestmentPage,
EEInvestmentPage, EEInvestmentDetailPage, EEProductInvestmentPage, EresInvestmentPage,
CprInvestmentPage, BNPInvestmentPage, BNPInvestmentApiPage, AxaInvestmentPage,
EpsensInvestmentPage, EcofiInvestmentPage,
)
class AmundiBrowser(LoginBrowser):
......@@ -32,6 +40,29 @@ class AmundiBrowser(LoginBrowser):
accounts = URL(r'api/individu/positionFonds\?flagUrlFicheFonds=true&inclurePositionVide=false', AccountsPage)
account_history = URL(r'api/individu/operations\?valeurExterne=false&filtreStatutModeExclusion=false&statut=CPTA', AccountHistoryPage)
# Amundi.fr investments
amundi_investments = URL(r'https://www.amundi.fr/fr_part/product/view', AmundiInvestmentsPage)
# EEAmundi browser investments
ee_investments = URL(r'https://www.amundi-ee.com/part/home_fp&partner=PACTEO_SYS', EEInvestmentPage)
ee_investment_details = URL(r'https://www.amundi-ee.com/psAmundiEEPart/ezjscore/call', EEInvestmentDetailPage)
# EEAmundi product investments
ee_product_investments = URL(r'https://www.amundi-ee.com/product', EEProductInvestmentPage)
# Allianz GI investments
allianz_investments = URL(r'https://fr.allianzgi.com', AllianzInvestmentPage)
# Eres investments
eres_investments = URL(r'https://www.eres-group.com/eres/new_fiche_fonds.php', EresInvestmentPage)
# CPR asset management investments
cpr_investments = URL(r'https://www.cpr-am.fr/particuliers/product/view', CprInvestmentPage)
# BNP Paribas Epargne Retraite Entreprises
bnp_investments = URL(r'https://www.epargne-retraite-entreprises.bnpparibas.com/entreprises/fonds', BNPInvestmentPage)
bnp_investment_api = URL(r'https://www.epargne-retraite-entreprises.bnpparibas.com/api2/funds/overview/(?P<fund_id>.*)', BNPInvestmentApiPage)
# AXA investments
axa_investments = URL(r'https://(.*).axa-im.fr/fr/fund-page', AxaInvestmentPage)
# Epsens investments
epsens_investments = URL(r'https://www.epsens.com/information-financiere', EpsensInvestmentPage)
# Ecofi investments
ecofi_investments = URL(r'http://www.ecofi.fr/fr/fonds/dynamis-solidaire', EcofiInvestmentPage)
def do_login(self):
data = {
'username': self.username,
......@@ -61,9 +92,89 @@ def iter_investment(self, account):
return
headers = {'X-noee-authorization': 'noeprd %s' % self.token}
self.accounts.go(headers=headers)
ignored_urls = (
'www.sggestion-ede.com/product', # Going there leads to a 404
'www.assetmanagement.hsbc.com', # Information not accessible
'www.labanquepostale-am.fr/nos-fonds', # Nothing interesting there
)
handled_urls = (
'www.amundi.fr/fr_part', # AmundiInvestmentsPage
'www.amundi-ee.com/part/home_fp', # EEInvestmentPage
'www.amundi-ee.com/product', # EEProductInvestmentPage
'fr.allianzgi.com/fr-fr', # AllianzInvestmentPage
'www.eres-group.com/eres', # EresInvestmentPage
'www.cpr-am.fr/particuliers/product', # CprInvestmentPage
'www.epargne-retraite-entreprises.bnpparibas.com', # BNPInvestmentPage
'axa-im.fr/fr/fund-page', # AxaInvestmentPage
'www.epsens.com/information-financiere', # EpsensInvestmentPage
'www.ecofi.fr/fr/fonds/dynamis-solidaire', # EcofiInvestmentPage
)
for inv in self.page.iter_investments(account_id=account.id):
if inv._details_url:
# Only go to known details pages to avoid logout on unhandled pages
if any(url in inv._details_url for url in handled_urls):
self.fill_investment_details(inv)
else:
if not any(url in inv._details_url for url in ignored_urls):
# Not need to raise warning if the URL is already known and ignored
self.logger.warning('Investment details on URL %s are not handled yet.', inv._details_url)
inv.asset_category = NotAvailable
inv.recommended_period = NotAvailable
yield inv
@need_login
def fill_investment_details(self, inv):
# Going to investment details may lead to various websites.
# This method handles all the already encountered pages.
try:
self.location(inv._details_url)
except ServerError:
# Some URLs return a 500 even on the website
inv.asset_category = NotAvailable
inv.recommended_period = NotAvailable
return inv
# Pages with only asset category available
if (self.amundi_investments.is_here() or
self.allianz_investments.is_here() or
self.axa_investments.is_here()):
inv.asset_category = self.page.get_asset_category()
inv.recommended_period = NotAvailable
# Pages with asset category & recommended period
elif (self.eres_investments.is_here() or
self.cpr_investments.is_here() or
self.ee_product_investments.is_here() or
self.epsens_investments.is_here() or
self.ecofi_investments.is_here()):
self.page.fill_investment(obj=inv)
# Particular cases
elif self.ee_investments.is_here():
inv.recommended_period = self.page.get_recommended_period()
details_url = self.page.get_details_url()
if details_url:
self.location(details_url)
if self.ee_investment_details.is_here():
inv.asset_category = self.page.get_asset_category()
elif self.bnp_investments.is_here():
# We fetch the fund ID and get the attributes directly from the BNP-ERE API
fund_id = self.page.get_fund_id()
if fund_id:
# Specify the 'Accept' header otherwise the server returns WSDL instead of JSON
self.bnp_investment_api.go(fund_id=fund_id, headers={'Accept': 'application/json'})
self.page.fill_investment(obj=inv)
else:
self.logger.warning('Could not fetch the fund_id for BNP investment %s.', inv.label)
inv.asset_category = NotAvailable
inv.recommended_period = NotAvailable
return inv
@need_login
def iter_history(self, account):
headers = {'X-noee-authorization': 'noeprd %s' % self.token}
......@@ -73,8 +184,10 @@ def iter_history(self, account):
class EEAmundi(AmundiBrowser):
# Careful if you modify the BASEURL, also verify Amundi's Abstract modules
BASEURL = 'https://www.amundi-ee.com/psf/'
class TCAmundi(AmundiBrowser):
# Careful if you modify the BASEURL, also verify Amundi's Abstract modules
BASEURL = 'https://epargnants.amundi-tc.com/psf/'
......@@ -18,8 +18,7 @@
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from weboob.capabilities.bank import CapBankWealth, AccountNotFound
from weboob.capabilities.base import find_object
from weboob.capabilities.bank import CapBankWealth
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword, Value
......@@ -46,9 +45,6 @@ def create_default_browser(self):
self.BROWSER = b[self.config['website'].get()]
return self.create_browser(self.config['login'].get(), self.config['password'].get())
def get_account(self, id):
return find_object(self.iter_accounts(), id=id, error=AccountNotFound)
def iter_accounts(self):
return self.browser.iter_accounts()
......
......@@ -19,18 +19,21 @@
from __future__ import unicode_literals
import re
from datetime import datetime
from weboob.browser.elements import ItemElement, method, DictElement
from weboob.browser.filters.standard import (
CleanDecimal, Date, Field, CleanText, Env, Eval,
CleanDecimal, Date, Field, CleanText,
Env, Eval, Map, Regexp, Title,
)
from weboob.browser.filters.html import Attr
from weboob.browser.filters.json import Dict
from weboob.browser.pages import LoggedPage, JsonPage
from weboob.browser.pages import LoggedPage, JsonPage, HTMLPage
from weboob.capabilities.bank import Account, Investment, Transaction
from weboob.capabilities.base import NotAvailable, empty
from weboob.capabilities.base import NotAvailable
from weboob.exceptions import NoAccountsException
from weboob.tools.capabilities.bank.investments import is_isin_valid
from weboob.tools.capabilities.bank.investments import IsinCode, IsinType
class LoginPage(JsonPage):
......@@ -38,6 +41,14 @@ def get_token(self):
return Dict('token')(self.doc)
ACCOUNT_TYPES = {
'PEE': Account.TYPE_PEE,
'PEG': Account.TYPE_PEE,
'PEI': Account.TYPE_PEE,
'PERCO': Account.TYPE_PERCO,
'RSP': Account.TYPE_RSP,
}
class AccountsPage(LoggedPage, JsonPage):
def get_company_name(self):
json_list = Dict('listPositionsSalarieFondsDto')(self.doc)
......@@ -45,14 +56,6 @@ def get_company_name(self):
return json_list[0].get('nomEntreprise', NotAvailable)
return NotAvailable
ACCOUNT_TYPES = {
'PEE': Account.TYPE_PEE,
'PEG': Account.TYPE_PEE,
'PEI': Account.TYPE_PEE,
'PERCO': Account.TYPE_PERCO,
'RSP': Account.TYPE_RSP,
}
@method
class iter_accounts(DictElement):
def parse(self, el):
......@@ -72,9 +75,7 @@ def obj_number(self):
return '%s_%s' % (Field('id')(self), self.page.browser.username)
obj_currency = 'EUR'
def obj_type(self):
return self.page.ACCOUNT_TYPES.get(Dict('typeDispositif')(self), Account.TYPE_LIFE_INSURANCE)
obj_type = Map(Dict('typeDispositif'), ACCOUNT_TYPES, Account.TYPE_LIFE_INSURANCE)
def obj_label(self):
try:
......@@ -97,11 +98,20 @@ def find_elements(self):
class item(ItemElement):
klass = Investment
def condition(self):
# Some additional invests are present in the JSON but are not
# displayed on the website, besides they have no valuation,
# so we check the 'valuation' key before parsing them
return Dict('mtBrut', default=None)(self)
obj_label = Dict('libelleFonds')
obj_unitvalue = Dict('vl') & CleanDecimal
obj_quantity = Dict('nbParts') & CleanDecimal
obj_valuation = Dict('mtBrut') & CleanDecimal
obj_vdate = Date(Dict('dtVl'))
obj__details_url = Dict('urlFicheFonds', default=None)
obj_code = IsinCode(Dict('codeIsin', default=NotAvailable), default=NotAvailable)
obj_code_type = IsinType(Dict('codeIsin', default=NotAvailable))
def obj_srri(self):
srri = Dict('SRRI')(self)
......@@ -110,25 +120,14 @@ def obj_srri(self):
return NotAvailable
return int(srri)
def obj_code(self):
code = Dict('codeIsin', default=NotAvailable)(self)
if is_isin_valid(code):
return code
return NotAvailable
def obj_code_type(self):
if empty(Field('code')(self)):
return NotAvailable
return Investment.CODE_TYPE_ISIN
def obj_performance_history(self):
# The Amundi JSON only contains 1 year and 5 years performances.
# It seems that when a value is unavailable, they display '0.0' instead...
perfs = {}
if Dict('performanceUnAn', default=None)(self) not in (0.0, None):
perfs[1] = Eval(lambda x: x/100, CleanDecimal(Dict('performanceUnAn')))(self)
perfs[1] = Eval(lambda x: x / 100, CleanDecimal(Dict('performanceUnAn')))(self)
if Dict('performanceCinqAns', default=None)(self) not in (0.0, None):
perfs[5] = Eval(lambda x: x/100, CleanDecimal(Dict('performanceCinqAns')))(self)
perfs[5] = Eval(lambda x: x / 100, CleanDecimal(Dict('performanceCinqAns')))(self)
return perfs
......@@ -164,5 +163,111 @@ def iter_history(self, account):
tr.date = tr.rdate
tr.label = hist.get('libelleOperation') or hist['libelleCommunication']
tr.type = Transaction.TYPE_UNKNOWN
yield tr
class AmundiInvestmentsPage(LoggedPage, HTMLPage):
def get_asset_category(self):
# Descriptions are like 'Fonds d'Investissement - (ISIN: FR001018 - Action'
# Fetch the last words of the description (e.g. 'Action' or 'Diversifié')
return Regexp(
CleanText('//div[@class="amundi-fund-legend"]//strong'),
r' ([^-]+)$',
default=NotAvailable
)(self.doc)
class EEInvestmentPage(LoggedPage, HTMLPage):
def get_recommended_period(self):
return Title(CleanText('//label[contains(text(), "Durée minimum de placement")]/following-sibling::span', default=NotAvailable))(self.doc)
def get_details_url(self):
return Attr('//a[contains(text(), "Caractéristiques")]', 'data-href', default=None)(self.doc)
class EEInvestmentDetailPage(LoggedPage, HTMLPage):
def get_asset_category(self):
return CleanText('//label[contains(text(), "Classe d\'actifs")]/following-sibling::span', default=NotAvailable)(self.doc)
class EEProductInvestmentPage(LoggedPage, HTMLPage):
@method
class fill_investment(ItemElement):
obj_asset_category = CleanText('//span[contains(text(), "Classe")]/following-sibling::span[@class="valeur"][1]')
obj_recommended_period = CleanText('//span[contains(text(), "Durée minimum")]/following-sibling::span[@class="valeur"][1]')
class AllianzInvestmentPage(LoggedPage, HTMLPage):
def get_asset_category(self):
# The format may be a very short description, or be
# included between quotation marks within a paragraph
asset_category = CleanText('//div[contains(@class, "fund-summary")]//h3/following-sibling::div', default=NotAvailable)(self.doc)
m = re.search(r'« (.*) »', asset_category)
if m:
return m.group(1)
return asset_category
class EresInvestmentPage(LoggedPage, HTMLPage):
@method
class fill_investment(ItemElement):
obj_asset_category = CleanText('//li[span[contains(text(), "Classification")]]', children=False, default=NotAvailable)
obj_recommended_period = CleanText('//li[span[contains(text(), "Durée")]]', children=False, default=NotAvailable)
def obj_performance_history(self):
perfs = {}
if CleanDecimal.French('(//tr[th[text()="1 an"]]/td[1])[1]', default=None)(self):
perfs[1] = Eval(lambda x: x / 100, CleanDecimal.French('(//tr[th[text()="1 an"]]/td[1])[1]'))(self)
if CleanDecimal.French('(//tr[th[text()="3 ans"]]/td[1])[1]', default=None)(self):
perfs[3] = Eval(lambda x: x / 100, CleanDecimal.French('(//tr[th[text()="3 ans"]]/td[1])[1]'))(self)
if CleanDecimal.French('(//tr[th[text()="5 ans"]]/td[1])[1]', default=None)(self):
perfs[5] = Eval(lambda x: x / 100, CleanDecimal.French('(//tr[th[text()="5 ans"]]/td[1])[1]'))(self)
return perfs
class CprInvestmentPage(LoggedPage, HTMLPage):
@method
class fill_investment(ItemElement):
obj_srri = CleanText('//span[@class="active"]', default=NotAvailable)
obj_asset_category = CleanText('//div[contains(text(), "Classe d\'actifs")]//strong', default=NotAvailable)
obj_recommended_period = CleanText('//div[contains(text(), "Durée recommandée")]//strong', default=NotAvailable)
class BNPInvestmentPage(LoggedPage, HTMLPage):
def get_fund_id(self):
return Regexp(
CleanText('//script[contains(text(), "GLB_ProductId")]'),
r'GLB_ProductId = "(\w+)',
default=None
)(self.doc)
class BNPInvestmentApiPage(LoggedPage, JsonPage):
@method
class fill_investment(ItemElement):
obj_asset_category = Dict('Classification', default=NotAvailable)
obj_recommended_period = Dict('DureePlacement', default=NotAvailable)
class AxaInvestmentPage(LoggedPage, HTMLPage):
def get_asset_category(self):
return Title(CleanText('//th[contains(text(), "Classe")]/following-sibling::td'))(self.doc)
class EpsensInvestmentPage(LoggedPage, HTMLPage):
@method
class fill_investment(ItemElement):
obj_asset_category = CleanText('//div[div[span[contains(text(), "Classification")]]]/div[2]/span', default=NotAvailable)
obj_recommended_period = CleanText('//div[div[span[contains(text(), "Durée de placement")]]]/div[2]/span', default=NotAvailable)
class EcofiInvestmentPage(LoggedPage, HTMLPage):
@method
class fill_investment(ItemElement):
# Recommended period is actually an image so we extract the
# information from its URL such as '/Horizon/Horizon_5_ans.png'
obj_recommended_period = Regexp(
CleanText(Attr('//img[contains(@src, "/Horizon/")]', 'src', default=NotAvailable), replace=[(u'_', ' ')]),
r'\/Horizon (.*)\.png'
)
obj_asset_category = CleanText('//div[contains(text(), "Classification")]/following-sibling::div[1]', default=NotAvailable)
# -*- coding: utf-8 -*-
# Copyright(C) 2016 Edouard Lambert
# Copyright(C) 2012-2019 Budget Insight
#
# This file is part of a weboob module.
#
......@@ -18,11 +18,7 @@
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from weboob.tools.test import BackendTest
from .module import AvivaModule
class MaterielnetTest(BackendTest):
MODULE = 'materielnet'
def test_materielnet(self):
raise NotImplementedError()
__all__ = ['AvivaModule']
# -*- coding: utf-8 -*-
# Copyright(C) 2012-2019 Budget Insight
#
# 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 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 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.browser import LoginBrowser, need_login
from weboob.browser.url import BrowserParamURL
from weboob.capabilities.base import empty, NotAvailable
from weboob.exceptions import BrowserIncorrectPassword, BrowserPasswordExpired, ActionNeeded, BrowserHTTPError
from weboob.tools.capabilities.bank.transactions import sorted_transactions
from .pages.detail_pages import (
LoginPage, MigrationPage, InvestmentPage, HistoryPage, ActionNeededPage,
InvestDetailPage, PrevoyancePage, ValidationPage, InvestPerformancePage,
)
from .pages.account_page import AccountsPage
class AvivaBrowser(LoginBrowser):
BASEURL = 'https://www.aviva.fr'
validation = BrowserParamURL(r'/conventions/acceptation\?backurl=/(?P<browser_subsite>[^/]+)/Accueil', ValidationPage)
login = BrowserParamURL(
r'/(?P<browser_subsite>[^/]+)/MonCompte/Connexion',
r'/(?P<browser_subsite>[^/]+)/conventions/acceptation',
LoginPage
)
migration = BrowserParamURL(r'/(?P<browser_subsite>[^/]+)/MonCompte/Migration', MigrationPage)
accounts = BrowserParamURL(r'/(?P<browser_subsite>[^/]+)/Accueil/Synthese-Contrats', AccountsPage)
investment = BrowserParamURL(r'/(?P<browser_subsite>[^/]+)/contrat/epargne/-(?P<page_id>[0-9]{10})', InvestmentPage)
prevoyance = BrowserParamURL(r'/(?P<browser_subsite>[^/]+)/contrat/prevoyance/-(?P<page_id>[0-9]{10})', PrevoyancePage)
history = BrowserParamURL(r'/(?P<browser_subsite>[^/]+)/contrat/getOperations\?param1=(?P<history_token>.*)', HistoryPage)
action_needed = BrowserParamURL(r'/(?P<browser_subsite>[^/]+)/coordonnees/detailspersonne\?majcontacts=true', ActionNeededPage)
invest_detail = BrowserParamURL(r'https://aviva-fonds.webfg.net/sheet/fund/(?P<isin>[A-Z0-9]+)', InvestDetailPage)
invest_performance = BrowserParamURL(r'https://aviva-fonds.webfg.net/sheet/fund-calculator', InvestPerformancePage)
def __init__(self, *args, **kwargs):
self.subsite = 'espaceclient'
super(AvivaBrowser, self).__init__(*args, **kwargs)
def do_login(self):
self.login.go()
self.page.login(self.username, self.password)
if self.login.is_here():
if 'acceptation' in self.url:
raise ActionNeeded("Veuillez accepter les conditions générales d'utilisation sur le site.")
else:
raise BrowserIncorrectPassword("L'identifiant ou le mot de passe est incorrect.")
elif self.migration.is_here():
# Usually landing here when customers have to renew their credentials
message = self.page.get_error()
raise BrowserPasswordExpired(message)
@need_login
def iter_accounts(self):
self.accounts.go()
for account in self.page.iter_accounts():
# Request to account details sometimes returns a 500
try:
self.location(account.url)
if not self.investment.is_here() or self.page.unavailable_details():
# We don't scrape insurances, guarantees, health contracts
# and accounts with unavailable balances
continue
self.page.fill_account(obj=account)
yield account
except BrowserHTTPError:
self.logger.warning('Could not get the account details: account %s will be skipped', account.id)
@need_login
def iter_investment(self, account):
# Request to account details sometimes returns a 500
try:
self.location(account.url)
except BrowserHTTPError:
self.logger.warning('Could not get the account investments for account %s', account.id)
return
for inv in self.page.iter_investment():
if not empty(inv.code):
# Need to go first on InvestDetailPage...
self.invest_detail.go(isin=inv.code)
# ...to then request the InvestPerformancePage tab
self.invest_performance.go()
self.page.fill_investment(obj=inv)
else:
inv.unitprice = inv.diff_ratio = inv.description = NotAvailable
yield inv
@need_login
def iter_history(self, account):
if empty(account.url):
# An account should always have a link to the details
raise NotImplementedError()
try:
self.location(account.url)
except BrowserHTTPError:
self.logger.warning('Could not get the history for account %s', account.id)
return
history_link = self.page.get_history_link()
if not history_link:
# accounts don't always have an history_link
raise NotImplementedError()
self.location(history_link)
assert self.history.is_here()
result = []
result.extend(self.page.iter_versements())
result.extend(self.page.iter_arbitrages())
return sorted_transactions(result)
def get_subscription_list(self):
return []
# -*- coding: utf-8 -*-
# Copyright(C) 2015 Baptiste Delpey
# Copyright(C) 2012-2019 Budget Insight
#
# This file is part of a weboob module.
#
......@@ -17,12 +17,43 @@
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.tools.test import BackendTest
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword
from weboob.capabilities.bank import CapBankWealth
from .browser import AvivaBrowser
class BnpcartesentrepriseTest(BackendTest):
MODULE = 'bnpcartesentreprise'
def test_bnpcartesentreprise(self):
raise NotImplementedError()
__all__ = ['AvivaModule']
class AvivaModule(Module, CapBankWealth):
NAME = 'aviva'
DESCRIPTION = 'Aviva'
MAINTAINER = 'Edouard Lambert'
EMAIL = 'elambert@budget-insight.com'
LICENSE = 'LGPLv3+'
VERSION = '1.6'
CONFIG = BackendConfig(
ValueBackendPassword('login', label='Identifiant', masked=False),
ValueBackendPassword('password', label='Mot de passe')
)
BROWSER = AvivaBrowser
def create_default_browser(self):
return self.create_browser(
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)
# -*- coding: utf-8 -*-
# Copyright(C) 2016 Edouard Lambert
#
# 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 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 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from weboob.browser.pages import HTMLPage, LoggedPage
from weboob.browser.elements import ListElement, ItemElement, method
from weboob.browser.filters.standard import CleanText, Capitalize, Format, Date, Regexp, CleanDecimal, Env, Field, Async, AsyncLoad
from weboob.browser.filters.html import Attr, Link
from weboob.capabilities.bank import Account, Investment, Transaction
from weboob.capabilities.base import NotAvailable
from weboob.exceptions import ActionNeeded, BrowserUnavailable
def MyDecimal(*args, **kwargs):
kwargs.update(replace_dots=True, default=NotAvailable)
return CleanDecimal(*args, **kwargs)
class BasePage(HTMLPage):
def on_load(self):
super(BasePage, self).on_load()
if 'Erreur' in CleanText('//div[@id="main"]/h1', default='')(self.doc):
err = CleanText('//div[@id="main"]/div[@class="content"]', default=u'Site indisponible')(self.doc)
raise BrowserUnavailable(err)
class LoginPage(BasePage, HTMLPage):
def get_values(self):
values = {}
virtual_keyboard = self.doc.xpath('//input[@keypad]')
for input in virtual_keyboard:
value = Attr(None, 'value').filter(input.xpath('.'))
index = Attr(None, 'data-index').filter(input.xpath('.'))
values[value] = index
return values
def get_password(self, password):
values = self.get_values()
if values:
password = [values[c] for c in password]
return "".join(password)
def login(self, login, password):
form = self.get_form(id="loginForm")
form['username'] = login
form['password'] = self.get_password(password)
form.submit()
class AccountsPage(LoggedPage, BasePage, HTMLPage):
@method
class iter_accounts(ListElement):
item_xpath = '//div[@id]/div[has-class("productDetails")]'
class item(ItemElement):
klass = Account
load_details = Field('_link') & AsyncLoad
obj_id = CleanText('.//dt[contains(text(), "contrat")]/following-sibling::dd[1]')
obj_label = CleanText('.//div[has-class("title")]')
obj_type = Account.TYPE_LIFE_INSURANCE
obj_balance = Async('details') & MyDecimal('//strong[contains(text(), "Valeur") '
'or contains(text(), "Epargne retraite")]/following-sibling::span')
obj_valuation_diff = Async('details') & MyDecimal('.//a[contains(@title, "diff")]/parent::p')
obj__link = Link(u'.//a[contains(text(), "Détail")]')
# Additional waranty : need to know what to do with this
obj__additionalwaranty = Env('additionalwaranty')
def condition(self):
return 'epargne' in Link(u'.//a[contains(text(), "Détail")]')(self)
def parse(self, el):
# Additional waranty
doc = self.page.browser.open(Field('_link')(self)).page.doc
additionalwaranty = []
for line in doc.xpath(u'//h2[contains(text(), "complémentaire")]/following-sibling::div//div[@class="line"]'):
values = {}
values['label'] = CleanText().filter(line.xpath('./div[@data-label="Nom du support"]'))
values['amount'] = CleanText().filter(line.xpath('./div[@data-label="Montant total investi"]'))
additionalwaranty.append(values)
self.env['additionalwaranty'] = additionalwaranty
class InvestmentPage(LoggedPage, HTMLPage):
def get_history_link(self):
link = self.doc.xpath(u'//a[contains(text(), "historique")]/@href')
if link:
return link[0]
@method
class iter_investment(ListElement):
item_xpath = '//h2[has-class("feature")]/following-sibling::div/div[@class="line"]'
class item(ItemElement):
klass = Investment
def load_details(self):
# create URL with ISIN code if exists
code = Field('code')(self)
if code:
url = 'https://aviva.sixtelekurs.fr/opcvm.hts?isin=%s&environnement=internet' % code
return self.page.browser.async_open(url=url)
obj_code = Regexp(Attr('./div[@data-label="Nom du support"]/a', 'onclick', default=''), '\"([^\"]+)', default=NotAvailable)
obj_quantity = MyDecimal('./div[@data-label="Nombre de parts"]', default=NotAvailable)
obj_unitvalue = MyDecimal('./div[@data-label="Valeur de la part"]')
obj_valuation = MyDecimal('./div[@data-label="Valeur"]', default=NotAvailable)
obj_vdate = Date(CleanText('./div[@data-label="Date de valeur"]'), dayfirst=True, default=NotAvailable)
obj_unitprice = Async('details') & CleanDecimal('//td[@class="donnees"]/span[@id="VL_achat"]', default=NotAvailable)
obj_diff_percent = Async('details') & CleanDecimal('//td[@class="donnees"]/span[@id="Performance"]', default=NotAvailable)
obj_description = Async('details') & CleanText('//td[@class="donnees"]/span[@id="Nature"]', default=NotAvailable)
def obj_label(self):
# label xpath changes if there is a link to details
label = CleanText('./div[@data-label="Nom du support"]', children=False)(self)
if label:
return label
return CleanText('./div[@data-label="Nom du support"]/a', children=False)(self)
class InvestmentElement(ItemElement):
klass = Investment
obj_label = CleanText('./div[@data-label="Nom du support" or @data-label="Support cible"]/span[1]')
obj_quantity = MyDecimal('./div[contains(@data-label, "Nombre")]')
obj_unitvalue = MyDecimal('./div[contains(@data-label, "Valeur")]')
obj_valuation = MyDecimal('./div[contains(@data-label, "Montant")]')
obj_vdate = Env('date')
#obj_vdate = Field('date')
class TransactionElement(ItemElement):
klass = Transaction
obj_label = Format('%s du %s', Field('_labeltype'), Field('date'))
obj_date = Date(Regexp(CleanText('./ancestor::div[@class="onerow" or starts-with(@id, "term") or has-class("grid")]/'
'preceding-sibling::h3[1]//div[contains(text(), "Date")]'),
r'(\d{2}\/\d{2}\/\d{4})'),
dayfirst=True)
obj_type = Transaction.TYPE_BANK
obj_amount = MyDecimal('./ancestor::div[@class="onerow" or starts-with(@id, "term") or has-class("grid")]/'
'preceding-sibling::h3[1]//div[has-class("montant-mobile")]', default=NotAvailable)
obj__labeltype = Regexp(Capitalize('./preceding::h2[@class="feature"][1]'),
'Historique Des\s+(\w+)')
def obj_investments(self):
return list(self.iter_investments(self.page, parent=self))
@method
class iter_investments(ListElement):
item_xpath = './div[@class="line"]'
class item(InvestmentElement):
pass
def parse(self, el):
self.env['date'] = Field('date')(self)
class HistoryPage(LoggedPage, HTMLPage):
@method
class iter_history(ListElement):
class iter_simple(ListElement):
def find_elements(self):
# "arbitrage" transactions are split in 2 table-responsive
# the html tree is a mess
for sub in self.el.xpath('//div[not(@id="divAvance")]/div[@class="table-responsive"]'):
if (sub.getparent().attrib.get('class') != 'onerow' or
'grid' not in sub.getparent().getparent().attrib.get('class')):
yield sub
obj_date = Date(Regexp(CleanText('./ancestor::div[starts-with(@id, "term")]/'
'preceding-sibling::h3[1]//div[contains(text(), "Date")]'),
r'\d{2}\/\d{2}\/\d{4}'),
dayfirst=True)
class item(TransactionElement):
class iter_investments(ListElement):
item_xpath = './div[@class="line"]'
class item(InvestmentElement):
pass
class iter_complex(ListElement):
item_xpath = '//div[has-class("grid")]/div[@class="onerow"]'
class item(TransactionElement):
class iter_investments(ListElement):
item_xpath = './div[@class="table-responsive"]/div[@class="line"]'
class item(InvestmentElement):
pass
class iter_avance(ListElement):
item_xpath = '//div[@id="divAvance"]/div[@class="table-responsive"]/div[@class="line"]'
class item(ItemElement):
klass = Transaction
obj_label = Format('%s du %s', Field('_labeltype'), Field('date'))
obj_type = Transaction.TYPE_BANK
obj_date = Date(CleanText(u'./div[@data-label="Date d\'effet"]', children=False), dayfirst=True)
obj_amount = CleanDecimal(u'./div[@data-label="Montant en €"]', replace_dots=True)
obj__labeltype = Regexp(Capitalize('./preceding::h2[@class="feature"][1]'),
'Historique Des\s+(\w+)')
class ActionNeededPage(LoggedPage, HTMLPage):
def on_load(self):
raise ActionNeeded(u'Veuillez mettre à jour vos coordonnées')
class InvestDetail(LoggedPage, HTMLPage):
pass
# -*- coding: utf-8 -*-
# Copyright(C) 2012-2019 Budget Insight
#
# 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 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 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.browser.pages import LoggedPage
from weboob.browser.elements import ListElement, ItemElement, method
from weboob.browser.filters.standard import (
CleanText, Field, Map, Regexp
)
from weboob.browser.filters.html import AbsoluteLink
from weboob.capabilities.bank import Account
from weboob.capabilities.base import NotAvailable
from .detail_pages import BasePage
ACCOUNT_TYPES = {
'Assurance vie': Account.TYPE_LIFE_INSURANCE,
'Epargne – Retraite': Account.TYPE_PERP,
}
class AccountsPage(LoggedPage, BasePage):
@method
class iter_accounts(ListElement):
item_xpath = '//div[contains(@class, "o-product-roundels")]/div[@data-policy]'
class item(ItemElement):
klass = Account
obj_id = CleanText('./@data-policy')
obj_number = Field('id')
obj_label = CleanText('.//p[has-class("a-heading")]', default=NotAvailable)
obj_url = AbsoluteLink('.//a[contains(text(), "Détail")]')
obj_type = Map(Regexp(CleanText('../../../div[contains(@class, "o-product-roundels-category")]'),
r'Vérifier votre (.*) contrats', default=NotAvailable),
ACCOUNT_TYPES, Account.TYPE_UNKNOWN)
def condition(self):
# 'Prévoyance' div is for insurance contracts -- they are not bank accounts and thus are skipped
ignored_accounts = (
'Prévoyance', 'Responsabilité civile', 'Complémentaire santé', 'Protection juridique',
'Habitation', 'Automobile',
)
return CleanText('../../div[has-class("o-product-tab-category")]', default=NotAvailable)(self) not in ignored_accounts
# -*- coding: utf-8 -*-
# Copyright(C) 2012-2019 Budget Insight
#
# 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 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 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.browser.pages import HTMLPage, LoggedPage
from weboob.browser.elements import ListElement, ItemElement, method
from weboob.browser.filters.standard import (
CleanText, Title, Format, Date, Regexp, CleanDecimal, Env,
Currency, Field, Eval, Coalesce,
)
from weboob.capabilities.bank import Investment, Transaction
from weboob.capabilities.base import NotAvailable
from weboob.exceptions import ActionNeeded, BrowserUnavailable
from weboob.tools.compat import urljoin
from weboob.tools.capabilities.bank.investments import IsinCode, IsinType
class BasePage(HTMLPage):
def on_load(self):
super(BasePage, self).on_load()
if 'Erreur' in CleanText('//div[@id="main"]/h1', default='')(self.doc):
err = CleanText('//div[@id="main"]/div[@class="content"]', default='Site indisponible')(self.doc)
raise BrowserUnavailable(err)
class PrevoyancePage(LoggedPage, HTMLPage):
pass
class LoginPage(BasePage):
def login(self, login, password):
form = self.get_form(id="loginForm")
form['username'] = login
form['password'] = password
form.submit()
class MigrationPage(LoggedPage, HTMLPage):
def get_error(self):
return CleanText('//h1[contains(text(), "Votre nouvel identifiant et mot de passe")]')(self.doc)
class InvestmentPage(LoggedPage, HTMLPage):
@method
class fill_account(ItemElement):
obj_balance = CleanDecimal.French('//ul[has-class("m-data-group")]//strong')
obj_currency = Currency('//ul[has-class("m-data-group")]//strong')
obj_valuation_diff = CleanDecimal.French('//h3[contains(., "value latente")]/following-sibling::p[1]', default=NotAvailable)
def get_history_link(self):
history_link = self.doc.xpath('//li/a[contains(text(), "Historique")]/@href')
return urljoin(self.browser.BASEURL, history_link[0]) if history_link else ''
def unavailable_details(self):
return CleanText('//p[contains(text(), "est pas disponible")]')(self.doc)
@method
class iter_investment(ListElement):
item_xpath = '(//div[contains(@class, "m-table")])[1]//table/tbody/tr[not(contains(@class, "total"))]'
class item(ItemElement):
klass = Investment
def condition(self):
return Field('label')(self) not in ('Total', '')
obj_quantity = CleanDecimal.French('./td[contains(@data-label, "Nombre de parts")]', default=NotAvailable)
obj_unitvalue = CleanDecimal.French('./td[contains(@data-label, "Valeur de la part")]', default=NotAvailable)
def obj_valuation(self):
# Handle discrepancies between aviva & afer (Coalesce does not work here)
if CleanText('./td[contains(@data-label, "Valeur de rachat")]')(self):
return CleanDecimal.French('./td[contains(@data-label, "Valeur de rachat")]')(self)
return CleanDecimal.French(CleanText('./td[contains(@data-label, "Montant")]', children=False))(self)
obj_vdate = Date(
CleanText('./td[@data-label="Date de valeur"]'), dayfirst=True, default=NotAvailable
)
obj_label = Coalesce(
CleanText('./th[@data-label="Nom du support"]/a'),
CleanText('./th[@data-label="Nom du support"]'),
CleanText('./td[@data-label="Nom du support"]'),
)
obj_code = IsinCode(
Regexp(
CleanText('./td[@data-label="Nom du support"]/a/@onclick|./th[@data-label="Nom du support"]/a/@onclick'),
r'"(.*)"',
default=NotAvailable
),
default=NotAvailable
)
obj_code_type = IsinType(Field('code'))
class TransactionElement(ItemElement):
klass = Transaction
obj_label = Format('%s du %s', Field('_labeltype'), Field('date'))
obj_date = Date(
Regexp(
CleanText(
'./ancestor::div[@class="onerow" or starts-with(@id, "term") or has-class("grid")]/'
'preceding-sibling::h3[1]//div[contains(text(), "Date")]'
),
r'(\d{2}\/\d{2}\/\d{4})'),
dayfirst=True
)
obj_type = Transaction.TYPE_BANK
obj_amount = CleanDecimal.French(
'./ancestor::div[@class="onerow" or starts-with(@id, "term") or has-class("grid")]/'
'preceding-sibling::h3[1]//div[has-class("montant-mobile")]',
default=NotAvailable
)
obj__labeltype = Regexp(
Title('./preceding::h2[@class="feature"][1]'),
r'Historique Des\s+(\w+)'
)
def obj_investments(self):
return list(self.iter_investments(self.page, parent=self))
@method
class iter_investments(ListElement):
item_xpath = './div[@class="line"]'
class item(ItemElement):
klass = Investment
obj_label = CleanText('./div[@data-label="Nom du support" or @data-label="Support cible"]/span[1]')
obj_quantity = CleanDecimal.French('./div[contains(@data-label, "Nombre")]', default=NotAvailable)
obj_unitvalue = CleanDecimal.French('./div[contains(@data-label, "Valeur")]', default=NotAvailable)
obj_valuation = CleanDecimal.French('./div[contains(@data-label, "Montant")]', default=NotAvailable)
obj_vdate = Env('date')
def parse(self, el):
self.env['date'] = Field('date')(self)
class HistoryPage(LoggedPage, HTMLPage):
@method
class iter_versements(ListElement):
item_xpath = '//div[contains(@id, "versementProgramme3") or contains(@id, "versementLibre3")]/h2'
class item(ItemElement):
klass = Transaction
obj_date = Date(
Regexp(CleanText('./div[1]'), r'(\d{2}\/\d{2}\/\d{4})'),
dayfirst=True
)
obj_amount = Eval(lambda x: x / 100, CleanDecimal('./div[2]'))
obj_label = Format(
'%s %s',
CleanText('./preceding::h3[1]'),
Regexp(CleanText('./div[1]'), r'(\d{2}\/\d{2}\/\d{4})')
)
def obj_investments(self):
investments = []
for elem in self.xpath('./following-sibling::div[1]//ul'):
inv = Investment()
inv.label = CleanText('./li[1]/p')(elem)
inv.portfolio_share = CleanDecimal('./li[2]/p', replace_dots=True, default=NotAvailable)(elem)
inv.quantity = CleanDecimal('./li[3]/p', replace_dots=True, default=NotAvailable)(elem)
inv.valuation = CleanDecimal('./li[4]/p', replace_dots=True)(elem)
investments.append(inv)
return investments
@method
class iter_arbitrages(ListElement):
item_xpath = '//div[contains(@id, "arbitrageLibre3")]/h2'
class item(ItemElement):
klass = Transaction
obj_date = Date(
Regexp(CleanText('.//div[1]'), r'(\d{2}\/\d{2}\/\d{4})'),
dayfirst=True
)
obj_label = Format(
'%s %s',
CleanText('./preceding::h3[1]'),
Regexp(CleanText('./div[1]'), r'(\d{2}\/\d{2}\/\d{4})')
)
def obj_amount(self):
return sum(x.valuation for x in Field('investments')(self))
def obj_investments(self):
investments = []
for elem in self.xpath('./following-sibling::div[1]//tbody/tr'):
inv = Investment()
inv.label = CleanText('./td[1]')(elem)
inv.valuation = Coalesce(
CleanDecimal.French('./td[2]/p', default=NotAvailable),
CleanDecimal.French('./td[2]')
)(elem)
investments.append(inv)
return investments
class ActionNeededPage(LoggedPage, HTMLPage):
def on_load(self):
raise ActionNeeded('Veuillez mettre à jour vos coordonnées')
class ValidationPage(LoggedPage, HTMLPage):
def on_load(self):
error_message = CleanText('//p[@id="errorSigned"]')(self.doc)
if error_message:
raise ActionNeeded(error_message)
class InvestDetailPage(LoggedPage, HTMLPage):
pass
class InvestPerformancePage(LoggedPage, HTMLPage):
@method
class fill_investment(ItemElement):
obj_unitprice = CleanDecimal.US('//span[contains(@data-module-target, "BuyValue")]')
obj_description = CleanText('//td[contains(text(), "Nature")]/following-sibling::td')
obj_diff_ratio = Eval(
lambda x: x / 100 if x else NotAvailable,
CleanDecimal.US('//span[contains(@data-module-target, "trhrthrth")]', default=NotAvailable)
)
# -*- coding: utf-8 -*-
# Copyright(C) 2012-2019 Budget Insight
#
# 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 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 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 Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from decimal import Decimal
from weboob.tools.test import BackendTest
from weboob.capabilities.base import empty
from weboob.tools.capabilities.bank.test import BankStandardTest
class AvivaTest(BackendTest, BankStandardTest):
MODULE = 'aviva'
def test_iter_accounts(self):
account_list = list(self.backend.iter_accounts())
# check unicity of the account numbers
self.assertEqual(
len(account_list),
len({account.number for account in account_list})
)
# check unicity of the account ids
self.assertEqual(
len(account_list),
len({account.id for account in account_list})
)
for account in account_list:
self.assertTrue(account.label)
def test_iter_investment(self):
account_list = list(self.backend.iter_accounts())
for account in account_list:
investments = list(self.backend.iter_investment(account))
self.assertEqual(
sum([invest.portfolio_share for invest in investments]),
Decimal('1.00')
)
for investment in investments:
self.assertFalse(empty(investment.vdate))
self.assertTrue(investment.vdate)
def test_iter_history(self):
account_list = list(self.backend.iter_accounts())
for account in account_list:
history = list(self.backend.iter_history(account))
self.assertTrue(
sum([transaction.value for transaction in history]),
Decimal('1.00')
)
for transaction in history:
self.assertTrue(transaction.amount)
......@@ -46,11 +46,11 @@ class AXABanqueModule(Module, CapBankWealth, CapBankTransferAddRecipient, CapDoc
def create_default_browser(self):
login = self.config['login'].get()
self.BROWSER = AXABanque if login.isdigit() else AXAAssurance
return self.create_browser(login, self.config['password'].get(),
weboob=self.weboob)
def get_account(self, _id):
return find_object(self.browser.iter_accounts(), id=_id, error=AccountNotFound)
return self.create_browser(
login,
self.config['password'].get(),
weboob=self.weboob
)
def iter_accounts(self):
return self.browser.iter_accounts()
......
......@@ -27,7 +27,7 @@
from dateutil.relativedelta import relativedelta
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable
from weboob.browser.exceptions import HTTPNotFound, ServerError
from weboob.browser.exceptions import HTTPNotFound, ClientError, ServerError
from weboob.browser import LoginBrowser, URL, need_login
from weboob.capabilities.bank import Account, AccountOwnership
from weboob.capabilities.base import NotAvailable, find_object
......@@ -219,7 +219,16 @@ def follow_back_button_if_any(self, params=None, actions=None):
@no_need_login
def do_login(self):
self.location(self.BASEURL)
try:
self.location(self.BASEURL)
except (ClientError, HTTPNotFound) as e:
if e.response.status_code in (403, 404):
# Sometimes the website makes some redirections that leads
# to a 404 or a 403 when we try to access the BASEURL
# (website is not stable).
raise BrowserUnavailable(e.message)
raise
# avoids trying to relog in while it's already on home page
if self.home_page.is_here():
return
......
......@@ -31,7 +31,6 @@
from weboob.capabilities.base import NotAvailable, empty
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.tools.capabilities.bank.investments import is_isin_valid
from weboob.browser.filters.standard import FilterError
def MyDecimal(*args, **kwargs):
......@@ -213,7 +212,6 @@ class item(ItemElement):
obj_label = Dict('SecurityName')
obj_quantity = MyDecimal(Dict('Quantity'))
obj_vdate = DateTime(CleanText(Dict('Time')), dayfirst=True, strict=False)
obj_unitvalue = Env('unitvalue', default=NotAvailable)
obj_unitprice = Env('unitprice', default=NotAvailable)
obj_valuation = MyDecimal(Dict('ValueInEuro'))
......@@ -227,14 +225,17 @@ class item(ItemElement):
obj__security_id = Dict('SecurityId')
def obj_vdate(self):
try:
# during stocks closing hours only date (d/m/y) is given
raw_date = CleanText(Dict('Time'))(self)
if raw_date == '---':
return NotAvailable
elif re.match(r'\d{2}\/\d{2}\/\d{4}', raw_date):
# during stocks closing hours only date (dd/mm/yyyy) is given
return Date(CleanText(Dict('Time')), dayfirst=True)(self)
except FilterError:
# during stocks opening hours only time (h:m:s) is given,
elif re.match(r'\d{2}:\d{2}:\d{2}', raw_date):
# during stocks opening hours only time (hh:mm:ss) is given,
# can even be realtime, depending on user settings,
# can be given in foreign markets time,
# e.g. 10:00 is displayed at 9:00 for an action in NASADQ Helsinki market
# e.g. 10:00 is displayed at 9:00 for an action in NASDAQ Helsinki market
return DateTime(CleanText(Dict('Time')), dayfirst=True, strict=False)(self)
def obj_code(self):
......
......@@ -24,8 +24,10 @@
from weboob.tools.capabilities.bank.transactions import sorted_transactions
from weboob.tools.compat import basestring
from .pages import LoginPage, ErrorPage, AccountsPage, TransactionsPage, \
TiCardPage, TiHistoPage, ComingPage, HistoPage, HomePage
from .pages import (
LoginPage, ErrorPage, AccountsPage, TransactionsPage,
TiCardPage, TiHistoPage, ComingPage, HistoPage, HomePage,
)
class BnpcartesentrepriseBrowser(LoginBrowser):
......
......@@ -32,32 +32,41 @@
class BnpcartesentrepriseCorporateBrowser(LoginBrowser):
BASEURL = 'https://www.cartesentreprise.bnpparibas.com'
login = URL('/ce_internet_public/seConnecter.builder.do', LoginPage)
error = URL('.*.seConnecter.event.do',
'.*.compteGestChgPWD.builder.do',
'/ce_internet_prive_ti/compteTituChgPWD.builder.do',
ErrorPage)
acc_home = URL('/ce_internet_prive_ge/carteCorporateParc.builder.do', AccountsPage)
accounts_page = URL('/ce_internet_prive_ge/operationVotreParcAfficherCorporate.event.do',
'/ce_internet_prive_ge/operationVotreParcAppliquerCorporate.event.do.*',
AccountsPage)
com_home = URL('/ce_internet_prive_ge/operationCorporateEnCours.builder.do', AccountsPage)
coming = URL('/ce_internet_prive_ge/operationEnCoursAfficherCorporate.event.do',
'/ce_internet_prive_ge/operationEnCoursAppliquerCorporate.event.do.*',
'ce_internet_prive_ge/operationEnCoursDetailAppliquerCorporate.event.do.*',
AccountsPage)
login = URL(r'/ce_internet_public/seConnecter.builder.do', LoginPage)
error = URL(
r'.*.seConnecter.event.do',
r'.*.compteGestChgPWD.builder.do',
r'/ce_internet_prive_ti/compteTituChgPWD.builder.do',
ErrorPage,
)
acc_home = URL(r'/ce_internet_prive_ge/carteCorporateParc.builder.do', AccountsPage)
accounts_page = URL(
r'/ce_internet_prive_ge/operationVotreParcAfficherCorporate.event.do',
r'/ce_internet_prive_ge/operationVotreParcAppliquerCorporate.event.do.*',
AccountsPage,
)
com_home = URL(r'/ce_internet_prive_ge/operationCorporateEnCours.builder.do', AccountsPage)
coming = URL(
r'/ce_internet_prive_ge/operationEnCoursAfficherCorporate.event.do',
r'/ce_internet_prive_ge/operationEnCoursAppliquerCorporate.event.do.*',
r'ce_internet_prive_ge/operationEnCoursDetailAppliquerCorporate.event.do.*',
AccountsPage,
)
his_home = URL('/ce_internet_prive_ge/operationCorporateHisto.builder.do', AccountsPage)
history = URL('/ce_internet_prive_ge/operationHistoriqueAfficherCorporate.event.do',
'/ce_internet_prive_ge/operationHistoriqueAppliquerCorporate.event.do.*',
AccountsPage)
transactions = URL('ce_internet_prive_ge/operationCorporateEnCoursDetail.builder.do.*',
'ce_internet_prive_ge/operationCorporateDetailHistorique.builder.do.*',
'ce_internet_prive_ge/operationDetail.*AppliquerCorporate.event.do.*',
TransactionsPage)
history = URL(
r'/ce_internet_prive_ge/operationHistoriqueAfficherCorporate.event.do',
r'/ce_internet_prive_ge/operationHistoriqueAppliquerCorporate.event.do.*',
AccountsPage,
)
transactions = URL(
r'ce_internet_prive_ge/operationCorporateEnCoursDetail.builder.do.*',
r'ce_internet_prive_ge/operationCorporateDetailHistorique.builder.do.*',
r'ce_internet_prive_ge/operationDetail.*AppliquerCorporate.event.do.*',
TransactionsPage,
)
TIMEOUT = 60.0
def __init__(self, user_type, *args, **kwargs):
super(BnpcartesentrepriseCorporateBrowser, self).__init__(*args, **kwargs)
self.accounts = []
......