Skip to content
Commits on Source (63)
......@@ -24,7 +24,7 @@
from weboob.browser.browsers import APIBrowser
from weboob.exceptions import BrowserIncorrectPassword, BrowserBanned
from weboob.capabilities.captcha import (
ImageCaptchaJob, RecaptchaJob, NocaptchaJob, FuncaptchaJob, CaptchaError,
ImageCaptchaJob, RecaptchaJob, RecaptchaV3Job, NocaptchaJob, FuncaptchaJob, CaptchaError,
InsufficientFunds, UnsolvableCaptcha, InvalidCaptcha,
)
......@@ -74,6 +74,20 @@ def post_gcaptcha(self, url, key, prefix):
r = self.request('/createTask', data=data)
return str(r['taskId'])
def post_gcaptchav3(self, url, key, action):
data = {
"clientKey": self.apikey,
"task":{
"type":"RecaptchaV3TaskProxyless",
"websiteURL": url,
"websiteKey": key,
"minScore": 0.3,
"pageAction": action
}
}
r = self.request('/createTask', data=data)
return str(r['taskId'])
def post_funcaptcha(self, url, key, sub_domain):
data = {
"clientKey": self.apikey,
......@@ -128,7 +142,7 @@ def poll(self, job):
elif isinstance(job, RecaptchaJob):
job.solution = sol['recaptchaResponse']
job.solution_challenge = sol['recaptchaChallenge']
elif isinstance(job, NocaptchaJob):
elif isinstance(job, NocaptchaJob) or isinstance(job, RecaptchaV3Job):
job.solution = sol['gRecaptchaResponse']
elif isinstance(job, FuncaptchaJob):
job.solution = sol['token']
......
......@@ -21,7 +21,9 @@
from weboob.tools.backend import Module, BackendConfig
from weboob.capabilities.captcha import CapCaptchaSolver, ImageCaptchaJob, RecaptchaJob, NocaptchaJob, FuncaptchaJob
from weboob.capabilities.captcha import (
CapCaptchaSolver, ImageCaptchaJob, RecaptchaJob, RecaptchaV3Job, NocaptchaJob, FuncaptchaJob
)
from weboob.tools.value import ValueBackendPassword
from .browser import AnticaptchaBrowser
......@@ -53,6 +55,8 @@ def create_job(self, job):
job.id = self.browser.post_image(job.image)
elif isinstance(job, RecaptchaJob):
job.id = self.browser.post_recaptcha(job.site_url, job.site_key)
elif isinstance(job, RecaptchaV3Job):
job.id = self.browser.post_gcaptchav3(job.site_url, job.site_key, job.action)
elif isinstance(job, NocaptchaJob):
job.id = self.browser.post_nocaptcha(job.site_url, job.site_key)
elif isinstance(job, FuncaptchaJob):
......
......@@ -42,7 +42,9 @@
AccountsPage as BankAccountsPage, CBTransactionsPage, TransactionsPage,
UnavailablePage, IbanPage, LifeInsuranceIframe, BoursePage, BankProfilePage,
)
from .pages.wealth import AccountsPage as WealthAccountsPage, InvestmentPage, HistoryPage, ProfilePage
from .pages.wealth import (
AccountsPage as WealthAccountsPage, InvestmentPage, HistoryPage, ProfilePage, AccountDetailsPage,
)
from .pages.transfer import (
RecipientsPage, AddRecipientPage, ValidateTransferPage, RegisterTransferPage,
ConfirmTransferPage, RecipientConfirmationPage,
......@@ -217,6 +219,7 @@ def iter_accounts(self):
self.transactions.go()
self.cache['accs'] = accounts
self.bank_accounts.go()
return self.cache['accs']
@need_login
......@@ -505,13 +508,14 @@ def get_profile(self):
class AXAAssurance(AXABrowser):
BASEURL = 'https://espaceclient.axa.fr'
accounts = URL('/accueil.html', WealthAccountsPage)
investment = URL('/content/ecc-popin-cards/savings/[^/]+/repartition', InvestmentPage)
history = URL('.*accueil/savings/(\w+)/contract',
'https://espaceclient.axa.fr/#', HistoryPage)
documents = URL('https://espaceclient.axa.fr/content/espace-client/accueil/mes-documents/attestations-d-assurances.content-inner.din_CERTIFICATE.html', DocumentsPage)
download = URL('/content/ecc-popin-cards/technical/detailed/document.downloadPdf.html',
'/content/ecc-popin-cards/technical/detailed/document/_jcr_content/',
accounts = URL(r'/accueil.html', WealthAccountsPage)
account_details = URL('.*accueil/savings/(\w+)/contract',
r'https://espaceclient.axa.fr/#', AccountDetailsPage)
investment = URL(r'/content/ecc-popin-cards/savings/[^/]+/repartition', InvestmentPage)
history = URL(r'/content/ecc-popin-cards/savings/savings/postsales.mawGetPostSalesOperations.json', HistoryPage)
documents = URL(r'https://espaceclient.axa.fr/content/espace-client/accueil/mes-documents/attestations-d-assurances.content-inner.din_CERTIFICATE.html', DocumentsPage)
download = URL(r'/content/ecc-popin-cards/technical/detailed/document.downloadPdf.html',
r'/content/ecc-popin-cards/technical/detailed/document/_jcr_content/',
DownloadPage)
profile = URL(r'/content/ecc-popin-cards/transverse/userprofile.content-inner.html\?_=\d+', ProfilePage)
......@@ -560,18 +564,21 @@ def iter_investment(self, account):
@need_login
def iter_history(self, account):
self.go_wealth_pages(account)
pagination_url = self.page.get_pagination_url()
try:
self.location(pagination_url, params={'skip': 0})
except ClientError as e:
assert e.response.status_code == 406
self.logger.info('not doing pagination for account %r, site seems broken', account)
for tr in self.page.iter_history(no_pagination=True):
yield tr
''' There is now an API for the accounts history, however transactions are not
sorted by date in the JSON. The website fetches 5 years of history maximum.
For some accounts, the access to the transactions JSON is not available yet. '''
params = {
'startDate': (date.today() - relativedelta(years=2)).year,
'endDate': date.today().year,
'pid': account.id,
}
self.history.go(params=params)
error_code = self.page.get_error_code()
if error_code:
self.logger.warning('Error when trying to access the history JSON, history will be skipped for this account.')
return
for tr in self.page.iter_history():
for tr in sorted_transactions(self.page.iter_history()):
yield tr
def iter_coming(self, account):
......
......@@ -82,6 +82,7 @@ class AccountsPage(LoggedPage, MyHTMLPage):
('livret', Account.TYPE_SAVINGS),
('ldd', Account.TYPE_SAVINGS),
('pel', Account.TYPE_SAVINGS),
('cel', Account.TYPE_SAVINGS),
('pea', Account.TYPE_PEA),
('titres', Account.TYPE_MARKET),
('valorisation', Account.TYPE_MARKET),
......
......@@ -17,28 +17,29 @@
# 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
import re
from weboob.browser.pages import HTMLPage, LoggedPage, pagination
from weboob.browser.elements import ListElement, ItemElement, method, TableElement
from decimal import Decimal
from weboob.browser.pages import HTMLPage, LoggedPage, JsonPage
from weboob.browser.elements import ListElement, DictElement, ItemElement, method, TableElement
from weboob.browser.filters.standard import (
Async, AsyncLoad, CleanDecimal, CleanText, Currency, Date, Eval, Field, Lower, MapIn, QueryValue, Regexp,
CleanDecimal, CleanText, Currency, Date, Eval, Field, Lower, MapIn, QueryValue, Regexp,
)
from weboob.browser.filters.html import Attr, Link, TableCell
from weboob.browser.filters.json import Dict
from weboob.capabilities.bank import Account, Investment
from weboob.capabilities.profile import Person
from weboob.capabilities.base import NotAvailable, NotLoaded
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
def MyDecimal(*args, **kwargs):
kwargs.update(replace_dots=True, default=NotAvailable)
return CleanDecimal(*args, **kwargs)
def float_to_decimal(f):
return Decimal(str(f))
class AccountsPage(LoggedPage, HTMLPage):
@method
class iter_accounts(ListElement):
item_xpath = '//div[contains(@data-route, "/savings/")]'
......@@ -55,10 +56,10 @@ class item(ItemElement):
condition = lambda self: Field('balance')(self) is not NotAvailable
obj_id = Regexp(CleanText('.//span[has-class("small-title")]'), '(\d+)')
obj_id = Regexp(CleanText('.//span[has-class("small-title")]'), r'([\d/]+)')
obj_label = CleanText('.//h3[has-class("card-title")]')
obj_balance = MyDecimal('.//p[has-class("amount-card")]')
obj_valuation_diff = MyDecimal('.//p[@class="performance"]')
obj_balance = CleanDecimal.French('.//p[has-class("amount-card")]')
obj_valuation_diff = CleanDecimal.French('.//p[@class="performance"]', default=NotAvailable)
def obj_url(self):
url = Attr('.', 'data-route')(self)
......@@ -154,66 +155,40 @@ def is_detail(self):
return bool(self.doc.xpath(u'//th[contains(text(), "Valeur de la part")]'))
class Transaction(FrenchTransaction):
PATTERNS = [(re.compile(u'^(?P<text>souscription.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile(u'^(?P<text>.*)'), FrenchTransaction.TYPE_BANK),
]
class HistoryPage(LoggedPage, HTMLPage):
def build_doc(self, content):
# we got empty pages at end of pagination
if not content.strip():
content = b"<html></html>"
return super(HistoryPage, self).build_doc(content)
class AccountDetailsPage(LoggedPage, HTMLPage):
def get_account_url(self, url):
return Attr(u'//a[@href="%s"]' % url, 'data-target')(self.doc)
return Attr('//a[@href="%s"]' % url, 'data-target')(self.doc)
def get_investment_url(self):
return Attr('//div[has-class("card-distribution")]', 'data-url', default=None)(self.doc)
def get_pagination_url(self):
return Attr('//div[contains(@class, "default")][@data-module-card-list--current-page]', 'data-module-card-list--url')(self.doc)
@method
class get_investments(ListElement):
item_xpath = '//div[@class="white-bg"][.//strong[contains(text(), "support")]]/following-sibling::div'
class item(ItemElement):
klass = Investment
class Transaction(FrenchTransaction):
PATTERNS = [
(re.compile('^(?P<text>souscription.*)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile('^(?P<text>.*)'), FrenchTransaction.TYPE_BANK),
]
obj_label = CleanText('.//div[has-class("t-data__label")]')
obj_valuation = MyDecimal('.//div[has-class("t-data__amount") and has-class("desktop")]')
obj_portfolio_share = Eval(lambda x: x / 100, CleanDecimal('.//div[has-class("t-data__amount_label")]'))
@pagination
class HistoryPage(LoggedPage, JsonPage):
@method
class iter_history(ListElement):
item_xpath = '//div[contains(@data-url, "savingsdetailledcard")]'
def next_page(self):
if not CleanText(self.item_xpath, default=None)(self):
return
elif self.env.get('no_pagination'):
return
return re.sub(r'(?<=\bskip=)(\d+)', lambda m: str(int(m.group(1)) + 10), self.page.url)
class iter_history(DictElement):
class item(ItemElement):
klass = Transaction
load_details = Attr('.', 'data-url') & AsyncLoad
obj_raw = Transaction.Raw(Dict('label'))
obj_date = Date(Dict('date'))
obj_amount = Eval(float_to_decimal, Dict('gross_amount/value'))
obj_raw = Transaction.Raw('.//div[has-class("desktop")]//em')
obj_date = Date(CleanText('.//div[has-class("t-data__date") and has-class("desktop")]'), dayfirst=True)
obj_amount = MyDecimal('.//div[has-class("t-data__amount") and has-class("desktop")]')
def validate(self, obj):
return CleanText(Dict('status'))(self) == 'DONE'
def obj_investments(self):
investments = list(Async('details').loaded_page(self).get_investments())
for inv in investments:
inv.vdate = Field('date')(self)
return investments
def get_error_code(self):
# The server returns a list if it worked and a dict in case of error
if isinstance(self.doc, dict) and 'return' in self.doc:
return self.doc['return']['error']['code']
return None
class ProfilePage(LoggedPage, HTMLPage):
......
......@@ -187,18 +187,39 @@ def __init__(self, website, *args, **kwargs):
self.investments = {}
# HACK, the website may crash with legacy passwords (legacy means not only digits)
# If the website crashes and if we have a legacy password, we raise WrongPass instead of BrowserUnavailable
self.is_password_only_digits = None
def deinit(self):
super(BanquePopulaire, self).deinit()
self.linebourse.deinit()
no_login = 0
def follow_back_button_if_any(self, params=None, actions=None):
"""
Look for a Retour button and follow it using a POST
:param params: Optional form params to use (default: call self.page.get_params())
:param actions: Optional actions to use (default: call self.page.get_button_actions())
:return: None
"""
if not self.page:
return
data = self.page.get_back_button_params(params=params, actions=actions)
if data:
self.location('/cyber/internet/ContinueTask.do', data=data)
@no_need_login
def do_login(self):
self.location(self.BASEURL)
# avoids trying to relog in while it's already on home page
if self.home_page.is_here():
return
self.is_password_only_digits = self.password.isdigit()
self.page.login(self.username, self.password)
if self.login_page.is_here():
raise BrowserIncorrectPassword()
......@@ -231,6 +252,9 @@ def go_on_accounts_list(self):
form['token'] = self.page.build_token(form['token'])
form.submit()
# In case of prevAction maybe we have reached an expanded accounts list page, need to go back
self.follow_back_button_if_any()
@retry(LoggedOut)
@need_login
def get_accounts_list(self, get_iban=True):
......@@ -301,13 +325,7 @@ def set_gocardless_transaction_details(self, transaction):
transaction.raw = '%s %s' % (transaction.raw, ref)
# Needed to preserve navigation.
btn = self.page.doc.xpath('.//button[span[text()="Retour"]]')
if len(btn):
_data = self.page.get_params()
actions = self.page.get_button_actions()
_data.update(actions[btn[0].attrib['id']])
_data['token'] = self.page.build_token(_data['token'])
self.location('/cyber/internet/ContinueTask.do', data=_data)
self.follow_back_button_if_any()
@retry(LoggedOut)
@need_login
......
......@@ -164,6 +164,19 @@ def get_button_actions(self):
}
return actions
def get_back_button_params(self, params=None, actions=None):
btn = self.doc.xpath('.//button[span[text()="Retour"]]')
if not btn:
return
params = params or self.get_params()
actions = actions or self.get_button_actions()
key = btn[0].attrib['id']
assert actions.get(key), "Key %s not found in actions %s" % (key, actions) # Currently it never happens
params.update(actions[key])
params['token'] = self.build_token(params['token'])
return params
class MyHTMLPage(BasePage, HTMLPage):
def build_doc(self, data, *args, **kwargs):
......@@ -258,6 +271,10 @@ def on_load(self):
class ErrorPage(LoggedPage, MyHTMLPage):
def on_load(self):
# HACK: some accounts with legacy password fails, people needs to update it
if not self.browser.is_password_only_digits:
raise BrowserIncorrectPassword()
if CleanText('//script[contains(text(), "momentanément indisponible")]')(self.doc):
raise BrowserUnavailable(u"Le service est momentanément indisponible")
elif CleanText('//h1[contains(text(), "Cette page est indisponible")]')(self.doc):
......@@ -630,6 +647,7 @@ def iter_accounts(self, next_pages):
if len(tds) >= 5 and len(tds[self.COL_COMING].xpath('.//a')) > 0:
_params = account._params.copy()
_params['dialogActionPerformed'] = 'ENCOURS_COMPTE'
_params['attribute($SEL_$%s)' % tr.attrib['id'].split('_')[0]] = tr.attrib['id'].split('_', 1)[1]
# If there is an action needed before going to the cards page, save it.
m = re.search('dialogActionPerformed=([\w_]+)', self.url)
......@@ -645,11 +663,7 @@ def iter_accounts(self, next_pages):
yield account
# Needed to preserve navigation.
btn = self.doc.xpath('.//button[span[text()="Retour"]]')
if len(btn) > 0:
_params = params.copy()
_params.update(actions[btn[0].attrib['id']])
self.browser.open('/cyber/internet/ContinueTask.do', data=_params)
self.browser.follow_back_button_if_any(params=params.copy(), actions=actions)
class AccountsFullPage(AccountsPage):
......@@ -746,12 +760,7 @@ def iter_accounts(self, next_pages):
yield account
# Needed to preserve navigation.
btn = self.doc.xpath('.//button[span[text()="Retour"]]')
if len(btn) > 0:
actions = self.get_button_actions()
_params = params.copy()
_params.update(actions[btn[0].attrib['id']])
self.browser.open('/cyber/internet/ContinueTask.do', data=_params)
self.browser.follow_back_button_if_any(params=params.copy())
class Transaction(FrenchTransaction):
......@@ -842,6 +851,7 @@ def get_account_history(self):
debit = cleaner(tds[self.COL_DEBIT])
credit = cleaner(tds[self.COL_CREDIT])
t.bdate = Date(dayfirst=True).filter(cleaner(tds[self.COL_COMPTA_DATE]))
t.parse(date, re.sub(r'[ ]+', ' ', raw), vdate)
t.set_amount(credit, debit)
t._amount_type = 'debit' if t.amount == debit else 'credit'
......@@ -900,7 +910,7 @@ def get_card_history(self, account, coming):
t.parse(debit_date, re.sub(r'[ ]+', ' ', label))
t.set_amount(amount)
t.rdate = t.parse_date(date)
t.rdate = t.bdate = t.parse_date(date)
t.original_currency = currency
if not t.type:
t.type = Transaction.TYPE_DEFERRED_CARD
......
......@@ -38,6 +38,7 @@
from weboob.browser.elements import DataError
from weboob.exceptions import BrowserIncorrectPassword
from weboob.tools.value import Value, ValueBool
from weboob.tools.capabilities.bank.investments import create_french_liquidity
from .pages import (
LoginPage, AccountsPage, AccountsIBANPage, HistoryPage, TransferInitPage,
......@@ -286,8 +287,8 @@ def iter_coming_operations(self, account):
@need_login
def iter_investment(self, account):
if account.type == Account.TYPE_PEA and account.label.endswith('Espèces'):
return []
if account.type == Account.TYPE_PEA and 'espèces' in account.label.lower():
return [create_french_liquidity(account.balance)]
# Life insurances and PERP may be scraped from the API or from the "Assurance Vie" space,
# so we need to discriminate between both using account._details:
......
......@@ -219,6 +219,7 @@ def is_here(self):
'comptes courants': Account.TYPE_CHECKING,
'cav': Account.TYPE_CHECKING,
'livret': Account.TYPE_SAVINGS,
'livret-a': Account.TYPE_SAVINGS,
'pel': Account.TYPE_SAVINGS,
'cel': Account.TYPE_SAVINGS,
'ldd': Account.TYPE_SAVINGS,
......
# -*- coding: utf-8 -*-
# Copyright(C) 2019 Budget Insight
from __future__ import unicode_literals
from .module import BouyguesModule
__all__ = ['BouyguesModule']
# -*- coding: utf-8 -*-
# Copyright(C) 2010-2015 Bezleputh
# Copyright(C) 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 Affero General Public License as published by
# 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 Affero General Public License for more details.
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# 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 time import time
from jose import jwt
from weboob.browser import LoginBrowser, URL, need_login
from weboob.exceptions import BrowserIncorrectPassword, BrowserUnavailable
from weboob.browser.exceptions import ClientError, HTTPNotFound
from weboob.tools.compat import urlparse, parse_qs
from weboob.browser.exceptions import HTTPNotFound
from weboob.tools.compat import urlparse, parse_qsl
from .pages import (
DocumentsPage, HomePage, LoginPage, SubscriberPage, SubscriptionPage, SubscriptionDetailPage,
SendSMSPage, SendSMSErrorPage, UselessPage, DocumentFilePage, ProfilePage,
LoginPage, AppConfigPage, SubscriberPage, SubscriptionPage, SubscriptionDetail, DocumentPage, DocumentDownloadPage,
DocumentFilePage,
)
from weboob.capabilities.messages import CantSendMessage
__all__ = ['BouyguesBrowser']
class MyURL(URL):
def go(self, *args, **kwargs):
kwargs['id_personne'] = self.browser.id_personne
kwargs['headers'] = self.browser.headers
return super(MyURL, self).go(*args, **kwargs)
class BouyguesBrowser(LoginBrowser):
BASEURL = 'https://api.bouyguestelecom.fr'
TIMEOUT = 20
login = URL(r'https://www.mon-compte.bouyguestelecom.fr/cas/login', LoginPage)
home = URL(r'https://www.bouyguestelecom.fr/mon-compte', HomePage)
subscriber = URL(r'/personnes/(?P<idUser>\d+)$', SubscriberPage)
subscriptions = URL(r'/personnes/(?P<idUser>\d+)/comptes-facturation', SubscriptionPage)
subscriptions_details = URL(r'/comptes-facturation/(?P<idSub>\d+)/contrats-payes', SubscriptionDetailPage)
document_file = URL(r'/comptes-facturation/(?P<idSub>\d+)/factures/.*/documents', DocumentFilePage)
documents = URL(r'/comptes-facturation/(?P<idSub>\d+)/factures', DocumentsPage)
sms_page = URL(r'https://www.secure.bbox.bouyguestelecom.fr/services/SMSIHD/sendSMS.phtml',
r'https://www.secure.bbox.bouyguestelecom.fr/services/SMSIHD/confirmSendSMS.phtml',
SendSMSPage)
confirm = URL(r'https://www.secure.bbox.bouyguestelecom.fr/services/SMSIHD/resultSendSMS.phtml', UselessPage)
sms_error_page = URL(r'https://www.secure.bbox.bouyguestelecom.fr/services/SMSIHD/SMS_erreur.phtml',
SendSMSErrorPage)
profile = URL(r'/personnes/(?P<idUser>\d+)/coordonnees', ProfilePage)
login_page = URL(r'https://www.mon-compte.bouyguestelecom.fr/cas/login', LoginPage)
app_config = URL(r'https://www.bouyguestelecom.fr/mon-compte/data/app-config.json', AppConfigPage)
subscriber_page = MyURL(r'/personnes/(?P<id_personne>\d+)$', SubscriberPage)
subscriptions_page = MyURL(r'/personnes/(?P<id_personne>\d+)/comptes-facturation', SubscriptionPage)
subscription_detail_page = URL(r'/comptes-facturation/(?P<id_account>\d+)/contrats-payes', SubscriptionDetail)
document_file_page = URL(r'/comptes-facturation/(?P<id_account>\d+)/factures/.*/documents/.*', DocumentFilePage)
documents_page = URL(r'/comptes-facturation/(?P<id_account>\d+)/factures(\?|$)', DocumentPage)
document_download_page = URL(r'/comptes-facturation/(?P<id_account>\d+)/factures/.*(\?|$)', DocumentDownloadPage)
def __init__(self, username, password, lastname, *args, **kwargs):
super(BouyguesBrowser, self).__init__(username, password, *args, **kwargs)
self.lastname = lastname
self.id_personne = None
self.headers = None
self.id_user = None
def do_login(self):
self.login.go()
if self.home.is_here():
return
self.login_page.go()
self.page.login(self.username, self.password, self.lastname)
if self.login.is_here():
error = self.page.get_error()
if error and 'mot de passe' in error:
raise BrowserIncorrectPassword(error)
raise AssertionError("Unhandled error at login: {}".format(error))
# q is timestamp millisecond
self.app_config.go(params={'q': int(time()*1000)})
client_id = self.page.get_client_id()
# after login we need to get some tokens to use bouygues api
data = {
params = {
'client_id': client_id,
'response_type': 'id_token token',
'client_id': 'a360.bouyguestelecom.fr',
'redirect_uri': 'https://www.bouyguestelecom.fr/mon-compte/'
}
self.location('https://oauth2.bouyguestelecom.fr/authorize', params=data)
parsed_url = urlparse(self.response.url)
fragment = parse_qs(parsed_url.fragment)
if not fragment:
query = parse_qs(parsed_url.query)
if 'server_error' in query.get('error', []):
raise BrowserUnavailable(query['error_description'][0])
claims = jwt.get_unverified_claims(fragment['id_token'][0])
self.headers = {'Authorization': 'Bearer %s' % fragment['access_token'][0]}
self.id_user = claims['id_personne']
self.location('https://oauth2.bouyguestelecom.fr/authorize', params=params)
fragments = dict(parse_qsl(urlparse(self.url).fragment))
@need_login
def post_message(self, message):
self.sms_page.go()
if self.sms_error_page.is_here():
raise CantSendMessage(self.page.get_error_message())
receivers = ";".join(message.receivers) if message.receivers else self.username
self.page.send_sms(message, receivers)
if self.sms_error_page.is_here():
raise CantSendMessage(self.page.get_error_message())
self.confirm.open()
self.id_personne = jwt.get_unverified_claims(fragments['id_token'])['id_personne']
authorization = 'Bearer ' + fragments['access_token']
self.headers = {'Authorization': authorization}
@need_login
def iter_subscriptions(self):
self.subscriber.go(idUser=self.id_user, headers=self.headers)
subscriber = self.page.get_subscriber()
phone_list = self.page.get_phone_list()
self.subscriptions.go(idUser=self.id_user, headers=self.headers)
for sub in self.page.iter_subscriptions(subscriber=subscriber):
try:
self.subscriptions_details.go(idSub=sub.id, headers=self.headers)
sub.label = self.page.get_label()
sub._is_holder = self.page.is_holder()
except ClientError:
# if another person pay for your subscription you may not have access to this page with your credentials
sub.label = phone_list
if not sub.label:
if not sub._is_holder:
sub.label = subscriber
else:
# If the subscriber is the holder but the subscription does not have a phone number anyway
# It means that the subscription has not been activated yet
continue
subscriber = self.subscriber_page.go().get_subscriber()
self.subscriptions_page.go()
for sub in self.page.iter_subscriptions():
sub.subscriber = subscriber
sub.label = self.subscription_detail_page.go(id_account=sub.id, headers=self.headers).get_label()
yield sub
@need_login
def iter_documents(self, subscription):
try:
self.location(subscription.url, headers=self.headers)
return self.page.iter_documents(subid=subscription.id)
except HTTPNotFound as error:
if error.response.json()['error'] in ('facture_introuvable', 'compte_jamais_facture'):
json_response = error.response.json()
if json_response['error'] in ('facture_introuvable', 'compte_jamais_facture'):
return []
raise
return self.page.iter_documents(subid=subscription.id)
@need_login
def download_document(self, document):
self.location(document.url, headers=self.headers)
return self.open(self.page.get_one_shot_download_url()).content
@need_login
def get_profile(self):
self.subscriber.go(idUser=self.id_user, headers=self.headers)
subscriber = self.page.get_subscriber()
self.profile.go(idUser=self.id_user, headers=self.headers)
return self.page.get_profile(subscriber=subscriber)
return self.location(document.url, headers=self.headers).content
# -*- coding: utf-8 -*-
# Copyright(C) 2010-2015 Bezleputh
# Copyright(C) 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 Affero General Public License as published by
# 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 Affero General Public License for more details.
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# 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.capabilities.bill import CapDocument, Subscription, Document, SubscriptionNotFound, DocumentNotFound
from weboob.capabilities.messages import CantSendMessage, CapMessages, CapMessagesPost
from weboob.capabilities.base import find_object
from weboob.capabilities.profile import CapProfile
from weboob.tools.backend import Module, BackendConfig
from weboob.tools.value import ValueBackendPassword, Value
from weboob.capabilities.base import find_object
from weboob.capabilities.bill import CapDocument, Document, SubscriptionNotFound, Subscription, DocumentNotFound
from weboob.tools.value import Value, ValueBackendPassword
from .browser import BouyguesBrowser
......@@ -32,25 +31,20 @@
__all__ = ['BouyguesModule']
class BouyguesModule(Module, CapMessages, CapMessagesPost, CapDocument, CapProfile):
class BouyguesModule(Module, CapDocument):
NAME = 'bouygues'
MAINTAINER = 'Bezleputh'
EMAIL = 'carton_ben@yahoo.fr'
DESCRIPTION = 'Bouygues Télécom'
MAINTAINER = 'Florian Duguet'
EMAIL = 'florian.duguet@budget-insight.com'
LICENSE = 'LGPLv3+'
VERSION = '1.6'
DESCRIPTION = u'Bouygues Télécom French mobile phone provider'
LICENSE = 'AGPLv3+'
CONFIG = BackendConfig(Value('login', label='E-mail / N° de Téléphone'),
CONFIG = BackendConfig(Value('login', label='Numéro de mobile, de clé/tablette ou e-mail en @bbox.fr'),
ValueBackendPassword('password', label='Mot de passe'),
ValueBackendPassword('lastname', label='Nom de famille', default=u''))
ValueBackendPassword('lastname', label='Nom de famille', default=''))
BROWSER = BouyguesBrowser
def create_default_browser(self):
return self.create_browser(username=self.config['login'].get(), password=self.config['password'].get(), lastname=self.config['lastname'].get())
def post_message(self, message):
if not message.content.strip():
raise CantSendMessage('Message content is empty.')
self.browser.post_message(message)
return self.create_browser(self.config['login'].get(), self.config['password'].get(), self.config['lastname'].get())
def iter_subscription(self):
return self.browser.iter_subscriptions()
......@@ -58,20 +52,17 @@ def iter_subscription(self):
def get_subscription(self, _id):
return find_object(self.iter_subscription(), id=_id, error=SubscriptionNotFound)
def get_document(self, _id):
subid = _id.rsplit('_', 1)[0]
subscription = self.get_subscription(subid)
return find_object(self.iter_documents(subscription), id=_id, error=DocumentNotFound)
def iter_documents(self, subscription):
if not isinstance(subscription, Subscription):
subscription = self.get_subscription(subscription)
return self.browser.iter_documents(subscription)
def get_document(self, _id):
subid = _id.rsplit('_', 1)[0]
subscription = self.get_subscription(subid)
return find_object(self.iter_documents(subscription), id=_id, error=DocumentNotFound)
def download_document(self, document):
if not isinstance(document, Document):
document = self.get_document(document)
return self.browser.download_document(document)
def get_profile(self):
return self.browser.get_profile()
# -*- coding: utf-8 -*-
# Copyright(C) 2010-2015 Bezleputh
# Copyright(C) 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 Affero General Public License as published by
# 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 Affero General Public License for more details.
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# 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
import re
from datetime import datetime, timedelta
from weboob.capabilities.messages import CantSendMessage
from weboob.exceptions import BrowserIncorrectPassword, ParseError
from datetime import timedelta
from weboob.capabilities.base import NotLoaded
from weboob.capabilities.bill import Bill, Subscription
from weboob.capabilities.profile import Profile
from weboob.browser.pages import HTMLPage, JsonPage, LoggedPage, PDFPage
from weboob.browser.filters.json import Dict
from weboob.browser.filters.standard import CleanDecimal, CleanText, Env, Format, Regexp
from weboob.browser.elements import DictElement, ItemElement, method
from weboob.browser.filters.json import Dict
from weboob.browser.pages import HTMLPage, JsonPage, LoggedPage, RawPage
from weboob.capabilities import NotAvailable
from weboob.capabilities.bill import Subscription, Bill
from weboob.browser.filters.standard import Date, CleanDecimal, Env, Format
from weboob.exceptions import BrowserIncorrectPassword
class LoginPage(HTMLPage):
def login(self, login, password, lastname):
form = self.get_form(id='log_data')
form['username'] = login
def login(self, username, password, lastname):
form = self.get_form()
form['username'] = username
form['password'] = password
if 'lastname' in form:
if not lastname:
raise BrowserIncorrectPassword('Le nom de famille est obligatoire.')
raise BrowserIncorrectPassword('Veuillez renseigner votre nom de famille.')
form['lastname'] = lastname
form.submit()
def get_error(self):
return CleanText('//div[@id="alert_msg"]//p')(self.doc)
class HomePage(LoggedPage, HTMLPage):
pass
class AppConfigPage(JsonPage):
def get_client_id(self):
return self.doc['config']['oauth']['clientId']
class SubscriberPage(LoggedPage, JsonPage):
def get_subscriber(self):
if self.doc['type'] == 'INDIVIDU':
sub_dict = self.doc
else:
sub_dict = self.doc['representantLegal']
return "%s %s %s" % (sub_dict['civilite'], sub_dict['prenom'], sub_dict['nom'])
def get_phone_list(self):
num_tel_list = []
for phone in self.doc.get('comptesAcces', []):
num_tel_list.append(' '.join(phone[i:i + 2] for i in range(0, len(phone), 2)))
return ' - '.join(num_tel_list)
class SubscriptionPage(LoggedPage, JsonPage):
@method
class iter_subscriptions(DictElement):
item_xpath = 'items'
assert self.doc['type'] in ('INDIVIDU', 'ENTREPRISE'), "%s is unknown" % self.doc['type']
class item(ItemElement):
klass = Subscription
if self.doc['type'] == 'INDIVIDU':
subscriber_dict = self.doc
elif self.doc['type'] == 'ENTREPRISE':
subscriber_dict = self.doc['representantLegal']
obj_id = Dict('id')
obj_url = Dict('_links/factures/href')
obj_subscriber = Env('subscriber')
return '%s %s %s' % (subscriber_dict['civilite'], subscriber_dict['prenom'], subscriber_dict['nom'])
class SubscriptionDetailPage(LoggedPage, JsonPage):
class SubscriptionDetail(LoggedPage, JsonPage):
def get_label(self):
label_list = []
for s in self.doc['items']:
......@@ -96,46 +74,33 @@ def get_label(self):
return ' - '.join(label_list)
def is_holder(self):
return any(CleanText(Dict('utilisateur/libelleProfilDroits'), default=None)(s) == 'Accès titulaire' for s in self.doc['items'] if 'utilisateur' in s)
class SendSMSPage(HTMLPage):
def send_sms(self, message, receivers):
sms_number = CleanDecimal(Regexp(CleanText('//span[@class="txt12-o"][1]/strong'), r'(\d*) SMS.*'))(self.doc)
if sms_number == 0:
msg = CleanText('//span[@class="txt12-o"][1]')(self.doc)
raise CantSendMessage(msg)
form = self.get_form('//form[@name="formSMS"]')
form["fieldMsisdn"] = receivers
form["fieldMessage"] = message.content
class SubscriptionPage(LoggedPage, JsonPage):
@method
class iter_subscriptions(DictElement):
item_xpath = 'items'
form.submit()
class item(ItemElement):
klass = Subscription
obj_id = Dict('id')
obj_url = Dict('_links/factures/href')
class SendSMSErrorPage(HTMLPage):
def get_error_message(self):
return CleanText('//span[@class="txt12-o"][1]')(self.doc)
class MyDate(Date):
"""
some date are datetime and contains date at GMT, and always at 22H or 23H
but date inside PDF file is at GMT +1H or +2H (depends of summer or winter hour)
so we add one day and skip time to get good date
"""
def filter(self, txt):
date = super(MyDate, self).filter(txt)
if date:
date += timedelta(days=1)
return date
class DocumentsPage(LoggedPage, JsonPage):
FRENCH_MONTHS = {
1: 'Janvier',
2: 'Février',
3: 'Mars',
4: 'Avril',
5: 'Mai',
6: 'Juin',
7: 'Juillet',
8: 'Août',
9: 'Septembre',
10: 'Octobre',
11: 'Novembre',
12: 'Décembre',
}
class DocumentPage(LoggedPage, JsonPage):
@method
class iter_documents(DictElement):
item_xpath = 'items'
......@@ -144,76 +109,22 @@ class item(ItemElement):
klass = Bill
obj_id = Format('%s_%s', Env('subid'), Dict('idFacture'))
def obj_url(self):
try:
link = Dict('_links/facturePDF/href')(self)
except ParseError:
# yes, sometimes it's just a misspelling word, but just sometimes...
link = Dict('_links/facturePDFDF/href')(self)
return 'https://api.bouyguestelecom.fr%s' % link
obj_date = Env('date')
obj_duedate = Env('duedate')
obj_format = 'pdf'
obj_label = Env('label')
obj_price = CleanDecimal(Dict('mntTotFacture'))
obj_url = Dict('_links/facturePDF/href')
obj_date = MyDate(Dict('dateFacturation'))
obj_duedate = MyDate(Dict('dateLimitePaieFacture', default=NotAvailable), default=NotAvailable)
obj_label = Format('Facture %s', Dict('idFacture'))
obj_format = 'pdf'
obj_currency = 'EUR'
def parse(self, el):
bill_date = datetime.strptime(Dict('dateFacturation')(self), "%Y-%m-%dT%H:%M:%SZ").date()
# dateFacturation is like: 'YYYY-MM-DDTHH:00:00Z' where Z is UTC time and HH 23 in winter and 22 in summer
# which always correspond to the day after at midnight in French time zone
# so we remove hour and consider the day after as date (which is also the date inside pdf)
self.env['date'] = bill_date + timedelta(days=1)
duedate = Dict('dateLimitePaieFacture', default=NotLoaded)(self)
if duedate:
self.env['duedate'] = datetime.strptime(duedate, "%Y-%m-%dT%H:%M:%SZ").date() + timedelta(days=1)
else:
# for some connections we don't have duedate (why ?)
self.env['duedate'] = NotLoaded
self.env['label'] = "%s %d" % (self.page.FRENCH_MONTHS[self.env['date'].month], self.env['date'].year)
def get_one_shot_download_url(self):
return self.doc['_actions']['telecharger']['action']
class ProfilePage(LoggedPage, JsonPage):
def get_profile(self, subscriber):
data = self.doc
last_address = data['adressesPostales'][0]
for address in data['adressesPostales']:
if address['dateMiseAJour'] > last_address['dateMiseAJour']:
last_address = address
p = Profile()
p.name = subscriber
p.address = '%s %s %s %s' % (last_address['numero'], last_address['rue'],
last_address['codePostal'], last_address['ville'])
p.country = last_address['pays']
for email in data['emails']:
if email['emailPrincipal']:
p.email = email['email']
break
if 'telephones' in data:
for phone in data['telephones']:
if phone['telephonePrincipal']:
p.phone = phone['numero']
break
return p
class UselessPage(HTMLPage):
pass
class DocumentDownloadPage(LoggedPage, JsonPage):
def on_load(self):
# this url change each time we want to download document, (the same one or another)
self.browser.location(self.doc['_actions']['telecharger']['action'])
class DocumentFilePage(PDFPage):
class DocumentFilePage(LoggedPage, RawPage):
# since url of this file is almost the same than url of DocumentDownloadPage (which is a JsonPage)
# we have to define it to avoid mismatching
pass
......@@ -75,7 +75,7 @@ class BPBrowser(LoginBrowser, StatesMixin):
'/voscomptes/canalXHTML/pret/encours/detaillerOffrePretConsoListe-encoursPrets.ea',
'/voscomptes/canalXHTML/pret/creditRenouvelable/init-consulterCreditRenouvelable.ea',
'/voscomptes/canalXHTML/pret/encours/rechercherPret-encoursPrets.ea',
'/voscomptes/canalXHTML/sso/commun/init-integration.ea\?partenaire',
'/voscomptes/canalXHTML/sso/commun/init-integration.ea\?partenaire=cristalCEC',
'/voscomptes/canalXHTML/sso/lbpf/souscriptionCristalFormAutoPost.jsp',
AccountList)
par_accounts_revolving = URL('https://espaceclientcreditconso.labanquepostale.fr/sav/accueil.do', AccountList)
......@@ -200,9 +200,9 @@ def deinit(self):
super(BPBrowser, self).deinit()
self.linebourse.deinit()
def location(self, url, **kwargs):
def open(self, *args, **kwargs):
try:
return super(BPBrowser, self).location(url, **kwargs)
return super(BPBrowser, self).open(*args, **kwargs)
except ServerError as err:
if "/../" not in err.response.url:
raise
......@@ -211,7 +211,7 @@ def location(self, url, **kwargs):
self.logger.debug('site has "/../" in their url, fixing url manually')
parts = list(urlsplit(err.response.url))
parts[2] = os.path.abspath(parts[2])
return self.location(urlunsplit(parts))
return self.open(urlunsplit(parts))
def do_login(self):
self.location(self.login_url)
......
......@@ -50,7 +50,12 @@ class item_account_generic(ItemElement):
klass = Account
def condition(self):
return len(self.el.xpath('.//span[@class="number"]')) > 0
# For some loans the following xpath is absent and we don't want to skip them
# Also a case of loan that is empty and has no information exists and will be ignored
return (len(self.el.xpath('.//span[@class="number"]')) > 0 or
(Field('type')(self) == Account.TYPE_LOAN and
(len(self.el.xpath('.//div//*[contains(text(),"pas la restitution de ces données.")]')) == 0 and
len(self.el.xpath('.//div[@class="amount"]/span[contains(text(), "Contrat résilié")]')) == 0)))
obj_id = CleanText('.//abbr/following-sibling::text()')
obj_currency = Currency('.//span[@class="number"]')
......@@ -70,7 +75,10 @@ def obj_label(self):
def obj_balance(self):
if Field('type')(self) == Account.TYPE_LOAN:
return -abs(CleanDecimal('.//span[@class="number"]', replace_dots=True)(self))
balance = CleanDecimal('.//span[@class="number"]', replace_dots=True, default=NotAvailable)(self)
if balance:
balance = -abs(balance)
return balance
return CleanDecimal('.//span[@class="number"]', replace_dots=True, default=NotAvailable)(self)
def obj_coming(self):
......@@ -102,7 +110,7 @@ def obj_coming(self):
return NotAvailable
def obj_iban(self):
if not Field('url')(self):
if not Field('url')(self) or Field('type')(self) == Account.TYPE_LOAN:
return NotAvailable
details_page = self.page.browser.open(Field('url')(self)).page
......@@ -192,9 +200,9 @@ def get_revolving_attributes(self, account):
loan.currency = account.currency
loan.url = account.url
loan.available_amount = CleanDecimal('//tr[td[contains(text(), "Montant Maximum Autorisé") or contains(text(), "Montant autorisé")]]/td[2]')(self.doc)
loan.used_amount = loan.used_amount = CleanDecimal('//tr[td[contains(text(), "Montant Utilisé") or contains(text(), "Montant utilisé")]]/td[2]')(self.doc)
loan.available_amount = CleanDecimal(Regexp(CleanText('//tr[td[contains(text(), "Montant Disponible") or contains(text(), "Montant disponible")]]/td[2]'), r'(.*) au'))(self.doc)
loan.used_amount = CleanDecimal.US('//tr[td[contains(text(), "Montant Utilisé") or contains(text(), "Montant utilisé")]]/td[2]')(self.doc)
loan.available_amount = CleanDecimal.US(Regexp(CleanText('//tr[td[contains(text(), "Montant Disponible") or contains(text(), "Montant disponible")]]/td[2]'), r'(.*) au'))(self.doc)
loan.balance = -loan.used_amount
loan._has_cards = False
loan.type = Account.TYPE_REVOLVING_CREDIT
return loan
......
......@@ -117,10 +117,12 @@ class AccountsPage(MyJsonPage):
'025': Account.TYPE_SAVINGS, # Livret Fidélis
'027': Account.TYPE_SAVINGS, # Livret A
'037': Account.TYPE_SAVINGS,
'070': Account.TYPE_SAVINGS, # Compte Epargne Logement
'077': Account.TYPE_SAVINGS, # Livret Bambino
'078': Account.TYPE_SAVINGS, # Livret jeunes
'080': Account.TYPE_SAVINGS, # Plan épargne logement
'081': Account.TYPE_SAVINGS,
'086': Account.TYPE_SAVINGS, # Compte épargne Moisson
'097': Account.TYPE_CHECKING, # Solde en devises
'730': Account.TYPE_DEPOSIT, # Compte à terme Optiplus
'999': Account.TYPE_MARKET, # no label, we use 'Portefeuille Titres' if needed
......
......@@ -143,6 +143,7 @@ def __init__(self, nuser, *args, **kwargs):
self.BASEURL = 'https://%s' % self.BASEURL
self.is_cenet_website = False
self.new_website = True
self.multi_type = False
self.accounts = None
self.loans = None
......@@ -433,32 +434,75 @@ def get_accounts_list(self):
assert self.linebourse.portfolio.is_here()
# We must declare "page" because this URL also matches MarketPage
account.valuation_diff = page.get_valuation_diff()
# We need to go back to the synthesis, else we can not go home later
self.home_tache.go(tache='CPTSYNT0')
else:
assert False, "new domain that hasn't been seen so far ?"
"""
Card cases are really tricky on the new website.
There are 2 kinds of page where we can find cards information
- CardsPage: List some of the PSU cards
- CardsComingPage: On the coming transaction page (for a specific checking account),
we can find all cards related to this checking account. Information to reach this
CC is in the home page
We have to go through this both kind of page for those reasons:
- If there is no coming yet, the card will not be found in the home page and we will not
be able to reach the CardsComingPage. But we can find it on CardsPage
- Some cards are only on the CardsComingPage and not the CardsPage
- In CardsPage, there are cards (with "Business" in the label) without checking account on the
website (neither history nor coming), so we skip them.
- Some card on the CardsPage that have a checking account parent, but if we follow the link to
reach it with CardsComingPage, we find an other card that not in CardsPage.
"""
if self.new_website:
for account in self.accounts:
# Adding card's account that we find in CardsComingPage of each Checking account
if account._card_links:
self.home.go()
self.page.go_history(account._card_links)
for card in self.page.iter_cards():
card.parent = account
card._coming_info = self.page.get_card_coming_info(card.number, card.parent._card_links.copy())
self.accounts.append(card)
self.home.go()
self.page.go_list()
self.page.go_cards()
if self.cards.is_here() or self.cards_old.is_here():
cards = list(self.page.iter_cards())
for card in cards:
# We are on the new website. We already added some card, but we can find more of them on the CardsPage
if self.cards.is_here():
for card in self.page.iter_cards():
card.parent = find_object(self.accounts, number=card._parent_id)
assert card.parent, 'card account %r parent was not found' % card
assert card.parent, 'card account parent %s was not found' % card
# If we already added this card, we don't have to add it a second time
if find_object(self.accounts, number=card.number):
continue
info = card.parent._card_links
# If we are in the new site, we have to get each card coming transaction link.
if self.cards.is_here():
for card in cards:
info = card.parent._card_links
# If card.parent._card_links is not filled, it mean this checking account
# has no coming transactions.
card._coming_info = None
if info:
self.page.go_list()
self.page.go_history(info)
card._coming_info = self.page.get_card_coming_info(card.number, info.copy())
# If info is filled, that mean there are comings transaction
card._coming_info = None
if info:
self.page.go_list()
self.page.go_history(info)
card._coming_info = self.page.get_card_coming_info(card.number, info.copy())
if not card._coming_info:
self.logger.warning('Skip card %s (not found on checking account)', card.number)
continue
self.accounts.append(card)
self.accounts.extend(cards)
# We are on the old website. We add all card that we can find on the CardsPage
elif self.cards_old.is_here():
for card in self.page.iter_cards():
card.parent = find_object(self.accounts, number=card._parent_id)
assert card.parent, 'card account parent %s was not found' % card.number
self.accounts.append(card)
# Some accounts have no available balance or label and cause issues
# in the backend so we must exclude them from the accounts list:
......@@ -606,8 +650,17 @@ def _get_history_invests(self, account):
if self.page.is_account_inactive(account.id):
self.logger.warning('Account %s %s is inactive.' % (account.label, account.id))
return []
# There is (currently ?) no history for MILLEVIE PREMIUM accounts
if "MILLEVIE" in account.label:
self.page.go_life_insurance(account)
try:
self.page.go_life_insurance(account)
except ServerError as ex:
if ex.response.status_code == 500 and 'MILLEVIE PREMIUM' in account.label:
self.logger.info("Can not reach history page for MILLEVIE PREMIUM account")
return []
raise
label = account.label.split()[-1]
try:
self.natixis_life_ins_his.go(id1=label[:3], id2=label[3:5], id3=account.id)
......@@ -730,6 +783,9 @@ def get_investment(self, account):
self.update_linebourse_token()
for investment in self.linebourse.iter_investments(account.id):
yield investment
# We need to go back to the synthesis, else we can not go home later
self.home_tache.go(tache='CPTSYNT0')
return
elif account.type in (Account.TYPE_LIFE_INSURANCE, Account.TYPE_CAPITALISATION):
......@@ -737,7 +793,14 @@ def get_investment(self, account):
self.logger.warning('Account %s %s is inactive.' % (account.label, account.id))
return
if "MILLEVIE" in account.label:
self.page.go_life_insurance(account)
try:
self.page.go_life_insurance(account)
except ServerError as ex:
if ex.response.status_code == 500 and 'MILLEVIE PREMIUM' in account.label:
self.logger.info("Can not reach investment page for MILLEVIE PREMIUM account")
return
raise
label = account.label.split()[-1]
self.natixis_life_ins_inv.go(id1=label[:3], id2=label[3:5], id3=account.id)
for tr in self.page.get_investments():
......
......@@ -30,7 +30,10 @@
from weboob.browser.pages import LoggedPage, HTMLPage, JsonPage, pagination, FormNotFound
from weboob.browser.elements import ItemElement, method, ListElement, TableElement, SkipItem, DictElement
from weboob.browser.filters.standard import Date, CleanDecimal, Regexp, CleanText, Env, Upper, Field, Eval, Format, Currency
from weboob.browser.filters.standard import (
Date, CleanDecimal, Regexp, CleanText, Env, Upper,
Field, Eval, Format, Currency, Coalesce,
)
from weboob.browser.filters.html import Link, Attr, TableCell
from weboob.capabilities import NotAvailable
from weboob.capabilities.bank import (
......@@ -291,6 +294,7 @@ def _add_account(self, accounts, link, label, account_type, balance, number=None
return
account = Account()
account._card_links = None
account.id = info['id']
if is_rib_valid(info['id']):
account.iban = rib2iban(info['id'])
......@@ -446,6 +450,7 @@ def get_loan_list(self):
tds = tr.findall('td')
account = Account()
account._card_links = None
account.id = CleanText('./a')(tds[2]).split('-')[0].strip()
account.label = CleanText('./a')(tds[2]).split('-')[-1].strip()
account.type = Account.TYPE_LOAN
......@@ -557,7 +562,13 @@ def go_list(self):
self.submit_form(form, eventargument, eventtarget, scriptmanager)
def go_cards(self):
# Do not try to go the card summary if we have no card, it breaks the session
if not CleanText('//form[@id="main"]//a/span[text()="Mes cartes bancaires"]')(self.doc):
self.logger.info("Do not try to go the CardsPage, there is not link on the main page")
return
form = self.get_form(id='main')
eventargument = ""
if "MM$m_CH$IsMsgInit" in form:
......@@ -864,6 +875,7 @@ class item(ItemElement):
obj__parent_id = CleanText(TableCell('parent'))
obj_balance = 0
obj_currency = Currency(TableCell('coming'))
obj__card_links = None
def obj_coming(self):
if CleanText(TableCell('coming'))(self) == '-':
......@@ -892,10 +904,37 @@ class CardsComingPage(IndexPage):
def is_here(self):
return CleanText('//h2[text()="Encours de carte à débit différé"]')(self.doc)
def get_card_coming_info(self, number, info):
@method
class iter_cards(ListElement):
item_xpath = '//table[contains(@class, "compte") and position() = 1]//tr[contains(@id, "MM_HISTORIQUE_CB") and position() < last()]'
class item(ItemElement):
klass = Account
def obj_id(self):
# We must handle two kinds of Regexp because the 'X' are not
# located at the same level for sub-modules such as palatine
return Coalesce(
Regexp(CleanText(Field('label'), replace=[('*', 'X')]), r'(\d{6}\X{6}\d{4})', default=NotAvailable),
Regexp(CleanText(Field('label'), replace=[('*', 'X')]), r'(\d{4}\X{6}\d{6})', default=NotAvailable),
)(self)
def obj_number(self):
return Coalesce(
Regexp(CleanText(Field('label')), r'(\d{6}\*{6}\d{4})', default=NotAvailable),
Regexp(CleanText(Field('label')), r'(\d{4}\*{6}\d{6})', default=NotAvailable),
)(self)
obj_type = Account.TYPE_CARD
obj_label = CleanText('./td[1]')
obj_balance = Decimal(0)
obj_coming = CleanDecimal.French('./td[2]')
obj_currency = Currency('./td[2]')
obj__card_links = None
def get_card_coming_info(self, number, info):
# If the xpath match, that mean there are only one card
# We have enought information in `info` to get its coming transaction
# We have enough information in `info` to get its coming transaction
if CleanText('//tr[@id="MM_HISTORIQUE_CB_rptMois0_ctl01_trItem"]')(self.doc):
return info
......@@ -907,12 +946,16 @@ def get_card_coming_info(self, number, info):
if Regexp(CleanText(xpath), r'(\d{6}\*{6}\d{4})')(self.doc) == number:
return info
# For all card except the first one for the same check account, we have to get info through their href info
link = CleanText(Attr('//a[contains(text(),"%s")]' % number, 'href'))(self.doc)
infos = re.match(r'.*(DETAIL_OP_M0&[^\"]+).*', link)
info['link'] = infos.group(1)
# Some cards redirect to a checking account where we cannot found them. Since we have no details or history,
# we return None and skip them in the browser.
if CleanText('//a[contains(text(),"%s")]' % number)(self.doc):
# For all cards except the first one for the same check account, we have to get info through their href info
link = CleanText(Link('//a[contains(text(),"%s")]' % number))(self.doc)
infos = re.match(r'.*(DETAIL_OP_M0&[^\"]+).*', link)
info['link'] = infos.group(1)
return info
return info
return None
class CardsOldWebsitePage(IndexPage):
......@@ -945,6 +988,7 @@ class item(ItemElement):
obj_balance = 0
obj_coming = CleanDecimal.French(TableCell('coming'))
obj_currency = Currency(TableCell('coming'))
obj__card_links = None
def obj__parent_id(self):
return self.page.get_account()
......
......@@ -73,15 +73,18 @@ def do_login(self):
self.incapsula_ressource.go(params={'SWCGHOEL': 'v2'}, data=data)
self.login.go()
# this cookie contains an ugly \x01 and make next request fail with a 400 if not removed
___utmvafIuFLPmB = self.session.cookies.pop('___utmvafIuFLPmB', None)
if ___utmvafIuFLPmB:
self.session.cookies['___utmvafIuFLPmB'] = ___utmvafIuFLPmB.replace('\x01', '')
# this cookie contains an ugly \n and make next request fail with a 400 if not removed
___utmvbfIuFLPmB = self.session.cookies.pop('___utmvbfIuFLPmB', None)
if ___utmvbfIuFLPmB:
self.session.cookies['___utmvbfIuFLPmB'] = ___utmvbfIuFLPmB.replace('\n', '')
# remove 2 cookies that make next request fail with a 400 if not removed
# cookie name can change depend on ip, but seems to be constant on same ip
# example:
# 1st cookie 2nd cookie
# ___utmvafIuFLPmB, ___utmvbfIuFLPmB
# ___utmvaYauFLPmB, ___utmvbYauFLPmB
# it may have other names...
for cookie in self.session.cookies:
if '___utmva' in cookie.name or '___utmvb' in cookie.name:
# ___utmva... contains an ugly \x01
# ___utmvb... contains an ugly \n
self.session.cookies.pop(cookie.name)
if self.incapsula_ressource.is_here():
if self.page.is_javascript:
......
......@@ -78,9 +78,10 @@ def iter_investment(self, account):
def iter_history(self, account):
self.operations_list.stay_or_go(subsite=self.subsite, client_space=self.client_space)
for idx in self.page.get_operations_idx():
tr = self.operation.go(subsite=self.subsite, client_space=self.client_space, idx=idx).get_transaction()
if account.label == tr._account_label:
yield tr
self.operation.go(subsite=self.subsite, client_space=self.client_space, idx=idx)
for tr in self.page.get_transactions():
if account.label == tr._account_label:
yield tr
@need_login
def iter_pocket(self, account):
......