remove unused code, fix timeout of history fetch

This commit is contained in:
Romain Bignon 2015-03-05 20:16:04 +01:00 committed by Romain Bignon
commit 3df00b6395
4 changed files with 141 additions and 547 deletions

View file

@ -18,10 +18,12 @@
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from weboob.deprecated.browser import Browser, BrowserIncorrectPassword
from .pages import LoginPage, AccountPage, DownloadHistoryPage, LastDownloadHistoryPage, SubmitPage, HistoryParser, UselessPage, HistoryPage, CSVAlreadyAsked, HistoryDetailsPage
from .newpages import NewHomePage, NewAccountPage, NewProHistoryPage, NewPartHistoryPage
import datetime
from dateutil.relativedelta import relativedelta
from weboob.deprecated.browser import Browser, BrowserIncorrectPassword
from .pages import LoginPage, AccountPage, UselessPage, HomePage, ProHistoryPage, PartHistoryPage, HistoryDetailsPage
__all__ = ['Paypal']
@ -36,33 +38,28 @@ class Paypal(Browser):
'/cgi-bin/webscr\?cmd=_login-run$': LoginPage,
'/cgi-bin/webscr\?cmd=_login-submit.+$': LoginPage, # wrong login
'/cgi-bin/webscr\?cmd=_login-processing.+$': UselessPage,
'/cgi-bin/webscr\?cmd=_account&nav=0.0$': AccountPage,
'/cgi-bin/webscr\?cmd=_login-done.+$': AccountPage,
'/cgi-bin/webscr\?cmd=_home&country_lang.x=true$': NewHomePage,
'/cgi-bin/webscr\?cmd=_history-download&nav=0.3.1$': DownloadHistoryPage,
'/cgi-bin/webscr\?cmd=_history&nav=0.3.0$': HistoryPage,
'/cgi-bin/webscr\?cmd=_history&dispatch=[a-z0-9]+$': HistoryPage,
'/cgi-bin/webscr\?cmd=_history-download-recent$': LastDownloadHistoryPage,
'/cgi-bin/webscr\?dispatch=[a-z0-9]+$': (SubmitPage, HistoryParser()),
'/cgi-bin/webscr\?cmd=_history-download-recent-submit&dispatch=[a-z0-9]+$': (SubmitPage, HistoryParser()),
'https://history.paypal.com/cgi-bin/webscr\?cmd=_history-details-from-hub&id=[A-Z0-9]+$': HistoryDetailsPage,
'https://www.paypal.com/webapps/business/\?nav=0.0': NewHomePage,
'https://www.paypal.com/webapps/business/\?country_lang.x=true': NewHomePage,
'https://www.paypal.com/myaccount/\?nav=0.0': NewHomePage,
'https://www.paypal.com/businessexp/money': NewAccountPage,
'https://www.paypal.com/webapps/business/activity\?.*': NewProHistoryPage,
'https://www.paypal.com/myaccount/activity/.*': (NewPartHistoryPage, 'json'),
'https://www.paypal.com/myaccount/': NewProHistoryPage,
'/cgi-bin/webscr\?cmd=_account.*$': UselessPage,
'/cgi-bin/webscr\?cmd=_login-done.+$': UselessPage,
'/cgi-bin/webscr\?cmd=_home&country_lang.x=true$': HomePage,
'https://\w+.paypal.com/cgi-bin/webscr\?cmd=_history-details-from-hub&id=[A-Z0-9]+$': HistoryDetailsPage,
'https://\w+.paypal.com/webapps/business/\?nav=0.0': HomePage,
'https://\w+.paypal.com/webapps/business/\?country_lang.x=true': HomePage,
'https://\w+.paypal.com/myaccount/\?nav=0.0': HomePage,
'https://\w+.paypal.com/businessexp/money': AccountPage,
'https://\w+.paypal.com/webapps/business/activity\?.*': ProHistoryPage,
'https://\w+.paypal.com/myaccount/activity/.*': (PartHistoryPage, 'json'),
'https://\w+.paypal.com/myaccount/': ProHistoryPage,
}
DEFAULT_TIMEOUT = 30 # CSV export is slow
DEFAULT_TIMEOUT = 60
BEGINNING = datetime.date(1998, 6, 1) # The day PayPal was founded
website = None
account_type = None
def find_website_version(self):
self.website = "new"
if self.is_on_page(NewHomePage):
def find_account_type(self):
if self.is_on_page(HomePage):
# XXX Unable to get more than 2 years of history on pro accounts.
self.BEGINNING = datetime.date.today() - relativedelta(months=24)
self.account_type = "pro"
return
self.location(self._response.info().getheader('refresh').split("bin/")[1])
@ -71,7 +68,7 @@ class Paypal(Browser):
self.account_type = "perso"
else:
self.location('/webapps/business/?nav=0.0')
if self.is_on_page(NewHomePage):
if self.is_on_page(HomePage):
self.account_type = "pro"
else:
self.account_type = "perso"
@ -95,73 +92,37 @@ class Paypal(Browser):
if self.is_on_page(LoginPage):
raise BrowserIncorrectPassword()
self.find_website_version()
self.find_account_type()
def get_accounts(self):
if self.website == "old":
if not self.is_on_page(AccountPage):
self.location('/en/cgi-bin/webscr?cmd=_account&nav=0.0')
elif not self.is_on_page(NewAccountPage):
if not self.is_on_page(AccountPage):
self.location('/businessexp/money')
return self.page.get_accounts()
def get_account(self, _id):
if self.website == "old":
if not self.is_on_page(AccountPage):
self.location('/en/cgi-bin/webscr?cmd=_account&nav=0.0')
elif not self.is_on_page(NewAccountPage):
if not self.is_on_page(AccountPage):
self.location('/businessexp/money')
return self.page.get_account(_id)
def get_history(self, account, step_min=90, step_max=365*10):
def fetch_fn(start, end):
def transactions():
parse = True
while parse:
for trans in self.page.iter_transactions(account):
yield trans
parse = self.page.next()
self.history(start=start, end=end)
if next(self.page.parse(), False):
return transactions()
return self.smart_fetch(beginning=self.BEGINNING,
end=datetime.date.today(),
step_min=step_min,
step_max=step_max,
fetch_fn=fetch_fn)
def history(self, start, end):
self.location('/en/cgi-bin/webscr?cmd=_history&nav=0.3.0')
self.page.filter(start, end)
assert self.is_on_page(HistoryPage)
def get_download_history(self, account, step_min=None, step_max=None):
if step_min is None and step_max is None:
if self.website == "old":
step_min = 90
step_max = 365*2
else:
step_min = 90
step_max = 180
step_min = 30
step_max = 180
def fetch_fn(start, end):
if self.website == "old" and self.download_history(start, end).rows:
return self.page.iter_transactions(account)
elif self.download_history(start, end):
if self.download_history(start, end):
return self.page.iter_transactions(account)
return iter([])
assert step_max <= 365*2 # PayPal limitations as of 2014-06-16
try:
for i in self.smart_fetch(beginning=self.BEGINNING,
end=datetime.date.today(),
step_min=step_min,
step_max=step_max,
fetch_fn=fetch_fn):
yield i
except CSVAlreadyAsked:
for i in self.download_last_history(account):
yield i
for i in self.smart_fetch(beginning=self.BEGINNING,
end=datetime.date.today(),
step_min=step_min,
step_max=step_max,
fetch_fn=fetch_fn):
yield i
def smart_fetch(self, beginning, end, step_min, step_max, fetch_fn):
"""
@ -172,57 +133,45 @@ class Paypal(Browser):
step = step_min
while end > beginning:
start = end - datetime.timedelta(step)
chunk = fetch_fn(start, end)
chunk = list(fetch_fn(start, end))
end = start - datetime.timedelta(1)
if chunk:
# If there're transactions in current period,
# decrease the period.
if len(chunk) > 50:
# If there're too much transactions in current period, decrease
# the period.
step = max(step_min, step/FACTOR)
for trans in chunk:
yield trans
else:
# If there's no transactions in current period,
# If there's no transactions, or only a bit, in current period,
# increase the period.
step = min(step_max, step*FACTOR)
for trans in chunk:
yield trans
def download_history(self, start, end):
"""
Download CSV history.
Download history.
However, it is not normalized, and sometimes the download is refused
and sent later by mail.
"""
if self.website == "old":
self.location('/en/cgi-bin/webscr?cmd=_history-download&nav=0.3.1')
assert self.is_on_page(DownloadHistoryPage)
self.page.download(start, end)
assert self.is_on_page(SubmitPage)
return self.page.document
s = start.strftime('%d/%m/%Y')
e = end.strftime('%d/%m/%Y')
# Settings a big magic number so we hope to get all transactions for the period
LIMIT = '9999'
if self.account_type == "pro":
self.location('https://www.paypal.com/webapps/business/activity?fromdate=' + s + '&todate=' + e + '&transactiontype=ALL_TRANSACTIONS&currency=ALL_TRANSACTIONS_CURRENCY&limit=' + LIMIT)
else:
s = start.strftime('%d/%m/%Y')
e = end.strftime('%d/%m/%Y')
# Settings a big magic number so we get all transaction for the period
LIMIT = '9999'
if self.account_type == "pro":
self.location('/webapps/business/activity?fromdate=' + s + '&todate=' + e + '&transactiontype=ALL_TRANSACTIONS&currency=ALL_TRANSACTIONS_CURRENCY&limit=' + LIMIT)
else:
self.location('/myaccount/activity/filter?typeFilter=all&isNewSearch=true&startDate=' + s + '&endDate=' + e + '&limit=' + LIMIT)
return self.page.transaction_left()
def download_last_history(self, account):
self.location('/en/cgi-bin/webscr?cmd=_history-download-recent')
self.page.download()
if self.page.document.rows:
return self.page.iter_transactions(account)
self.location('https://www.paypal.com/myaccount/activity/filter?typeFilter=all&isNewSearch=true&startDate=' + s + '&endDate=' + e + '&limit=' + LIMIT)
return self.page.transaction_left()
def transfer(self, from_id, to_id, amount, reason=None):
raise NotImplementedError()
def convert_amount(self, account, trans):
if(trans['actions']['details']['action'] == 'ACTIVITY_DETAILS'):
if trans['actions']['details']['action'] == 'ACTIVITY_DETAILS':
self.location(trans['actions']['details']['url'])
if self.is_on_page(HistoryDetailsPage):
cc = self.page.get_converted_amount(account)
if cc:
trans['originalAmount'] = trans['netAmount']
trans['netAmount'] = cc
return trans

View file

@ -47,14 +47,12 @@ class PaypalModule(Module, CapBank):
return self.browser.get_accounts().itervalues()
def get_account(self, _id):
with self.browser:
account = self.browser.get_account(_id)
account = self.browser.get_account(_id)
if account:
return account
else:
raise AccountNotFound()
def iter_history(self, account):
with self.browser:
for history in self.browser.get_download_history(account):
yield history
for history in self.browser.get_download_history(account):
yield history

View file

@ -1,133 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright(C) 2014-2015 Budget Insight
#
# 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 <http://www.gnu.org/licenses/>.
import re
from decimal import Decimal
from weboob.deprecated.browser import Page
from weboob.capabilities.bank import Account
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.tools.date import parse_french_date
class NewHomePage(Page):
pass
class NewAccountPage(Page):
def get_account(self, _id):
return self.get_accounts().get(_id)
def get_accounts(self):
accounts = {}
content = self.document.xpath('//div[@id="moneyPage"]')[0]
# Primary currency account
primary_account = Account()
primary_account.type = Account.TYPE_CHECKING
balance = self.parser.tocleanstring(content.xpath('//div[contains(@class, "col-md-6")][contains(@class, "available")]')[0])
primary_account.currency = Account.get_currency(balance)
primary_account.id = unicode(primary_account.currency)
primary_account.balance = Decimal(FrenchTransaction.clean_amount(balance))
primary_account.label = u'%s %s*' % (self.browser.username, primary_account.currency)
accounts[primary_account.id] = primary_account
return accounts
class NewProHistoryPage(Page):
def iter_transactions(self, account):
for trans in self.parse():
if trans._currency == account.currency:
yield trans
def parse(self):
for i, tr in enumerate(self.document.xpath('//tr')):
t = FrenchTransaction(tr.xpath('./td[@class="transactionId"]/span')[0].text.strip())
date = parse_french_date(tr.xpath('./td[@class="date"]')[0].text.strip())
status = tr.xpath('./td[@class="desc"]/ul/li[@class="first"]')[0].text.strip()
#We pass this because it's not transaction
if status == u'Créé' or status == u'Annulé' or status == u'Suspendu':
continue
raw = tr.xpath('./td[@class="desc"]/strong')[0].text.strip()
t.parse(date=date, raw=raw)
amount = tr.xpath('./td[@class="price"]/span')[0].text.strip()
t.set_amount(amount)
t._currency = Account.get_currency(amount)
yield t
def transaction_left(self):
return (len(self.document.xpath('//div[@class="no-records"]')) == 0)
class NewPartHistoryPage(Page):
def transaction_left(self):
return (len(self.document['data']['activity']['COMPLETED']) > 0 or len(self.document['data']['activity']['PENDING']) > 0)
def iter_transactions(self, account):
for trans in self.parse(account):
yield trans
def parse(self, account):
transactions = list()
for status in ['PENDING', 'COMPLETED']:
transac = self.document['data']['activity'][status]
for t in transac:
tran = self.parse_transaction(t, account)
if tran:
transactions.append(tran)
transactions.sort(key=lambda tr: tr.rdate, reverse=True)
for t in transactions:
yield t
def parse_transaction(self, transaction, account):
t = FrenchTransaction(transaction['activityId'])
date = parse_french_date(transaction['date'])
raw = transaction.get('counterparty', transaction['displayType'])
t.parse(date=date, raw=raw)
if transaction['currencyCode'] != account.currency:
transaction = self.browser.convert_amount(account, transaction)
try:
t.original_amount = self.format_amount(transaction['originalAmount'], transaction["isCredit"])
t.original_currency = transaction["currencyCode"]
except KeyError:
return
try:
t.amount = self.format_amount(transaction['netAmount'], transaction["isCredit"])
except KeyError:
return
t._currency = transaction['currencyCode']
return t
def format_amount(self, to_format, is_credit):
m = re.search(r"\D", to_format[::-1])
amount = Decimal(re.sub(r'[^\d]', '', to_format))/Decimal((10 ** m.start()))
if is_credit:
return abs(amount)
else:
return -abs(amount)

View file

@ -17,24 +17,14 @@
# You should have received a copy of the GNU Affero General Public License
# along with weboob. If not, see <http://www.gnu.org/licenses/>.
from decimal import InvalidOperation
from decimal import Decimal
import re
import datetime
import dateutil.parser
from weboob.deprecated.browser import Page, BrokenPageError
from weboob.deprecated.browser.parsers.csvparser import CsvParser
from weboob.tools.misc import to_unicode
from weboob.tools.date import parse_french_date
from weboob.capabilities.bank import Account, Transaction
from weboob.tools.capabilities.bank.transactions import \
AmericanTransaction as AmTr
from weboob.capabilities.bank import Account
from weboob.capabilities.base import NotAvailable
class CSVAlreadyAsked(Exception):
pass
from weboob.deprecated.browser import Page
from weboob.tools.capabilities.bank.transactions import FrenchTransaction
from weboob.tools.date import parse_french_date
class LoginPage(Page):
@ -45,331 +35,121 @@ class LoginPage(Page):
self.browser.submit(nologin=True)
class UselessPage(Page):
pass
class HomePage(Page):
pass
class AccountPage(Page):
def get_account(self, _id):
return self.get_accounts().get(_id)
def get_accounts(self):
accounts = {}
content = self.document.xpath('//div[@id="main"]//div[@class="col first"]')[0]
content = self.document.xpath('//div[@id="moneyPage"]')[0]
# Primary currency account
primary_account = Account()
primary_account.type = Account.TYPE_CHECKING
# Total currency balance.
# If there are multiple currencies, this balance is all currencies
# converted to the main currency.
try:
balance = content.xpath('.//h3/span[@class="balance"]')
if not balance:
balance = content.xpath('.//li[@class="balance"]//span/strong')
balance = balance[0].text_content().strip()
primary_account.balance = AmTr.decimal_amount(balance)
primary_account.currency = Account.get_currency(balance)
primary_account.id = unicode(primary_account.currency)
primary_account.label = u'%s %s*' % (self.browser.username, balance.split()[-1])
balance = self.parser.tocleanstring(content.xpath('//div[contains(@class, "col-md-6")][contains(@class, "available")]')[0])
except IndexError:
primary_account.id = 'EUR'
primary_account.currency = 'EUR'
primary_account.balance = NotAvailable
primary_account.label = u'%s' % (self.browser.username)
else:
primary_account.currency = Account.get_currency(balance)
primary_account.id = unicode(primary_account.currency)
primary_account.balance = Decimal(FrenchTransaction.clean_amount(balance))
primary_account.label = u'%s %s*' % (self.browser.username, primary_account.currency)
accounts[primary_account.id] = primary_account
# The following code will only work if the user enabled multiple currencies.
balance = content.xpath('.//div[@class="body"]//ul/li[@class="balance"]/span')
table = content.xpath('.//table[@id="balanceDetails"]//tbody//tr')
# sanity check
if bool(balance) is not bool(table):
raise BrokenPageError('Unable to find all required multiple currency entries')
# Primary currency balance.
# If the user enabled multiple currencies, we get this one instead.
# An Account object has only one currency; secondary currencies should be other accounts.
if balance:
balance = balance[0].text_content().strip()
primary_account.balance = AmTr.decimal_amount(balance)
# The primary currency of the "head balance" is the same; ensure we got the right one
assert primary_account.currency == primary_account.get_currency(balance)
for row in table:
balance = row.xpath('.//td')[-1].text_content().strip()
account = Account()
account.type = Account.TYPE_CHECKING
# XXX it ignores 5+ devises, so it's bad, but it prevents a crash, cf #1216
try:
account.balance = AmTr.decimal_amount(balance)
except InvalidOperation:
continue
account.currency = Account.get_currency(balance)
account.id = unicode(account.currency)
account.label = u'%s %s' % (self.browser.username, balance.split()[-1])
if account.id == primary_account.id:
assert account.balance == primary_account.balance
assert account.currency == primary_account.currency
elif account.currency:
accounts[account.id] = account
return accounts
class DownloadHistoryPage(Page):
def download(self, start, end):
tr_last_file_request = self.document.xpath('//table//table//table[@width="100%"]//tr[2]//td')
if len(tr_last_file_request) > 1 and tr_last_file_request[1].text is not None:
last_file_request = tr_last_file_request[1].text[:-1]
try:
last_file_request = dateutil.parser.parse(last_file_request.encode('utf-8')).date()
except ValueError:
last_file_request = parse_french_date(last_file_request).date()
if last_file_request == datetime.date.today():
raise CSVAlreadyAsked('')
self.browser.select_form(name='form1')
self.browser['to_c'] = str(end.year)
self.browser['to_a'] = str(end.month)
self.browser['to_b'] = str(end.day)
self.browser['from_c'] = str(start.year)
self.browser['from_a'] = str(start.month)
self.browser['from_b'] = str(start.day)
self.browser['custom_file_type'] = ['comma_allactivity']
self.browser['latest_completed_file_type'] = ['']
self.browser.submit()
class LastDownloadHistoryPage(Page):
def download(self):
self.browser.select_form(nr=1)
log_select = self.document.xpath('//table//form//input[@type="radio"]')[0].attrib['value']
self.browser['log_select'] = [log_select]
self.browser.submit()
class SubmitPage(Page):
"""
Any result of form submission
"""
def iter_transactions(self, account):
csv = self.document
if len(csv.header) == 42 or len(csv.header) == 43:
# Merchant multi-currency account
# 42 is for when the user can't access the balance on the website
# 43 is for full acces to the account
DATE = 0
TIME = 1
NAME = 3
TYPE = 4
if csv.header[7] == "Devise":
CURRENCY = 7
GROSS = 8
FEE = 9
NET = 10
FROM = 11
TO = 12
TRANS_ID = 13
ITEM = 16
SITE = -1
else:
CURRENCY = 6
GROSS = 7
FEE = 8
NET = 9
FROM = 10
TO = 11
TRANS_ID = 12
ITEM = 15
SITE = 24
elif len(csv.header) == 11:
# Regular multi-currency account
DATE = 0
TIME = 1
NAME = 3
TYPE = 4
CURRENCY = 6
GROSS = -1
FEE = -1
NET = 7
FROM = -1
TO = -1
TRANS_ID = -1
ITEM = -1
SITE = -1
else:
raise ValueError('CSV fields count of %i is not supported' % len(csv.header))
for row in csv.rows:
# we filter transaction currceny to match account currency, except if we don't now the account currency
# we ignore canceled transactions
if (account.balance != NotAvailable and account.get_currency(row[CURRENCY]) != account.currency) or row[NET] == '...':
continue
# analog to dict.get()
get = lambda i, v=None: row[i] if 0 <= i < len(row) else v
trans = Transaction(get(TRANS_ID, u''))
# silly American locale
if re.search(r'\d\.\d\d$', row[NET]):
date = datetime.datetime.strptime(row[DATE] + ' ' + row[TIME], "%m/%d/%Y %H:%M:%S")
else:
date = datetime.datetime.strptime(row[DATE] + ' ' + row[TIME], "%d/%m/%Y %H:%M:%S")
trans.date = date
trans.rdate = date
line = row[NAME]
if get(ITEM):
line += u' ' + row[ITEM]
if get(SITE):
line += u"(" + row[SITE] + u")"
trans.raw = line
trans.label = row[NAME]
if row[TYPE].startswith(u'Update to eCheck') or \
row[TYPE].startswith(u'Order'):
continue
if row[TYPE].endswith(u'Credit Card') or row[TYPE].endswith(u'carte bancaire'):
trans.type = Transaction.TYPE_CARD
elif row[TYPE].endswith(u'Payment Sent') or row[TYPE].startswith(u'Paiement'):
trans.type = Transaction.TYPE_ORDER
elif row[TYPE] in (u'Currency Conversion', u'Conversion de devise'):
trans.type = Transaction.TYPE_BANK
else:
trans.type = Transaction.TYPE_UNKNOWN
# Net is what happens after the fee (0 for most users), so what is the most "real"
trans.amount = AmTr.decimal_amount(row[NET])
trans._gross = AmTr.decimal_amount(get(GROSS, row[NET]))
trans._fees = AmTr.decimal_amount(get(FEE, u'0.00'))
trans._to = get(TO)
trans._from = get(FROM)
yield trans
class HistoryParser(CsvParser):
HEADER = True
FMTPARAMS = {'skipinitialspace': True}
def decode_row(self, row, encoding):
"""
PayPal returns different encodings (latin-1 and utf-8 are know ones)
"""
return [to_unicode(cell) for cell in row]
class UselessPage(Page):
pass
class HistoryPage(Page):
def guess_format(self):
rp = re.compile('PAYPAL\.widget\.CalendarLocales\.MDY_([A-Z]+)_POSITION\s*=\s*(\d)')
rd = re.compile('PAYPAL\.widget\.CalendarLocales\.DATE_DELIMITER\s*=\s*"(.)"')
rm = re.compile('PAYPAL\.widget\.CalendarLocales\.MONTH_NAMES\s*=\s*\[(.+)\]')
translate = {'DAY': '%d', 'MONTH': '%m', 'YEAR': '%Y'}
pos = {}
delim = '/'
months = {}
for script in self.document.xpath('//script'):
for line in script.text_content().splitlines():
m = rp.match(line)
if m and m.groups():
pos[int(m.groups()[1])] = translate[m.groups()[0]]
else:
m = rd.match(line)
if m:
delim = m.groups()[0]
else:
m = rm.match(line)
if m:
months = [month.strip("'").strip().lower()[0:3]
for month
in m.groups()[0].split(',')]
date_format = delim.join((pos[0], pos[1], pos[2]))
if date_format == "%m/%d/%Y":
time_format = "%I:%M:%S %p"
else:
time_format = "%H:%M:%S"
return date_format, time_format, months
def filter(self, start, end):
date_format = self.guess_format()[0]
self.browser.select_form(name='history')
self.browser['dateoption'] = ['dateselect']
self.browser['from_date'] = start.strftime(date_format)
self.browser['to_date'] = end.strftime(date_format)
self.browser.submit(name='show')
self.browser.select_form(name='history')
self.browser.submit(name='filter_2')
def next(self):
if self.document.xpath('//input[@name="next"]'):
self.browser.select_form(name='history')
self.browser.submit(name='next')
return True
def parse(self):
emonths = ['January', 'February', 'March', 'April',
'May', 'June', 'July', 'August',
'September', 'October', 'November', 'December']
date_format, time_format, months = self.guess_format()
for row in self.document.xpath('//table[@id="transactionTable"]/tbody/tr'):
if len(row.xpath('.//td')) < 5:
continue
amount = row.xpath('.//td[@headers="gross"]')[-1].text_content().strip()
if re.search('\d', amount):
currency = Account.get_currency(amount)
amount = AmTr.decimal_amount(amount)
else:
continue
idtext = row.xpath('.//td[@class="detailsNoPrint"]//span[@class="accessAid"]')[0] \
.text_content().replace(u'\xa0', u' ').strip().rpartition(' ')[-1]
trans = Transaction(idtext)
trans.amount = amount
trans._currency = currency
datetext = row.xpath('.//td[@class="dateInfo"]')[0].text_content().strip()
for i in range(0, 12):
datetext = datetext.replace(months[i], emonths[i])
date = dateutil.parser.parse(datetext)
trans.date = date
trans.rdate = date
trans.label = to_unicode(row.xpath('.//td[@class="emailInfo"]')[0].text_content().strip())
info = to_unicode(row.xpath('.//td[@class="paymentTypeInfo"]')[0].text_content().strip())
trans.raw = info + u' ' + trans.label
if u'Authorization' in info or u'Autorisation' in info or \
u'Order' in info:
continue
if u'Credit Card' in trans.label or u'Carte bancaire' in trans.label:
trans.type = Transaction.TYPE_CARD
elif info.startswith(u'Payment') or info.startswith(u'Paiement'):
trans.type = Transaction.TYPE_ORDER
elif u'Currency Conversion' in info or u'Conversion de devise' in info:
trans.type = Transaction.TYPE_BANK
else:
trans.type = Transaction.TYPE_UNKNOWN
yield trans
class ProHistoryPage(Page):
def iter_transactions(self, account):
for trans in self.parse():
if trans._currency == account.currency:
yield trans
def parse(self):
for tr in self.document.xpath('//tr'):
t = FrenchTransaction(tr.xpath('./td[@class="transactionId"]/span')[0].text.strip())
date = parse_french_date(tr.xpath('./td[@class="date"]')[0].text.strip())
status = tr.xpath('./td[@class="desc"]/ul/li[@class="first"]')[0].text.strip()
#We pass this because it's not transaction
if status in [u'Créé', u'Annulé', u'Suspendu', u'Mis à jour']:
continue
raw = tr.xpath('./td[@class="desc"]/strong')[0].text.strip()
t.parse(date=date, raw=raw)
amount = tr.xpath('./td[@class="price"]/span')[0].text.strip()
t.set_amount(amount)
t._currency = Account.get_currency(amount)
yield t
def transaction_left(self):
return len(self.document.xpath('//div[@class="no-records"]')) == 0
class PartHistoryPage(Page):
def transaction_left(self):
return len(self.document['data']['activity']['COMPLETED']) > 0 or len(self.document['data']['activity']['PENDING']) > 0
def iter_transactions(self, account):
for trans in self.parse(account):
yield trans
def parse(self, account):
transactions = list()
for status in ['PENDING', 'COMPLETED']:
transac = self.document['data']['activity'][status]
for t in transac:
tran = self.parse_transaction(t, account)
if tran:
transactions.append(tran)
transactions.sort(key=lambda tr: tr.rdate, reverse=True)
for t in transactions:
yield t
def parse_transaction(self, transaction, account):
t = FrenchTransaction(transaction['activityId'])
date = parse_french_date(transaction['date'])
raw = transaction.get('counterparty', transaction['displayType'])
t.parse(date=date, raw=raw)
if transaction['currencyCode'] != account.currency:
transaction = self.browser.convert_amount(account, transaction)
try:
t.original_amount = self.format_amount(transaction['originalAmount'], transaction["isCredit"])
t.original_currency = transaction["currencyCode"]
except KeyError:
return
try:
t.amount = self.format_amount(transaction['netAmount'], transaction["isCredit"])
except KeyError:
return
t._currency = transaction['currencyCode']
return t
def format_amount(self, to_format, is_credit):
m = re.search(r"\D", to_format[::-1])
amount = Decimal(re.sub(r'[^\d]', '', to_format))/Decimal((10 ** m.start()))
if is_credit:
return abs(amount)
else:
return -abs(amount)
class HistoryDetailsPage(Page):
def get_converted_amount(self, account):
find_td = self.document.xpath('//td[contains(text(),"' + account.currency + ')")]')
if len(find_td) > 0 :