Commit 1311781d authored by David Kremer's avatar David Kremer Committed by Romain Bignon
Browse files

[hsbc] iter investments on pea

- Add an almost full client to retrieve informations from the
  `pea_wealth` subsite through its JSON API.
- Split pages.py in several files for maintenance purpose
parent 800be4ae
......@@ -31,11 +31,17 @@
from weboob.browser import LoginBrowser, URL, need_login
from weboob.browser.exceptions import HTTPNotFound
from .pages import (
from .pages.account_pages import (
AccountsPage, CBOperationPage, CPTOperationPage, LoginPage, AppGonePage, RibPage,
LifeInsurancesPage, FrameContainer, LifeInsurancePortal, LifeInsuranceMain,
LifeInsuranceUseless, UnavailablePage, OtherPage,
UnavailablePage, OtherPage, FrameContainer
)
from .pages.life_insurances import (
LifeInsurancesPage, LifeInsurancePortal, LifeInsuranceMain, LifeInsuranceUseless,
)
from .pages.investments import (
LogonInvestmentPage, ProductViewHelper, RetrieveAccountsPage, RetrieveInvestmentsPage, RetrieveLiquidityPage
)
from .pages.landing_pages import JSMiddleFramePage, JSMiddleAuthPage, InvestmentFormPage
__all__ = ['HSBC']
......@@ -43,6 +49,7 @@
class HSBC(LoginBrowser):
BASEURL = 'https://client.hsbc.fr'
app_gone = False
connection = URL(r'https://www.hsbc.fr/1/2/hsbc-france/particuliers/connexion', LoginPage)
......@@ -74,6 +81,28 @@ class HSBC(LoginBrowser):
life_insurance_main = URL('https://assurances.hsbc.fr/fr/accueil/b2c/accueil.html\?pointEntree=PARTIEGENERIQUEB2C', LifeInsuranceMain)
life_insurances = URL('https://assurances.hsbc.fr/navigation', LifeInsurancesPage)
# investment pages
middle_frame_page = URL(r'/cgi-bin/emcgi', JSMiddleFramePage)
middle_auth_page = URL(r'/cgi-bin/emcgi', JSMiddleAuthPage)
investment_form_page = URL(
r'https://www.hsbc.fr/1/[0-9]/authentication/sso-cwd\?customerFullName=.*',
InvestmentFormPage
)
logon_investment_page = URL(r'https://investissements.clients.hsbc.fr/group-wd-gateway-war/gateway/LogonAuthentication', LogonInvestmentPage)
retrieve_accounts_view = URL(
r'https://investissements.clients.hsbc.fr/group-wd-gateway-war/gateway/wd/RetrieveProductView',
RetrieveAccountsPage
)
retrieve_investments_page = URL(
r'https://investissements.clients.hsbc.fr/group-wd-gateway-war/gateway/wd/RetrieveProductView',
RetrieveInvestmentsPage
)
retrieve_liquidity_page = URL(
r'https://investissements.clients.hsbc.fr/group-wd-gateway-war/gateway/wd/RetrieveProductView',
RetrieveLiquidityPage
)
# catch-all
other_page = URL(r'/cgi-bin/emcgi', OtherPage)
......@@ -81,6 +110,7 @@ def __init__(self, username, password, secret, *args, **kwargs):
super(HSBC, self).__init__(username, password, *args, **kwargs)
self.accounts_list = dict()
self.secret = secret
self.PEA_LISTING = {}
def load_state(self, state):
return
......@@ -282,10 +312,22 @@ def _get_history(self):
for tr in self.page.get_history():
yield tr
def get_investments(self, account, retry_li=True):
if account.type != Account.TYPE_LIFE_INSURANCE:
def get_investments(self, account):
if account.type == Account.TYPE_LIFE_INSURANCE:
return self.get_life_investments(account)
elif account.type in (Account.TYPE_MARKET, Account.TYPE_PEA):
return self.get_pea_investments(account)
else:
raise NotImplementedError()
def get_pea_investments(self, account):
assert account.type in (Account.TYPE_PEA, Account.TYPE_MARKET)
if not self.PEA_LISTING:
self._go_to_wealth_accounts()
return self.PEA_LISTING['investments']
def get_life_investments(self, account, retry_li=True):
self._quit_li_space()
self.update_accounts_list(False)
......@@ -315,3 +357,20 @@ def get_investments(self, account, retry_li=True):
self._quit_li_space()
return investments
def _go_to_wealth_accounts(self):
if not hasattr(self.page, 'get_middle_frame_url'):
# if we can catch the URL, we go directly, else we need to browse
# the website
self.update_accounts_list()
self.location(self.page.get_middle_frame_url())
self.location(self.page.get_patrimoine_url())
self.page.go_next()
self.page.go_to_logon()
helper = ProductViewHelper(self)
# we need to go there to initialize the session
self.PEA_LISTING['accounts'] = list(helper.retrieve_accounts())
self.PEA_LISTING['liquidities'] = list(helper.retrieve_liquidity_account())
self.PEA_LISTING['investments'] = list(helper.retrieve_invests())
self.connection.go()
......@@ -20,21 +20,22 @@
from __future__ import unicode_literals
import re
from decimal import Decimal
from weboob.capabilities import NotAvailable
from weboob.capabilities.bank import Account, Investment, AccountNotFound
from weboob.capabilities.bank import Account
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.tools.compat import urlparse, parse_qs, urljoin
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable, ActionNeeded
from weboob.browser.elements import TableElement, ListElement, ItemElement, method
from weboob.browser.pages import HTMLPage, LoggedPage, pagination, FormNotFound
from weboob.browser.filters.standard import Filter, Env, CleanText, CleanDecimal, Field, DateGuesser, TableCell, Regexp, \
Eval, Date
from weboob.browser.filters.html import Link, XPathNotFound, AbsoluteLink
from weboob.browser.elements import ListElement, ItemElement, method
from weboob.browser.pages import HTMLPage, pagination
from weboob.browser.filters.standard import (
Filter, Env, CleanText, CleanDecimal, Field, DateGuesser, TableCell, Regexp
)
from weboob.browser.filters.html import Link, AbsoluteLink
from weboob.browser.filters.javascript import JSVar
from .landing_pages import GenericLandingPage
class Transaction(FrenchTransaction):
PATTERNS = [(re.compile(r'^VIR(EMENT)? (?P<text>.*)'), FrenchTransaction.TYPE_TRANSFER),
......@@ -50,7 +51,7 @@ class Transaction(FrenchTransaction):
]
class FrameContainer(LoggedPage, HTMLPage):
class FrameContainer(GenericLandingPage):
is_here = '//frameset'
# main page, a frameset
......@@ -72,18 +73,14 @@ def get_frame(self):
return a.attrib['src']
class LifeInsuranceUseless(LoggedPage, HTMLPage):
is_here = '//h1[text()="Assurance Vie"]'
class UnavailablePage(LoggedPage, HTMLPage):
class UnavailablePage(GenericLandingPage):
is_here = '//strong[contains(text(),"Service momentanément indisponible.")]'
def on_load(self):
raise BrowserUnavailable()
class AccountsPage(LoggedPage, HTMLPage):
class AccountsPage(GenericLandingPage):
is_here = '//h1[text()="Synthèse"]'
@method
......@@ -151,7 +148,7 @@ def obj_id(self):
return CleanText(replace=[('.', ''), (' ', '')]).filter(self.el.xpath('./td[2]'))
class RibPage(LoggedPage, HTMLPage):
class RibPage(GenericLandingPage):
def is_here(self):
return bool(self.doc.xpath('//h1[contains(text(), "RIB/IBAN")]'))
......@@ -188,7 +185,7 @@ def next_page(self):
return
class CBOperationPage(LoggedPage, HTMLPage):
class CBOperationPage(GenericLandingPage):
is_here = '//h1[text()="Historique des opérations"]'
def get_params(self, url):
......@@ -214,7 +211,7 @@ def obj_date(self):
return DateGuesser(Regexp(CleanText(self.page.doc.xpath('//table/tr[2]/td[1]')), r'(\d{2}/\d{2})'), Env("date_guesser"))(self)
class CPTOperationPage(LoggedPage, HTMLPage):
class CPTOperationPage(GenericLandingPage):
is_here = '''//h1[text()="Historique des opérations"] and //h2[text()="Recherche d'opération"]'''
def get_history(self):
......@@ -262,6 +259,9 @@ def on_load(self):
for message in self.doc.xpath('//div[has-class("csPanelErrors")]'):
raise BrowserIncorrectPassword(CleanText('.')(message))
def is_here(self):
return not self.doc.xpath('//form[@name="launch"]')
def login(self, login):
form = self.get_form(id='idv_auth_form')
form['userid'] = form['__hbfruserid'] = login
......@@ -311,103 +311,3 @@ def on_load(self):
raise exc(CleanText('.')(tag))
## Life insurance subsite
class LITransaction(FrenchTransaction):
PATTERNS = [(re.compile(u'^(?P<text>Arbitrage.*)'), FrenchTransaction.TYPE_ORDER),
(re.compile(u'^(?P<text>Versement.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile(r'^(?P<text>.*)'), FrenchTransaction.TYPE_BANK),
]
class LifeInsurancePortal(LoggedPage, HTMLPage):
def is_here(self):
try:
self.get_form(name='FORM_ERISA')
except FormNotFound:
return False
return True
def on_load(self):
self.logger.debug('automatically following form')
form = self.get_form(name='FORM_ERISA')
form['token'] = JSVar(CleanText('//script'), var='document.FORM_ERISA.token.value')(self.doc)
form.submit()
class LifeInsuranceMain(LoggedPage, HTMLPage):
def on_load(self):
self.logger.debug('automatically following form')
form = self.get_form(name='formAccueil')
form.url = 'https://assurances.hsbc.fr/navigation'
form.submit()
class LifeInsurancesPage(LoggedPage, HTMLPage):
@method
class iter_history(TableElement):
head_xpath = '(//table)[1]/thead/tr/th'
item_xpath = '(//table)[1]/tbody/tr'
col_label = 'Actes'
col_date = 'Date d\'effet'
col_amount = 'Montant net'
col_gross_amount = 'Montant brut'
class item(ItemElement):
klass = LITransaction
obj_raw = LITransaction.Raw(CleanText(TableCell('label')))
obj_date = Date(CleanText(TableCell('date')))
obj_amount = Transaction.Amount(TableCell('amount'), TableCell('gross_amount'), replace_dots=False)
@method
class iter_investments(TableElement):
head_xpath = u'//div[contains(., "Détail de vos supports")]/following-sibling::div/table/thead/tr/th'
item_xpath = u'//div[contains(., "Détail de vos supports")]/following-sibling::div/table/tbody/tr'
col_label = 'Support'
col_vdate = u'Date de valorisation *'
col_quantity = u'Nombre d\'unités de compte'
col_portfolio_share = u'Répartition'
col_unitvalue = u'Valeur liquidative'
col_support_value = u'Valeur support'
class item(ItemElement):
klass = Investment
obj_label = CleanText(TableCell('label'))
obj_vdate = Date(CleanText(TableCell('vdate')), dayfirst=True)
obj_portfolio_share = Eval(lambda x: x / 100, CleanDecimal(TableCell('portfolio_share')))
obj_unitvalue = CleanDecimal(TableCell('unitvalue'), default=Decimal('1'))
obj_valuation = CleanDecimal(TableCell('support_value'))
def obj_code(self):
if "Fonds en euros" in Field('label')(self):
return NotAvailable
return Regexp(Link('.//a'), r'javascript:openSupportPerformanceWindow\(\'(.*?)\', \'(.*?)\'\)', '\\2')(self)
def obj_quantity(self):
# default for euro funds
return CleanDecimal(TableCell('quantity'), default=CleanDecimal(TableCell('support_value'))(self))(self)
def condition(self):
return len(self.el.xpath('.//td')) > 1
def get_lf_attributes(self, lfid):
attributes = {}
# values can be in JS var format but it's not mandatory param so we don't go to get the real value
try:
values = Regexp(Link('//a[contains(., "%s")]' % lfid[:-3].lstrip('0')), r'\((.*?)\)')(self.doc).replace(' ', '').replace('\'', '').split(',')
except XPathNotFound:
raise AccountNotFound('cannot find account id %s on life insurance site' % lfid)
keys = Regexp(CleanText('//script'), r'consultationContrat\((.*?)\)')(self.doc).replace(' ', '').split(',')
attributes = dict(zip(keys, values))
return attributes
def disconnect(self):
self.get_form(name='formDeconnexion').submit()
##
# coding: utf-8
from __future__ import unicode_literals
from __future__ import division
import datetime
import json
import time
from weboob.capabilities import NotAvailable
from weboob.capabilities.bank import Account, Investment
from weboob.browser.elements import ItemElement, DictElement, method
from weboob.browser.pages import HTMLPage, JsonPage, LoggedPage
from weboob.browser.filters.standard import (
CleanText, CleanDecimal, Regexp, Eval, Currency
)
from weboob.browser.filters.json import Dict
from weboob.browser.filters.javascript import JSVar
class LogonInvestmentPage(LoggedPage, HTMLPage):
"""Transient page to the real application page."""
SESSION_INFO = {}
def on_load(self):
_, app_data = self.get_session_storage()
self.SESSION_INFO['app_location'] = JSVar(var='window.location').filter(self.content.decode())
self.SESSION_INFO['app_data'] = app_data
self.browser.SESSION_INFO = self.SESSION_INFO
def is_here(self):
return 'appPage.min.html' in self.content.decode('iso-8859-1')
def get_session_storage(self):
sessionContent = Regexp(
CleanText('//script[@type="text/javascript"]'),
'sessionStorage.setItem\((.*)\)'
)(self.doc)
key, value = map(lambda x: x.strip("'").strip(), sessionContent.split(",", 1))
return key, json.decoder.JSONDecoder().decode(value)
class ProductViewHelper():
URL = 'https://investissements.clients.hsbc.fr/group-wd-gateway-war/gateway/wd/RetrieveProductView'
def __init__(self, browser):
self.browser = browser
def raw_post_data(self):
null = None
return {
"aggregateXRaySegmentFilter": [],
"businessOpUnit": "141",
"cacheRefreshIndicator": null,
"functionIndicator": [
{"functionMessageTriggerDescription": "MyPortfolio-MyHoldings|R01"}
],
"holdingAccountInformation": {
"accountFilterIndicator": "N",
"accountFilterRefreshIndicator": "Y",
"cacheRefreshIndicator": "Y",
"holdingGroupingViewConfig": "ASSETTYPE",
"investmentHistoryRequestTypeCode": "CURR",
"priceQuoteTypeCode": "Delay",
"productDashboardTypeInformation": [
{"productDashboardTypeCode": "EQ"},
{"productDashboardTypeCode": "BOND"},
{"productDashboardTypeCode": "MNYUT"},
{"productDashboardTypeCode": "DIVUT"},
{"productDashboardTypeCode": "EURO"},
{"productDashboardTypeCode": "SI"},
{"productDashboardTypeCode": "FCPI"},
{"productDashboardTypeCode": "SCPI"},
{"productDashboardTypeCode": "ALT"},
{"productDashboardTypeCode": "LCYDEP"},
{"productDashboardTypeCode": "FCYDEP"},
{"productDashboardTypeCode": "INVTINSUR"},
{"productDashboardTypeCode": "NONINVTINSUR"},
{"productDashboardTypeCode": "LOAN"},
{"productDashboardTypeCode": "MORTGAGE"},
{"productDashboardTypeCode": "CARD"},
{"productDashboardTypeCode": "UWCASH"}
],
"transactionRangeStartDate": null,
"watchListFilterIndicator": "N"
},
"holdingSegmentFilter": [],
"orderStatusFilter": [
{"orderStatusGroupIdentifier": "HOLDING", "productCode": null, "productDashboardTypeCode": null},
{"orderStatusGroupIdentifier": "PENDING", "productCode": null, "productDashboardTypeCode": null}
],
"paginationRequest": [],
"portfolioAnalysisFilter": [],
"segmentFilter": [
{"dataSegmentGroupIdentifier": "PRTFDTLINF"},
{"dataSegmentGroupIdentifier": "PORTFTLINF"},
{"dataSegmentGroupIdentifier": "ACCTGRPINF"},
{"dataSegmentGroupIdentifier": "ACCTFLTINF"}
],
"sortingCriterias": [],
"staffId": null,
"watchlistFilter": []
}
def investment_list_post_data(self):
raw_data = self.raw_post_data()
raw_data.pop('aggregateXRaySegmentFilter')
raw_data.pop('holdingSegmentFilter')
raw_data.pop('portfolioAnalysisFilter')
raw_data.pop('watchlistFilter')
raw_data.pop('cacheRefreshIndicator')
raw_data.update({
"functionIndicator": [
{"functionMessageTriggerDescription": "MyPortfolio-MyHoldings"}
],
"holdingAccountInformation": {
"accountFilterIndicator": "N",
"accountFilterRefreshIndicator": "N",
"cacheRefreshIndicator": "N",
"holdingGroupingViewConfig": "ASSETTYPE",
"investmentHistoryRequestTypeCode": "CURR",
"priceQuoteTypeCode": "Delay",
"productDashboardTypeInformation": [
{"productDashboardTypeCode": "EQ"}
],
"watchListFilterIndicator": "N"
},
"orderStatusFilter": [
{"orderStatusGroupIdentifier": "HOLDING"},
{"orderStatusGroupIdentifier": "PENDING"}
],
"segmentFilter": [
{"dataSegmentGroupIdentifier": "HLDORDRINF"},
{"dataSegmentGroupIdentifier": "HLDGSUMINF"}
],
"sortingCriterias": [
{"sortField": "PROD-DSHBD-TYP-CDE", "sortOrder": "+"},
{"sortField": "PRD-DSHBD-STYP-CDE", "sortOrder": "+"},
{"sortField": "PROD-SHRT-NAME", "sortOrder": "+"}
],
})
return raw_data
def liquidity_account_post_data(self):
base_data = self.investment_list_post_data()
base_data.update({
"segmentFilter": [{"dataSegmentGroupIdentifier": "HLDORDRINF"}],
"sortingCriterias": [
{"sortField": "ACCT-NUM", "sortOrder": "+"},
{"sortField": "ACCT-PROD-TYPE-STR", "sortOrder": "+"},
{"sortField": "CCY-PROD-CDE", "sortOrder": "+"},
{"sortField": "PROD-MTUR-DT", "sortOrder": "+"}
]
})
base_data['holdingAccountInformation']['productDashboardTypeInformation'] = [
{"productDashboardTypeCode": "UWCASH"}
]
return base_data
def build_request(self, kind=None):
return dict(
url=self.URL,
data=self.build_request_data(kind=kind),
headers=self.build_request_headers(),
cookies=self.build_request_cookies(),
)
def build_request_headers(self):
xsrf_token = self.browser.session.cookies['XSRF-TOKEN']
return {
"Content-Type": "application/json;charset=UTF-8",
"Accept-Encoding": "gzip, deflate, br",
'Accept': '*/*',
"Connection": "keep-alive",
"X-HDR-App-Role": "ALL",
"X-HDR-Target-Function": "currentholdings",
'X-XSRF-TOKEN': xsrf_token,
}
def build_request_cookies(self):
mandatory_cookies = {
'opt_in_status': "1",
'CAMToken': self.browser.session.cookies.get('CAMToken', domain='.investissements.clients.hsbc.fr')
}
for key in ('JSESSIONID', 'XSRF-TOKEN', 'WEALTH-FR-CUST-PORTAL-COOKIE'):
value = self.browser.session.cookies.get(key, domain='investissements.clients.hsbc.fr')
assert value, key + " cookie is not set"
mandatory_cookies.update({key: value})
return mandatory_cookies
def build_request_data(self, kind=None):
d = self.browser.SESSION_INFO['app_data'].get('data')
assert d, 'No Session Data to perform a request'
localeCode = '_'.join((d['localeLanguage'], d['localeCountry']))
holdingAccountInformation = {
'customerNumber': d['customerID'],
'localeLocalCode': localeCode,
'transactionRangeEndDate': int(time.time() * 1000),
}
baseHeader = {
'sessionId': d['sessionID'],
'userDeviceId': d['userDeviceID'],
'userId': d['userId'],
}
request_data = {
'channelId': d['channelID'],
'countryCode': d['customerCountryCode'],
'customerNumber': d['customerID'],
'frameworkHeader': {
'customerElectronicBankingChangeableIdentificationNumber': d['userId'],
'customerElectronicBankingIdentificationNumber': d['userId'],
},
'groupMember': d['customerGroupMemberID'],
'lineOfBusiness': d['customerBusinessLine'],
'localeCode': localeCode,
'swhcbApplicationHeader': {
'hubUserId': d['userLegacyID'],
'hubWorkstationId': d['userLegacyDeviceID'],
},
}
if kind == 'account_list':
holdingAccountInformation.update(self.raw_post_data()['holdingAccountInformation'])
request_data.update(self.raw_post_data())
elif kind == 'investment_list' or kind == 'liquidity_account':
""" Build request data to fetch the list of investments """
request_data.pop("localeCode")
if kind == 'investment_list':
holdingAccountInformation.update(self.investment_list_post_data()['holdingAccountInformation'])
request_data.update(self.investment_list_post_data())
elif kind == 'liquidity_account':
holdingAccountInformation.update(self.liquidity_account_post_data()['holdingAccountInformation'])
request_data.update(self.liquidity_account_post_data())
if 'req_id' in self.browser.SESSION_INFO: # update request identification number
holdingAccountInformation['requestIdentificationNumber'] = self.browser.SESSION_INFO['req_id']
else:
raise NotImplementedError()
# set up common keys for the request
request_data['holdingAccountInformation'] = holdingAccountInformation
request_data['baseHeader'] = baseHeader
return request_data
def retrieve_products(self, kind=None):
""" Build the request from scratch according to 'kind' parameter """
req = self.build_request(kind=kind)
# self.browser.location(self.browser.SESSION_INFO['app_location'])
# cookies may be optionals but headers are mandatory.
self.browser.location(req['url'], method='POST', data=json.dumps(req['data']), headers=req['headers'], cookies=req['cookies'])
self.browser.SESSION_INFO['req_id'] = self.browser.response.json()['sessionInformation']['requestIdentificationNumber']
def retrieve_invests(self):
self.retrieve_products(kind='investment_list')
assert isinstance(self.browser.page, RetrieveInvestmentsPage)
return self.browser.page.iter_investments()
def retrieve_liquidity_account(self):
self.retrieve_products(kind='liquidity_account')
assert isinstance(self.browser.page, RetrieveLiquidityPage)
return self.browser.page.iter_liquidity_accounts()
def retrieve_accounts(self):
self.retrieve_products(kind='account_list')
assert isinstance(self.browser.page, RetrieveAccountsPage)
return self.browser.page.iter_accounts()
class RetrieveAccountsPage(LoggedPage, JsonPage):
def is_here(self):
# We should never have both informations at the same time
assert bool(self.response.json()['holdingOrderInformation']) != bool(self.response.json()['accountFilterInformation'])
return bool(self.response.json()['accountFilterInformation'])
@method
class iter_accounts(DictElement):
TYPE_ACCOUNTS = {
'INVESTMENT': Account.TYPE_PEA,
'CURRENT': Account.TYPE_CHECKING,
}
item_xpath = 'accountGroupInformation'
class item(ItemElement):