Commit c2504a84 authored by Romain Bignon's avatar Romain Bignon

Update of modules

parent 721de7d3
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Oleg Plakhotniuk
# Copyright(C) 2019 Vincent A
#
# This file is part of a weboob module.
#
......@@ -17,7 +17,10 @@
# 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 .module import WellsFargoModule
__all__ = ['WellsFargoModule']
from .module import AirparifModule
__all__ = ['AirparifModule']
# -*- coding: utf-8 -*-
# Copyright(C) 2014 Oleg Plakhotniuk
# Copyright(C) 2019 Vincent A
#
# This file is part of a weboob module.
#
......@@ -17,18 +17,23 @@
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from weboob.tools.test import BackendTest
from itertools import chain
from __future__ import unicode_literals
class WellsFargoTest(BackendTest):
MODULE = 'wellsfargo'
from weboob.browser import PagesBrowser, URL
def test_history(self):
"""
Test that there's at least one transaction in the whole history.
"""
b = self.backend
ts = chain(*[b.iter_history(a) for a in b.iter_accounts()])
t = next(ts, None)
self.assertNotEqual(t, None)
from .pages import AllPage
class AirparifBrowser(PagesBrowser):
BASEURL = 'https://airparif.asso.fr'
all_page = URL(r'/stations/indicepolluant/', AllPage)
def iter_gauges(self):
self.all_page.go(method='POST', headers={
# don't remove the following headers, site returns 404 else...
'X-Requested-With': 'XMLHttpRequest',
'Referer': 'https://airparif.asso.fr/stations/index/',
})
return self.page.iter_gauges()
# -*- coding: utf-8 -*-
# Copyright(C) 2019 Vincent A
#
# This file is part of a weboob module.
#
# This weboob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This weboob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.tools.backend import Module
from weboob.capabilities.base import find_object
from weboob.capabilities.gauge import (
CapGauge, SensorNotFound, Gauge, GaugeSensor,
)
from .browser import AirparifBrowser
__all__ = ['AirparifModule']
class AirparifModule(Module, CapGauge):
NAME = 'airparif'
DESCRIPTION = 'airparif website'
MAINTAINER = 'Vincent A'
EMAIL = 'dev@indigo.re'
LICENSE = 'LGPLv3+'
VERSION = '1.6'
BROWSER = AirparifBrowser
def iter_gauges(self, pattern=None):
if pattern:
pattern = pattern.lower()
for gauge in self.browser.iter_gauges():
if not pattern or pattern in gauge._searching:
yield gauge
def _get_gauge_by_id(self, id):
return find_object(self.browser.iter_gauges(), id=id)
def iter_sensors(self, gauge, pattern=None):
if pattern:
pattern = pattern.lower()
if not isinstance(gauge, Gauge):
gauge = self._get_gauge_by_id(gauge)
if gauge is None:
raise SensorNotFound()
if pattern is None:
for sensor in gauge.sensors:
yield sensor
else:
for sensor in gauge.sensors:
if pattern in sensor.name.lower():
yield sensor
def _get_sensor_by_id(self, id):
gid = id.partition('.')[0]
return find_object(self.iter_sensors(gid), id=id)
def get_last_measure(self, sensor):
if not isinstance(sensor, GaugeSensor):
sensor = self._get_sensor_by_id(sensor)
if sensor is None:
raise SensorNotFound()
return sensor.lastvalue
# -*- coding: utf-8 -*-
# Copyright(C) 2019 Vincent A
#
# This file is part of a weboob module.
#
# This weboob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This weboob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.browser.pages import JsonPage
from weboob.browser.elements import ItemElement, DictElement, method
from weboob.browser.filters.standard import (
Env, Format, Regexp, DateTime, CleanDecimal, Lower, Map,
)
from weboob.browser.filters.json import Dict
from weboob.capabilities.address import GeoCoordinates, PostalAddress
from weboob.capabilities.gauge import Gauge, GaugeSensor, GaugeMeasure
SENSOR_NAMES = {
'PM25': 'PM 2.5',
'PM10': 'PM 10',
'O3': 'O₃',
'NO3': 'NO₃',
'NO2': 'NO₂',
}
class AllPage(JsonPage):
@method
class iter_gauges(DictElement):
def find_elements(self):
return self.el.values()
class item(ItemElement):
klass = Gauge
def condition(self):
# sometimes the "date" field (which contains the hour) is empty
# and no measure is present in it, so we discard it
return bool(self.el['date'])
def parse(self, el):
for k in el:
self.env[k] = el[k]
self.env['city'] = Regexp(Dict('commune'), r'^(\D+)')(self)
obj_id = Dict('nom_court_sit')
obj_name = Dict('isit_long')
obj_city = Env('city')
obj_object = 'Pollution'
obj__searching = Lower(Format(
'%s %s %s %s',
Dict('isit_long'),
Dict('commune'),
Dict('ninsee'),
Dict('adresse'),
))
class obj_sensors(DictElement):
def find_elements(self):
return [dict(zip(('key', 'value'), tup)) for tup in self.el['indices'].items()]
class item(ItemElement):
klass = GaugeSensor
obj_name = Map(Dict('key'), SENSOR_NAMES)
obj_gaugeid = Env('nom_court_sit')
obj_id = Format('%s.%s', obj_gaugeid, Dict('key'))
obj_unit = 'µg/m³'
class obj_lastvalue(ItemElement):
klass = GaugeMeasure
obj_date = DateTime(
Format(
'%s %s',
Env('min_donnees'),
Env('date'), # "date" contains the time...
)
)
obj_level = CleanDecimal(Dict('value'))
class obj_geo(ItemElement):
klass = GeoCoordinates
obj_latitude = CleanDecimal(Env('latitude'))
obj_longitude = CleanDecimal(Env('longitude'))
class obj_location(ItemElement):
klass = PostalAddress
obj_street = Env('adresse')
obj_postal_code = Env('ninsee')
obj_city = Env('city')
obj_region = 'Ile-de-France'
obj_country = 'France'
# -*- coding: utf-8 -*-
# Copyright(C) 2019 Vincent A
#
# This file is part of a weboob module.
#
# This weboob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This weboob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this weboob module. If not, see <http://www.gnu.org/licenses/>.
from __future__ import unicode_literals
from weboob.tools.test import BackendTest
class AirparifTest(BackendTest):
MODULE = 'airparif'
def test_gauges(self):
all_gauges = list(self.backend.iter_gauges())
paris_gauges = list(self.backend.iter_gauges(pattern='paris'))
self.assertTrue(all_gauges)
self.assertTrue(paris_gauges)
self._check_gauge(all_gauges[0])
def _check_gauge(self, g):
self.assertTrue(g.id)
self.assertTrue(g.name)
self.assertTrue(g.city)
self.assertTrue(g.object)
self.assertTrue(g.sensors)
self._check_sensor(g.sensors[0], g)
def _check_sensor(self, s, g):
self.assertTrue(s.id)
self.assertTrue(s.name)
self.assertTrue(s.unit)
self.assertTrue(s.gaugeid == g.id)
self.assertTrue(s.lastvalue.date)
self.assertTrue(s.lastvalue.level)
self.assertTrue(s.geo.latitude)
self.assertTrue(s.geo.longitude)
self.assertTrue(s.location.street)
self.assertTrue(s.location.city)
self.assertTrue(s.location.region)
self.assertTrue(s.location.country)
self.assertTrue(s.location.postal_code)
......@@ -566,7 +566,7 @@ class AllocineBrowser(APIBrowser):
]
if query.summary:
movie = self.iter_movies(query.summary).next()
movie = next(self.iter_movies(query.summary))
params.append(('movie', movie.id))
result = self.__do_request('showtimelist', params)
......
......@@ -31,13 +31,6 @@ import re
import json
try:
cmp = cmp
except NameError:
def cmp(x, y):
return (x > y) - (x < y)
class SomePage(HTMLPage):
@property
def logged(self):
......@@ -86,7 +79,7 @@ class ActivityPage(SomePage):
records = json.loads(self.doc.xpath(
'//div[@id="completedActivityRecords"]//input[1]/@value')[0])
recent = [x for x in records if x['PDF_LOC'] is None]
for rec in sorted(recent, ActivityPage.cmp_records, reverse=True):
for rec in sorted(recent, key=lambda rec: ActivityPage.parse_date(rec['TRANS_DATE']), reverse=True):
desc = u' '.join(rec['TRANS_DESC'].split())
trans = Transaction((rec['REF_NUM'] or u'').strip())
trans.date = ActivityPage.parse_date(rec['TRANS_DATE'])
......@@ -97,11 +90,6 @@ class ActivityPage(SomePage):
trans.amount = -AmTr.decimal_amount(rec['TRANS_AMOUNT'])
yield trans
@staticmethod
def cmp_records(rec1, rec2):
return cmp(ActivityPage.parse_date(rec1['TRANS_DATE']),
ActivityPage.parse_date(rec2['TRANS_DATE']))
@staticmethod
def parse_date(recdate):
return datetime.strptime(recdate, u'%B %d, %Y')
......@@ -141,10 +129,13 @@ class StatementPage(RawPage):
self._tok = ReTokenizer(self._pdf, '\n', self.LEX)
def iter_transactions(self):
return sorted(self.read_transactions(),
cmp=lambda t1, t2: cmp(t2.date, t1.date) or
cmp(t1.label, t2.label) or
cmp(t1.amount, t2.amount))
trs = self.read_transactions()
# since the sorts are not in the same direction, we can't do in one pass
# python sorting is stable, so sorting in 2 passes can achieve a multisort
# the official docs give this way
trs = sorted(trs, key=lambda tr: (tr.label, tr.amount))
trs = sorted(trs, key=lambda tr: tr.date, reverse=True)
return trs
def read_transactions(self):
# Statement typically cover one month.
......
......@@ -21,7 +21,7 @@ from __future__ import unicode_literals
import datetime
from weboob.exceptions import BrowserIncorrectPassword
from weboob.exceptions import BrowserIncorrectPassword, ActionNeeded
from weboob.browser.browsers import LoginBrowser, need_login
from weboob.browser.exceptions import HTTPNotFound, ServerError
from weboob.browser.url import URL
......@@ -29,8 +29,8 @@ from dateutil.parser import parse as parse_date
from .pages import (
AccountsPage, JsonBalances, JsonPeriods, JsonHistory,
JsonBalances2, CurrencyPage, LoginPage, WrongLoginPage, AccountSuspendedPage,
NoCardPage, NotFoundPage,
JsonBalances2, CurrencyPage, LoginPage, NoCardPage,
NotFoundPage,
)
......@@ -40,11 +40,9 @@ __all__ = ['AmericanExpressBrowser']
class AmericanExpressBrowser(LoginBrowser):
BASEURL = 'https://global.americanexpress.com'
login = URL('/myca/logon/.*', LoginPage)
wrong_login = URL('/myca/fuidfyp/emea/.*', WrongLoginPage)
account_suspended = URL('/myca/onlinepayments/', AccountSuspendedPage)
login = URL(r'/myca/logon/emea/action/login', LoginPage)
accounts = URL(r'/accounts', AccountsPage)
accounts = URL(r'/api/servicing/v1/member', AccountsPage)
js_balances = URL(r'/account-data/v1/financials/balances', JsonBalances)
js_balances2 = URL(r'/api/servicing/v1/financials/transaction_summary\?type=split_by_cardmember&statement_end_date=(?P<date>[\d-]+)', JsonBalances2)
js_pending = URL(r'/account-data/v1/financials/transactions\?limit=1000&offset=(?P<offset>\d+)&status=pending',
......@@ -69,51 +67,49 @@ class AmericanExpressBrowser(LoginBrowser):
self.cache = {}
def do_login(self):
if not self.login.is_here():
self.location('/myca/logon/emea/action?request_type=LogonHandler&DestPage=https%3A%2F%2Fglobal.americanexpress.com%2Fmyca%2Fintl%2Facctsumm%2Femea%2FaccountSummary.do%3Frequest_type%3D%26Face%3Dfr_FR%26intlink%3Dtopnavvotrecompteneligne-HPmyca&Face=fr_FR&Info=CUExpired')
self.page.login(self.username, self.password)
if self.wrong_login.is_here() or self.login.is_here() or self.account_suspended.is_here():
self.login.go(data={
'request_type': 'login',
'UserID': self.username,
'Password': self.password,
'Logon': 'Logon',
})
if self.page.get_status_code() != 0:
if self.page.get_error_code() == 'LGON004':
# This error happens when the website needs that the user
# enter his card information and reset his password.
# There is no message returned when this error happens.
raise ActionNeeded()
raise BrowserIncorrectPassword()
@need_login
def get_accounts(self):
self.accounts.go()
accounts = list(self.page.iter_accounts())
def iter_accounts(self):
loc = self.session.cookies.get_dict(domain=".americanexpress.com")['axplocale'].lower()
self.currency_page.go(locale=loc)
currency = self.page.get_currency()
for account in accounts:
self.accounts.go()
account_list = list(self.page.iter_accounts(currency=currency))
for account in account_list:
try:
# for the main account
self.js_balances.go(headers={'account_tokens': account._balances_token})
self.js_balances.go(headers={'account_tokens': account.id})
except HTTPNotFound:
# for secondary accounts
self.js_periods.go(headers={'account_token': account._balances_token})
self.js_periods.go(headers={'account_token': account._history_token})
periods = self.page.get_periods()
self.js_balances2.go(date=periods[1], headers={'account_tokens': account._balances_token})
self.page.set_balances(accounts)
# get currency
loc = self.session.cookies.get_dict(domain=".americanexpress.com")['axplocale'].lower()
self.currency_page.go(locale=loc)
currency = self.page.get_currency()
for acc in accounts:
acc.currency = currency
yield acc
@need_login
def get_accounts_list(self):
for account in self.get_accounts():
self.js_balances2.go(date=periods[1], headers={'account_tokens': account.id})
self.page.fill_balances(obj=account)
yield account
@need_login
def iter_history(self, account):
self.js_periods.go(headers={'account_token': account._token})
self.js_periods.go(headers={'account_token': account._history_token})
periods = self.page.get_periods()
today = datetime.date.today()
# TODO handle pagination
for p in periods:
self.js_posted.go(offset=0, end=p, headers={'account_token': account._token})
self.js_posted.go(offset=0, end=p, headers={'account_token': account._history_token})
for tr in self.page.iter_history(periods=periods):
# As the website is very handy, passing account_token is not enough:
# it will return every transactions of each account, so we
......@@ -128,14 +124,14 @@ class AmericanExpressBrowser(LoginBrowser):
# ('Enregistrées' tab on the website)
# "pending" have no vdate and debit date is in future
self.js_periods.go(headers={'account_token': account._token})
self.js_periods.go(headers={'account_token': account._history_token})
periods = self.page.get_periods()
date = parse_date(periods[0]).date()
today = datetime.date.today()
# when the latest period ends today we can't know the coming debit date
if date != today:
try:
self.js_pending.go(offset=0, headers={'account_token': account._token})
self.js_pending.go(offset=0, headers={'account_token': account._history_token})
except ServerError as exc:
# At certain times of the month a connection might not have pendings;
# in that case, `js_pending.go` would throw a 502 error Bad Gateway
......@@ -150,7 +146,7 @@ class AmericanExpressBrowser(LoginBrowser):
# "posted" have a vdate but debit date can be future or past
for p in periods:
self.js_posted.go(offset=0, end=p, headers={'account_token': account._token})
self.js_posted.go(offset=0, end=p, headers={'account_token': account._history_token})
for tr in self.page.iter_history(periods=periods):
if tr.date > today or not tr.date:
if tr._owner == account._idforJSON:
......
......@@ -44,7 +44,7 @@ class AmericanExpressModule(Module, CapBank):
self.config['password'].get())
def iter_accounts(self):
return self.browser.get_accounts_list()
return self.browser.iter_accounts()
def iter_history(self, account):
return self.browser.iter_history(account)
......
......@@ -19,18 +19,14 @@
from __future__ import unicode_literals
from ast import literal_eval
from decimal import Decimal
import re
from weboob.browser.pages import LoggedPage, JsonPage, HTMLPage
from weboob.browser.elements import ItemElement, DictElement, method
from weboob.browser.filters.standard import Date, Eval, Env, CleanText, Field, CleanDecimal
from weboob.browser.filters.standard import Date, Eval, Env, CleanText, Field, CleanDecimal, Format, Currency
from weboob.browser.filters.json import Dict
from weboob.capabilities.bank import Account, Transaction
from weboob.capabilities.base import NotAvailable
from weboob.tools.json import json
from weboob.tools.compat import basestring
from weboob.exceptions import ActionNeeded, BrowserUnavailable
from dateutil.parser import parse as parse_date
......@@ -48,14 +44,6 @@ def parse_decimal(s):
return CleanDecimal(replace_dots=comma).filter(s)
class WrongLoginPage(HTMLPage):
pass
class AccountSuspendedPage(HTMLPage):
pass
class NoCardPage(HTMLPage):
def on_load(self):
raise ActionNeeded()
......@@ -68,77 +56,70 @@ class NotFoundPage(HTMLPage):
raise BrowserUnavailable(alert_header, alert_content)
class LoginPage(HTMLPage):
def login(self, username, password):
form = self.get_form(name='ssoform')
form['UserID'] = username
form['USERID'] = username
form['Password'] = password
form['PWD'] = password
form.submit()
class AccountsPage(LoggedPage, HTMLPage):
def iter_accounts(self):
for line in self.doc.xpath('//script[@id="initial-state"]')[0].text.split('\n'):
m = re.search('window.__INITIAL_STATE__ = (.*);', line)
if m:
data = json.loads(literal_eval(m.group(1)))
break
else:
assert False, "data was not found"
assert data[15] == 'core'
assert len(data[16]) == 3
# search for products to get products list
for index, el in enumerate(data[16][2]):
if 'products' in el:
accounts_data = data[16][2][index + 1]
assert len(accounts_data) == 2
assert accounts_data[1][4] == 'productsList'
accounts_data = accounts_data[1][5]
token = []
for account_data in accounts_data:
if isinstance(account_data, basestring):
balances_token = account_data
elif isinstance(account_data, list) and not account_data[4][2][0] == "Canceled":
acc = Account()
if len(account_data) > 15:
token.append(account_data[-11])
acc._idforJSON = account_data[10][-1]
else:
acc._idforJSON = account_data[-5][-1]
acc._idforJSON = re.sub(r'\s+', ' ', acc._idforJSON)
acc.number = '-%s' % account_data[2][2]
acc.label = '%s %s' % (account_data[6][4], account_data[10][-1])
acc._balances_token = acc.id = balances_token
acc._token = token[-1]
acc.type = Account.TYPE_CARD
yield acc
class LoginPage(JsonPage):
def get_status_code(self):
# - 0 = OK
# - 1 = Incorrect login/password
return CleanDecimal(Dict('statusCode'))(self.doc)
def get_error_code(self):
# - LGON004 = ActionNeeded
return CleanText(Dict('errorCode'))(self.doc)
class AccountsPage(LoggedPage, JsonPage):
@method
class iter_accounts(DictElement):
def find_elements(self):
for obj in self.page.doc.get('accounts', []):
obj['_history_token'] = obj['account_token']
yield obj
for secondary_acc in obj.get('supplementary_accounts', []):
# Secondary accounts use the id of the parrent account
# when searching history/coming. History/coming are filtered
# on the owner name (_idforJSON).
secondary_acc['_history_token'] = obj['account_token']
yield secondary_acc
class item(ItemElement):
klass = Account
def condition(self):
return any(status == 'Active' for status in Dict('status/account_status')(self))
obj_id = Dict('account_token')
obj__history_token = Dict('_history_token')
obj__account_type = Dict('account/relationship')
obj_number = Format('-%s', Dict('account/display_account_number'))
obj_type = Account.TYPE_CARD
obj_currency = Currency(Env('currency'))
obj__idforJSON = Dict('profile/embossed_name')
def obj_label(self):
if Dict('account/relationship')(self) == 'SUPP':
return Format(
'%s %s',
Dict('platform/amex_region'),
Dict('profile/embossed_name'),
)(self)
return CleanText(Dict('product/description'))(self)
class JsonBalances(LoggedPage, JsonPage):
def set_balances(self, accounts):
by_token = {a._balances_token: a for a in accounts}
for d in self.doc:
# coming is what should be refunded at a futur deadline
by_token[d['account_token']].coming = -float_to_decimal(d['total_debits_balance_amount'