pax_global_header 0000666 0000000 0000000 00000000064 13400742225 0014511 g ustar 00root root 0000000 0000000 52 comment=1a75c09ff1683f83d027e99041a83bcf90e8144c
woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/ 0000775 0000000 0000000 00000000000 13400742225 0022520 5 ustar 00root root 0000000 0000000 woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/modules/ 0000775 0000000 0000000 00000000000 13400742225 0024170 5 ustar 00root root 0000000 0000000 woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/modules/cmso/ 0000775 0000000 0000000 00000000000 13400742225 0025131 5 ustar 00root root 0000000 0000000 woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/modules/cmso/par/ 0000775 0000000 0000000 00000000000 13400742225 0025713 5 ustar 00root root 0000000 0000000 woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/modules/cmso/par/__init__.py 0000664 0000000 0000000 00000000000 13400742225 0030012 0 ustar 00root root 0000000 0000000 woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/modules/cmso/par/browser.py 0000664 0000000 0000000 00000036003 13400742225 0027752 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2016 Edouard Lambert
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see .
from __future__ import unicode_literals
import re
import json
from functools import wraps
from weboob.browser import LoginBrowser, URL, need_login, StatesMixin
from weboob.browser.exceptions import ClientError, ServerError
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable
from weboob.capabilities.bank import Account, Transaction, AccountNotFound
from weboob.capabilities.base import find_object
from weboob.tools.capabilities.bank.transactions import sorted_transactions
from .pages import (
LogoutPage, InfosPage, AccountsPage, HistoryPage, LifeinsurancePage, MarketPage,
AdvisorPage, LoginPage, RecipientsPage, ProfilePage,
)
def retry(exc_check, tries=4):
"""Decorate a function to retry several times in case of exception.
The decorated function is called at max 4 times. It is retried only when it
raises an exception of the type `exc_check`.
If the function call succeeds and returns an iterator, a wrapper to the
iterator is returned. If iterating on the result raises an exception of type
`exc_check`, the iterator is recreated by re-calling the function, but the
values already yielded will not be re-yielded.
For consistency, the function MUST always return values in the same order.
"""
def decorator(func):
@wraps(func)
def wrapper(browser, *args, **kwargs):
cb = lambda: func(browser, *args, **kwargs)
for i in range(tries, 0, -1):
try:
ret = cb()
except exc_check as exc:
browser.headers = None
browser.do_login()
browser.logger.info('%s raised, retrying', exc)
continue
if not hasattr(ret, 'next'):
return ret # simple value, no need to retry on items
return iter_retry(cb, browser, value=ret, remaining=i, exc_check=exc_check, logger=browser.logger)
raise BrowserUnavailable('Site did not reply successfully after multiple tries')
return wrapper
return decorator
class CmsoParBrowser(LoginBrowser, StatesMixin):
__states__ = ('headers',)
STATE_DURATION = 1
headers = None
login = URL('/securityapi/tokens',
'/auth/checkuser', LoginPage)
logout = URL('/securityapi/revoke',
'/auth/errorauthn', LogoutPage)
infos = URL('/comptes/', InfosPage)
accounts = URL('/domiapi/oauth/json/accounts/synthese(?P.*)', AccountsPage)
history = URL('/domiapi/oauth/json/accounts/(?P.*)', HistoryPage)
loans = URL('/creditapi/rest/oauth/v1/synthese', AccountsPage)
lifeinsurance = URL('/assuranceapi/v1/oauth/sso/suravenir/DETAIL_ASSURANCE_VIE/(?P.*)',
'https://domiweb.suravenir.fr/', LifeinsurancePage)
market = URL('/domiapi/oauth/json/ssoDomifronttitre',
'https://www.(?P.*)/domifronttitre/front/sso/domiweb/01/(?P.*)Portefeuille\?csrf=',
'https://www.*/domiweb/prive/particulier', MarketPage)
advisor = URL('/edrapi/v(?P\w+)/oauth/(?P\w+)', AdvisorPage)
recipients = URL(r'/domiapi/oauth/json/transfer/transferinfos', RecipientsPage)
profile = URL(r'/domiapi/oauth/json/edr/infosPerson', ProfilePage)
json_headers = {'Content-Type': 'application/json'}
ARKEA = {'cmso.com': '03', 'cmb.fr': '01', 'cmmc.fr': '02', 'bpe.fr' : '08', 'arkeabanqueprivee.fr': '70',}
def __init__(self, website, *args, **kwargs):
super(CmsoParBrowser, self).__init__(*args, **kwargs)
# Arkea Banque Privee uses specific URL prefix and name
if website == 'arkeabanqueprivee.fr':
self.BASEURL = "https://m.%s" % website
self.name = 'abp'
else:
self.BASEURL = "https://mon.%s" % website
self.name = website.split('.')[0]
self.website = website
self.arkea = self.ARKEA[website]
self.accounts_list = []
self.logged = False
def do_login(self):
if self.headers:
self.session.headers = self.headers
else:
self.set_profile(self.PROFILE) # reset headers but don't clear them
self.session.cookies.clear()
self.accounts_list = []
data = {
'accessCode': self.username,
'password': self.password,
'clientId': 'com.arkea.%s.siteaccessible' % self.name,
'redirectUri': '%s/auth/checkuser' % self.BASEURL,
'errorUri': '%s/auth/errorauthn' % self.BASEURL
}
self.login.go(data=data)
if self.logout.is_here():
raise BrowserIncorrectPassword()
m = re.search('access_token=([^&]+).*id_token=(.*)', self.url)
self.session.headers.update({
'Authentication': "Bearer %s" % m.group(2),
'Authorization': "Bearer %s" % m.group(1),
'X-ARKEA-EFS': self.arkea,
'X-Csrf-Token': m.group(1)
})
self.headers = self.session.headers
def get_account(self, _id):
return find_object(self.iter_accounts(), id=_id, error=AccountNotFound)
@retry((ClientError, ServerError))
@need_login
def iter_accounts(self):
if self.accounts_list:
return self.accounts_list
seen = {}
self.recipients.go(data='{"beneficiaryType":"INTERNATIONAL"}', headers=self.json_headers)
numbers = self.page.get_numbers()
# First get all checking accounts...
data = dict(self.infos.stay_or_go().get_typelist())
self.accounts.go(data=json.dumps(data), type='comptes', headers=self.json_headers)
self.page.check_response()
for key in self.page.get_keys():
for a in self.page.iter_accounts(key=key):
# Can have duplicate account, avoid them
if a._index not in seen:
self.accounts_list.append(a)
seen[a._index] = a
# Next, get saving accounts
numbers.update(self.page.get_numbers())
page = self.accounts.go(data=json.dumps({}), type='epargne', headers=self.json_headers)
for key in page.get_keys():
for a in page.iter_savings(key=key, numbers=numbers):
if a._index in seen:
acc = seen[a._index]
self.accounts_list.remove(acc)
self.logger.warning('replace %s because it seems to be a duplicate of %s', seen[a._index], a)
self.accounts_list.append(a)
# Then, get loans
for key in self.loans.go().get_keys():
for a in self.page.iter_loans(key=key):
if a.id in seen:
self.logger.warning('skipping %s because it seems to be a duplicate of %s', seen[a.id], a)
account_found = False
for account in list(self.accounts_list):
# Loan id can be not unique when it also appears in json account page
if a.id == account._index:
account_found = True
# Merge information from account to loan
a.id = account.id
a.currency = account.currency
a.coming = account.coming
a.total_amount = account._total_amount
a._index = account._index
self.accounts_list.remove(account)
break
assert account_found
self.accounts_list.append(a)
return self.accounts_list
def _go_market_history(self):
content = self.market.go(data=json.dumps({'place': 'SITUATION_PORTEFEUILLE'}), headers=self.json_headers).content
self.location(json.loads(content)['urlSSO'])
return self.market.go(website=self.website, action='historique')
@retry((ClientError, ServerError))
@need_login
def iter_history(self, account):
account = self.get_account(account.id)
if account.type is Account.TYPE_LOAN:
return iter([])
if account.type == Account.TYPE_LIFE_INSURANCE:
url = json.loads(self.lifeinsurance.go(accid=account._index).content)['url']
url = self.location(url).page.get_link("opérations")
return self.location(url).page.iter_history()
elif account.type in (Account.TYPE_PEA, Account.TYPE_MARKET):
self._go_market_history()
if not self.page.go_account(account.label, account._owner):
return []
if not self.page.go_account_full():
return []
# Display code ISIN
self.location(self.url, params={'reload': 'oui', 'convertirCode': 'oui'})
# don't rely on server-side to do the sorting, not only do you need several requests to do so
# but the site just toggles the sorting, resulting in reverse order if you browse multiple accounts
return sorted_transactions(self.page.iter_history())
# Getting a year of history
nbs = ["UN", "DEUX", "TROIS", "QUATRE", "CINQ", "SIX", "SEPT", "HUIT", "NEUF", "DIX", "ONZE", "DOUZE"]
trs = []
self.history.go(data=json.dumps({"index": account._index}), page="pendingListOperations", headers=self.json_headers)
has_deferred_cards = self.page.has_deferred_cards()
self.history.go(data=json.dumps({'index': account._index}), page="detailcompte", headers=self.json_headers)
self.trs = {'lastdate': None, 'list': []}
for tr in self.page.iter_history(index=account._index, nbs=nbs):
if has_deferred_cards and tr.type == Transaction.TYPE_CARD:
tr.type = Transaction.TYPE_DEFERRED_CARD
trs.append(tr)
return trs
@retry((ClientError, ServerError))
@need_login
def iter_coming(self, account):
account = self.get_account(account.id)
if account.type is Account.TYPE_LOAN:
return iter([])
comings = []
self.history.go(data=json.dumps({"index": account._index}), page="pendingListOperations", headers=self.json_headers)
for key in self.page.get_keys():
self.trs = {'lastdate': None, 'list': []}
for c in self.page.iter_history(key=key):
if hasattr(c, '_deferred_date'):
c.date = c._deferred_date
c.type = Transaction.TYPE_DEFERRED_CARD # force deferred card type for comings inside cards
c.vdate = None # vdate don't work for comings
comings.append(c)
return iter(comings)
@retry((ClientError, ServerError))
@need_login
def iter_investment(self, account):
account = self.get_account(account.id)
if account.type == Account.TYPE_LIFE_INSURANCE:
url = json.loads(self.lifeinsurance.go(accid=account._index).content)['url']
url = self.location(url).page.get_link("supports")
if not url:
return iter([])
return self.location(url).page.iter_investment()
elif account.type in (Account.TYPE_MARKET, Account.TYPE_PEA):
data = {"place": "SITUATION_PORTEFEUILLE"}
response = self.market.go(data=json.dumps(data), headers=self.json_headers)
self.location(json.loads(response.content)['urlSSO'])
self.market.go(website=self.website, action="situation")
if self.page.go_account(account.label, account._owner):
return self.page.iter_investment()
return []
raise NotImplementedError()
@retry((ClientError, ServerError))
@need_login
def get_advisor(self):
advisor = self.advisor.go(version="2", page="conseiller").get_advisor()
return iter([self.advisor.go(version="1", page="agence").update_agency(advisor)])
@retry((ClientError, ServerError))
@need_login
def get_profile(self):
return self.profile.go(data=json.dumps({})).get_profile()
class iter_retry(object):
# when the callback is retried, it will create a new iterator, but we may already yielded
# some values, so we need to keep track of them and seek in the middle of the iterator
def __init__(self, cb, browser, remaining=4, value=None, exc_check=Exception, logger=None):
self.cb = cb
self.it = value
self.items = []
self.remaining = remaining
self.exc_check = exc_check
self.logger = logger
self.browser = browser
self.delogged = False
def __iter__(self):
return self
def __next__(self):
if self.remaining <= 0:
raise BrowserUnavailable('Site did not reply successfully after multiple tries')
if self.delogged:
self.browser.do_login()
self.delogged = False
if self.it is None:
self.it = self.cb()
# recreated iterator, consume previous items
try:
nb = -1
for nb, sent in enumerate(self.items):
new = next(self.it)
if hasattr(new, 'iter_fields'):
equal = dict(sent.iter_fields()) == dict(new.iter_fields())
else:
equal = sent == new
if not equal:
# safety is not guaranteed
raise BrowserUnavailable('Site replied inconsistently between retries, %r vs %r', sent, new)
except StopIteration:
raise BrowserUnavailable('Site replied fewer elements (%d) than last iteration (%d)', nb + 1, len(self.items))
except self.exc_check as exc:
self.delogged = True
if self.logger:
self.logger.info('%s raised, retrying', exc)
self.it = None
self.remaining -= 1
return next(self)
# return one item
try:
obj = next(self.it)
except self.exc_check as exc:
self.delogged = True
if self.logger:
self.logger.info('%s raised, retrying', exc)
self.it = None
self.remaining -= 1
return next(self)
else:
self.items.append(obj)
return obj
next = __next__
woob-1a75c09ff1683f83d027e99041a83bcf90e8144c-modules-cmso-par/modules/cmso/par/pages.py 0000664 0000000 0000000 00000060430 13400742225 0027367 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2016 Edouard Lambert
#
# This file is part of weboob.
#
# weboob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# weboob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see .
from __future__ import unicode_literals
import re
import requests
import json
import datetime as dt
from collections import OrderedDict
from weboob.browser.pages import HTMLPage, JsonPage, RawPage, LoggedPage, pagination
from weboob.browser.elements import DictElement, ItemElement, TableElement, SkipItem, method
from weboob.browser.filters.standard import CleanText, Upper, Date, Regexp, Format, CleanDecimal, Filter, Env, Slugify, Field
from weboob.browser.filters.json import Dict
from weboob.browser.filters.html import Attr, Link, TableCell
from weboob.browser.exceptions import ServerError
from weboob.capabilities.bank import Account, Investment, Loan
from weboob.capabilities.contact import Advisor
from weboob.capabilities.base import NotAvailable
from weboob.capabilities.profile import Profile
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.exceptions import ParseError
from weboob.tools.capabilities.bank.investments import is_isin_valid
from weboob.tools.compat import unicode
def MyDecimal(*args, **kwargs):
kwargs.update(replace_dots=True, default=NotAvailable)
return CleanDecimal(*args, **kwargs)
class LoginPage(HTMLPage):
pass
class LogoutPage(RawPage):
pass
class InfosPage(LoggedPage, HTMLPage):
def get_typelist(self):
url = Attr(None, 'src').filter(self.doc.xpath('//script[contains(@src, "comptes/scripts")]'))
m = re.search(r'synthesecomptes[^\w]+([^:]+)[^\w]+([^"]+)', self.browser.open(url).text)
return {m.group(1): m.group(2)}
class AccountsPage(LoggedPage, JsonPage):
TYPES = OrderedDict([('courant', Account.TYPE_CHECKING),
('pee', Account.TYPE_PEE),
('epargne en actions', Account.TYPE_PEA),
('pea', Account.TYPE_PEA),
('preference', Account.TYPE_LOAN),
('livret', Account.TYPE_SAVINGS),
('vie', Account.TYPE_LIFE_INSURANCE),
('previ_option', Account.TYPE_LIFE_INSURANCE),
('actions', Account.TYPE_MARKET),
('titres', Account.TYPE_MARKET),
('ldd cm', Account.TYPE_SAVINGS),
('librissime', Account.TYPE_SAVINGS),
('epargne logement', Account.TYPE_SAVINGS),
('plan bleu', Account.TYPE_SAVINGS),
('capital plus', Account.TYPE_SAVINGS),
])
def get_keys(self):
"""Returns the keys for which the value is a list or dict"""
if "exception" in self.doc:
return []
return [k for k, v in self.doc.items() if v and isinstance(v, (dict, list))]
def check_response(self):
if "exception" in self.doc:
self.logger.warning("There are no checking accounts: exception '{}' with code {}".format(
self.doc['exception']['message'],
self.doc['exception']['code'])
)
def get_numbers(self):
keys = self.get_keys()
numbers = {}
for key in keys:
if isinstance(self.doc[key], dict):
keys_ = [k for k in self.doc[key] if isinstance(k, unicode)]
contracts = [v for k in keys_ for v in self.doc[key][k]]
else:
contracts = [v for v in self.doc[key]]
numbers.update({c['index']: c['numeroContratSouscrit'] for c in contracts})
return numbers
@method
class iter_accounts(DictElement):
def parse(self, el):
self.item_xpath = "%s/*" % Env('key')(self)
def find_elements(self):
selector = self.item_xpath.split('/')
for sub_element in selector:
if isinstance(self.el, dict) and self.el and sub_element == '*':
self.el = next(iter(self.el.values())) # replace self.el with its first value
if sub_element == '*':
continue
self.el = self.el[sub_element]
for sub_element in self.el:
yield sub_element
class item(ItemElement):
klass = Account
condition = lambda self: "LIVRET" not in Dict('accountType')(self.el)
obj_id = Dict('numeroContratSouscrit')
obj_label = Upper(Dict('lib'))
obj_currency = Dict('deviseCompteCode')
obj_coming = CleanDecimal(Dict('AVenir', default=None), default=NotAvailable)
# Iban is available without last 5 numbers, or by sms
obj_iban = NotAvailable
obj__index = Dict('index')
def obj_balance(self):
balance = CleanDecimal(Dict('soldeEuro', default="0"))(self)
return -abs(balance) if Field('type')(self) == Account.TYPE_LOAN else balance
# It can have revolving credit on this page
def obj__total_amount(self):
return CleanDecimal(Dict('grantedAmount', default=None), default=NotAvailable)(self)
def obj_type(self):
return self.page.TYPES.get(Dict('accountType', default=None)(self).lower(), Account.TYPE_UNKNOWN)
@method
class iter_savings(DictElement):
@property
def item_xpath(self):
return "%s/*/savingsProducts" % Env('key')(self)
def store(self, obj):
id = obj.id
n = 1
while id in self.objects:
self.logger.warning('There are two objects with the same ID! %s' % id)
n += 1
id = '%s-%s' % (obj.id, n)
obj.id = id
self.objects[obj.id] = obj
return obj
# the accounts really are deeper, but the account type is in a middle-level
class iter_accounts(DictElement):
item_xpath = 'savingsAccounts'
def parse(self, el):
# accounts may have a user-entered label, so it shouldn't be relied too much on for parsing the account type
self.env['type_label'] = el['libelleProduit']
def store(self, obj):
id = obj.id
n = 1
while id in self.objects:
self.logger.warning('There are two objects with the same ID! %s' % id)
n += 1
id = '%s-%s' % (obj.id, n)
obj.id = id
self.objects[obj.id] = obj
return obj
class item(ItemElement):
klass = Account
obj_label = Upper(Dict('libelleContrat'))
obj_balance = CleanDecimal(Dict('solde', default="0"))
obj_currency = 'EUR'
obj_coming = CleanDecimal(Dict('AVenir', default=None), default=NotAvailable)
obj__index = Dict('index')
obj__owner = Dict('nomTitulaire')
def obj_id(self):
type = Field('type')(self)
if type == Account.TYPE_LIFE_INSURANCE:
number = self.get_lifenumber()
if number:
return number
elif type in (Account.TYPE_PEA, Account.TYPE_MARKET):
number = self.get_market_number()
if number:
return number
try:
return Env('numbers')(self)[Dict('index')(self)]
except KeyError:
# index often changes, so we can't use it... and have to do something ugly
return Slugify(Format('%s-%s', Dict('libelleContrat'), Dict('nomTitulaire')))(self)
def obj_type(self):
for key in self.page.TYPES:
if key in Env('type_label')(self).lower():
return self.page.TYPES[key]
return Account.TYPE_UNKNOWN
def get_market_number(self):
label = Field('label')(self)
page = self.page.browser._go_market_history()
return page.get_account_id(label, Field('_owner')(self))
def get_lifenumber(self):
index = Dict('index')(self)
data = json.loads(self.page.browser.lifeinsurance.open(accid=index).content)
if not data:
raise SkipItem('account seems unavailable')
url = data['url']
page = self.page.browser.open(url).page
return page.get_account_id()
@method
class iter_loans(DictElement):
def parse(self, el):
self.item_xpath = Env('key')(self)
if "Pret" in Env('key')(self):
self.item_xpath = "%s/*/lstPret" % self.item_xpath
class item(ItemElement):
klass = Loan
def obj_id(self):
# it seems that if we don't have "numeroContratSouscrit", "identifiantTechnique" is unique : only this direction !
return Dict('numeroContratSouscrit', default=None)(self) or Dict('identifiantTechnique')(self)
obj_label = Dict('libelle')
obj_currency = 'EUR'
obj_type = Account.TYPE_LOAN
def obj_total_amount(self):
# Json key change depending on loan type, consumer credit or revolving credit
return CleanDecimal(Dict('montantEmprunte', default=None)(self) or Dict('montantUtilise'))(self)
# Key not always available, when revolving credit not yet consummed
obj_next_payment_amount = CleanDecimal(Dict('montantProchaineEcheance', default=None), default=NotAvailable)
# obj_rate = can't find the info on website except pdf :(
# Dates scraped are timestamp, to remove last '000' we divide by 1000
def obj_maturity_date(self):
# Key not always available, when revolving credit not yet consummed
timestamp = Dict('dateFin', default=None)(self)
if timestamp:
return dt.date.fromtimestamp(timestamp/1000)
return NotAvailable
def obj_next_payment_date(self):
# Key not always available, when revolving credit not yet consummed
timestamp = Dict('dateProchaineEcheance', default=None)(self)
if timestamp:
return dt.date.fromtimestamp(timestamp/1000)
return NotAvailable
def obj_balance(self):
return -abs(CleanDecimal().filter(self.el.get('montantRestant', self.el.get('montantUtilise'))))
# only for revolving loans
obj_available_amount = CleanDecimal(Dict('montantDisponible', default=None), default=NotAvailable)
class Transaction(FrenchTransaction):
PATTERNS = [(re.compile(r'^CARTE (?P\d{2})/(?P\d{2}) (?P.*)'), FrenchTransaction.TYPE_CARD),
(re.compile(r'^(?P(PRLV|PRELEVEMENTS).*)'), FrenchTransaction.TYPE_ORDER),
(re.compile(r'^(?PRET DAB.*)'), FrenchTransaction.TYPE_WITHDRAWAL),
(re.compile(r'^(?PECH.*)'), FrenchTransaction.TYPE_LOAN_PAYMENT),
(re.compile(r'^(?PVIR.*)'), FrenchTransaction.TYPE_TRANSFER),
(re.compile(r'^(?PANN.*)'), FrenchTransaction.TYPE_PAYBACK),
(re.compile(r'^(?P(VRST|VERSEMENT).*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile(r'^(?P.*)'), FrenchTransaction.TYPE_BANK)
]
class HistoryPage(LoggedPage, JsonPage):
def has_deferred_cards(self):
return Dict('pendingDeferredDebitCardList/currentMonthCardList', default=None)
def get_keys(self):
if 'exception' in self.doc:
return []
return [k for k, v in self.doc.items() if v and isinstance(v, (dict, list))]
@pagination
@method
class iter_history(DictElement):
def next_page(self):
if len(Env('nbs', default=[])(self)):
data = {'index': Env('index')(self),
'filtreOperationsComptabilisees': "MOIS_MOINS_%s" % Env('nbs')(self)[0]
}
Env('nbs')(self).pop(0)
return requests.Request('POST', data=json.dumps(data), headers={'Content-Type': 'application/json'})
def parse(self, el):
# Key only if coming
key = Env('key', default=None)(self)
if key:
if "CardList" in key:
self.item_xpath = "%s/currentMonthCardList/*/listeOperations" % key
elif "futureOperationList" in key:
self.item_xpath = "%s/futurePrelevementList" % key
else:
self.item_xpath = "%s/operationList" % key
else:
self.item_xpath = "listOperationProxy"
class item(ItemElement):
klass = Transaction
class FromTimestamp(Filter):
def filter(self, timestamp):
try:
return dt.date.fromtimestamp(int(timestamp[:-3]))
except TypeError:
return self.default_or_raise(ParseError('Element %r not found' % self.selector))
obj_date = FromTimestamp(Dict('dateOperation', default=NotAvailable), default=NotAvailable)
obj_raw = Transaction.Raw(Dict('libelleCourt'))
obj_vdate = Date(Dict('dateValeur', NotAvailable), dayfirst=True, default=NotAvailable)
obj_amount = CleanDecimal(Dict('montantEnEuro'), default=NotAvailable)
def parse(self, el):
key = Env('key', default=None)(self)
if key and "DeferredDebit" in key:
for x in Dict('%s/currentMonthCardList' % key)(self.page.doc):
deferred_date = Dict('dateDiffere', default=None)(x)
if deferred_date:
break
setattr(self.obj, '_deferred_date', self.FromTimestamp().filter(deferred_date))
# Skip duplicate transactions
amount = Dict('montantEnEuro', default=None)(self)
tr = Dict('libelleCourt')(self) + Dict('dateOperation', '')(self) + str(amount)
if amount is None or (tr in self.page.browser.trs['list'] and self.page.browser.trs['lastdate'] <= Field('date')(self)):
raise SkipItem()
self.page.browser.trs['lastdate'] = Field('date')(self)
self.page.browser.trs['list'].append(tr)
class LifeinsurancePage(LoggedPage, HTMLPage):
def get_account_id(self):
account_id = Regexp(CleanText('//h1[@class="portlet-title"]'), r'n° ([\d\s]+)', default=NotAvailable)(self.doc)
if account_id:
return re.sub(r'\s', '', account_id)
def get_link(self, page):
return Link(default=NotAvailable).filter(self.doc.xpath('//a[contains(text(), "%s")]' % page))
@pagination
@method
class iter_history(TableElement):
item_xpath = '//table/tbody/tr[contains(@class, "results")]'
head_xpath = '//table/thead/tr/th'
col_date = re.compile('Date')
col_label = re.compile('Libellé')
col_amount = re.compile('Montant')
next_page = Link('//a[contains(text(), "Suivant") and not(contains(@href, "javascript"))]', default=None)
class item(ItemElement):
klass = Transaction
obj_raw = Transaction.Raw(TableCell('label'))
obj_date = Date(CleanText(TableCell('date')), dayfirst=True)
obj_amount = MyDecimal(TableCell('amount'))
@method
class iter_investment(TableElement):
item_xpath = '//table/tbody/tr[contains(@class, "results")]'
head_xpath = '//table/thead/tr/th'
col_label = re.compile('Libellé')
col_quantity = re.compile('Nb parts')
col_vdate = re.compile('Date VL')
col_unitvalue = re.compile('VL')
col_unitprice = re.compile('Prix de revient')
col_valuation = re.compile('Solde')
class item(ItemElement):
klass = Investment
obj_label = CleanText(TableCell('label'))
obj_code = Regexp(Link('./td/a'), r'Isin%253D([^%]+)')
obj_quantity = MyDecimal(TableCell('quantity'))
obj_unitprice = MyDecimal(TableCell('unitprice'))
obj_unitvalue = MyDecimal(TableCell('unitvalue'))
obj_valuation = MyDecimal(TableCell('valuation'))
obj_vdate = Date(CleanText(TableCell('vdate')), dayfirst=True, default=NotAvailable)
def obj_code_type(self):
return Investment.CODE_TYPE_ISIN if is_isin_valid(Field('code')(self)) else NotAvailable
class MarketPage(LoggedPage, HTMLPage):
def find_account(self, acclabel, accowner):
accowner = sorted(accowner.lower().split()) # first name and last name may not be ordered the same way on market site...
# Check if history is present
if CleanText(default=None).filter(self.doc.xpath('//body/p[contains(text(), "indisponible pour le moment")]')):
return False
ids = None
for a in self.doc.xpath('//a[contains(@onclick, "indiceCompte")]'):
self.logger.debug("get investment from onclick")
label = CleanText('.')(a)
owner = CleanText('./ancestor::tr/preceding-sibling::tr[@class="LnMnTiers"][1]')(a)
owner = sorted(owner.lower().split())
if label == acclabel and owner == accowner:
ids = list(re.search(r'indiceCompte[^\d]+(\d+).*idRacine[^\d]+(\d+)', Attr('.', 'onclick')(a)).groups())
ids.append(CleanText('./ancestor::td/preceding-sibling::td')(a))
self.logger.debug("assign value to ids: {}".format(ids))
return ids
for a in self.doc.xpath('//a[contains(@href, "indiceCompte")]'):
self.logger.debug("get investment from href")
if CleanText('.')(a) == acclabel:
ids = list(re.search(r'indiceCompte[^\d]+(\d+).*idRacine[^\d]+(\d+)', Attr('.', 'href')(a)).groups())
ids.append(CleanText('./ancestor::td/preceding-sibling::td')(a))
self.logger.debug("assign value to ids: {}".format(ids))
return ids
def get_account_id(self, acclabel, owner):
account = self.find_account(acclabel, owner)
if account:
return account[2].replace(' ', '')
def go_account(self, acclabel, owner):
ids = self.find_account(acclabel, owner)
if not ids:
return
form = self.get_form(name="formCompte")
form['indiceCompte'] = ids[0]
form['idRacine'] = ids[1]
try:
return form.submit()
except ServerError:
return False
def go_account_full(self):
form = self.get_form(name="formOperation")
form['dateDebut'] = "02/01/1970"
try:
return form.submit()
except ServerError:
return False
@method
class iter_history(TableElement):
item_xpath = '//table[has-class("domifrontTb")]/tr[not(has-class("LnTit") or has-class("LnTot"))]'
head_xpath = '//table[has-class("domifrontTb")]/tr[1]/td'
col_date = re.compile('Date')
col_label = 'Opération'
col_code = 'Code'
col_quantity = 'Quantité'
col_amount = re.compile('Montant')
class item(ItemElement):
klass = Transaction
obj_label = CleanText(TableCell('label'))
obj_type = Transaction.TYPE_BANK
obj_date = Date(CleanText(TableCell('date')), dayfirst=True)
obj_amount = CleanDecimal(TableCell('amount'))
obj_investments = Env('investments')
def parse(self, el):
i = Investment()
i.label = Field('label')(self)
i.code = CleanText(TableCell('code'))(self)
i.quantity = MyDecimal(TableCell('quantity'))(self)
i.valuation = Field('amount')(self)
i.vdate = Field('date')(self)
self.env['investments'] = [i]
@method
class iter_investment(TableElement):
item_xpath = '//table[has-class("domifrontTb")]/tr[not(has-class("LnTit") or has-class("LnTot"))]'
head_xpath = '//table[has-class("domifrontTb")]/tr[1]/td'
col_label = 'Valeur'
col_code = 'Code'
col_quantity = 'Qté'
col_vdate = 'Date cours'
col_unitvalue = 'Cours'
col_unitprice = re.compile('P.R.U')
col_valuation = 'Valorisation'
class item(ItemElement):
klass = Investment
condition = lambda self: not CleanText('//div[has-class("errorConteneur")]', default=None)(self.el)
obj_label = Upper(TableCell('label'))
obj_quantity = MyDecimal(TableCell('quantity'))
obj_unitprice = MyDecimal(TableCell('unitprice'))
obj_unitvalue = MyDecimal(TableCell('unitvalue'))
obj_valuation = CleanDecimal(TableCell('valuation'), replace_dots=True)
obj_vdate = Date(CleanText(TableCell('vdate')), dayfirst=True, default=NotAvailable)
def obj_code(self):
if Field('label')(self) == "LIQUIDITES":
return 'XX-liquidity'
code = CleanText(TableCell('code'))(self)
return code if is_isin_valid(code) else NotAvailable
def obj_code_type(self):
return Investment.CODE_TYPE_ISIN if is_isin_valid(Field('code')(self)) else NotAvailable
class AdvisorPage(LoggedPage, JsonPage):
@method
class get_advisor(ItemElement):
klass = Advisor
obj_name = Dict('nomPrenom')
obj_email = obj_mobile = NotAvailable
def obj_phone(self):
return Dict('numeroTelephone')(self) or NotAvailable
@method
class update_agency(ItemElement):
obj_fax = CleanText(Dict('numeroFax'), replace=[(' ', '')])
obj_agency = Dict('nom')
obj_address = Format('%s %s', Dict('adresse1'), Dict('adresse3'))
class RecipientsPage(LoggedPage, JsonPage):
def get_numbers(self):
# If account information is not available when asking for the
# recipients (server error for ex.), return an empty dictionary
# that will be filled later after being returned the json of the
# account page (containing the accounts IDs too).
if 'listCompteTitulaireCotitulaire' not in self.doc and 'exception' in self.doc:
return {}
ret = {}
ret.update({
d['index']: d['numeroContratSouscrit']
for d in self.doc['listCompteTitulaireCotitulaire']
})
ret.update({
d['index']: d['numeroContratSouscrit']
for p in self.doc['listCompteMandataire'].values()
for d in p
})
ret.update({
d['index']: d['numeroContratSouscrit']
for p in self.doc['listCompteLegalRep'].values()
for d in p
})
return ret
class ProfilePage(LoggedPage, JsonPage):
# be careful, this page is used in CmsoProBrowser too!
@method
class get_profile(ItemElement):
klass = Profile
def obj_id(self):
return (Dict('identifiantExterne',default=None)(self)
or Dict('login')(self))
obj_name = Format('%s %s', Dict('firstName'), Dict('lastName'))
obj_email = Dict('email', default=NotAvailable) # can be unavailable on pro website for example