bnpparibas: bnpparibas.py comparison

comparison bnpparibas.py @ 17:80648a7a119c default tip

Update

author	Daniele Nicolodi <daniele@grinta.net>
date	Sun, 21 Jan 2018 01:29:18 +0000
parents	af2e222f2dad
children

comparison

equal deleted inserted replaced

-:af2e222f2dad
+:80648a7a119c
 import subprocess
 import sys
 import textwrap
 import time
+from datetime import datetime
 from contextlib import contextmanager
-from datetime import datetime
 from email.mime.text import MIMEText
 from email.utils import format_datetime, localtime, parseaddr
 from io import BytesIO
 from pprint import pprint
 from urllib.parse import urljoin
 import requests
 import click
 from html2text import HTML2Text
 from PIL import Image
+from dateutil.relativedelta import relativedelta
 URL = 'https://mabanque.bnpparibas/'
 # message template
 MESSAGE = """\
 self.quoted = quoted
 @classmethod
 def fromjson(cls, x):
 data = {'id': x['msg']['id'],
-'date': datetime.strptime(x['msg']['id'], '%Y-%m-%d-%H.%M.%S.%f'),
+'date': datetime.datetime.strptime(x['msg']['id'], '%Y-%m-%d-%H.%M.%S.%f'),
 'subject': x['msg']['objet'],
 'sender': x['msg']['emetteur']['nom'],
 'content': x['msg']['contenu'],
 'quoted': None, }
 quoted = x.get('msgAttache')
 url = urljoin(URL, 'serviceinfosclient-wspl/rpc/InfosClient')
 r = self.session.get(url, params={'modeAppel': 0})
 v = self.validate(r)
 return v['data']
-def recent(self):
+def recent(self, startdate, enddate):
 url = urljoin(URL, 'udc-wspl/rest/getlstcpt')
 r = self.session.get(url)
 v = self.validate(r)
 account = v['data']['infoUdc']['familleCompte'][0]['compte'][0]
 url = urljoin(URL, 'rop-wspl/rest/releveOp')
 data = json.dumps({'ibanCrypte': account['key'],
 'pastOrPending': 1, 'triAV': 0,
-'startDate': None, # ddmmyyyy
+'startDate': '{:%d%m%Y}'.format(startdate), # ddmmyyyy
-'endDate': None})
+'endDate': '{:%d%m%Y}'.format(enddate)})
 headers = {'Content-Type': 'application/json'}
 r = self.session.post(url, headers=headers, data=data)
 v = self.validate(r)
 return v['data']
 url = urljoin(URL, 'bmm-wspl/recupMsg')
 r = self.session.get(url, params={'identifiant': mid})
 v = self.validate(r)
 return v['data']
-def records0(self):
-# required to set some cookies required by the next call
-url = urljoin(URL, 'fr/connexion/virements-services/releves-en-ligne')
-r = self.session.get(url)
-self.validate(r)
-url = urljoin(URL, 'demat-wspl/rest/initialisationDemat')
-r = self.session.get(url)
-v = self.validate(r)
-for branch in v['data']['initialisationDemat']['arbres']:
-for leave in branch.get('arbre', ( )):
-if leave['typeDoc'] == 'RELEV':
-data = leave
-break
-self.iban = data['ibans'][0]['ibanCrypte']
-query = {'famDoc': data['codeFamille'],
-'idTypeDocument': data['idBranche'],
-'listeIbanCrypte': [ self.iban, ],
-'typeCpt': data['typeCompte'],
-'typeDoc': data['typeDoc'],
-'typeFamille': 'R001'} # ???
-url = urljoin(URL, 'demat-wspl/rest/consultationDemat')
-data = json.dumps(query)
-headers = {'Content-Type': 'application/json'}
-r = self.session.post(url, headers=headers, data=data)
-v = self.validate(r)
-years = v['data']['consultationDemat']['listeCompte'][0]['listeAnnee']
-query['codeProduit'] = ''
-documents = []
-url = urljoin(URL, 'demat-wspl/rest/rechercheDemat')
-for year in years:
-query['anneeSelectionnee'] = year
-data = json.dumps(query)
-r = self.session.post(url, headers=headers, data=data)
-v = self.validate(r)
-documents += v['data']['consultationDemat']['listeCompte'][0]['listeDocument']
-return documents
 def records(self):
 url = urljoin(URL, 'fr/connexion/mes-outils/dematerialisation')
 r = self.session.get(url)
 self.validate(r)
 def transactions(conf):
 db = sqlite3.connect(conf['DATABASE'])
 db.execute('''CREATE TABLE IF NOT EXISTS transactions (id INTEGER PRIMARY KEY)''')
+db.execute('''CREATE TABLE IF NOT EXISTS last (date TEXT)''')
 sendmail = Mailer(host=conf['SMTPHOST'],
 port=conf['SMTPPORT'],
 starttls=conf['SMTPSTARTTLS'],
 username=conf['SMTPUSER'],
 encrypt = GPG(conf['GNUPGHOME']).encrypt
 remote = BNPParibas()
 remote.login(conf['USERNAME'], conf['PASSWORD'])
-recent = remote.recent()
+# last date
-data = recent['listerOperations']['compte']
+curs = db.cursor()
-transactions = [ Transaction.fromjson(x) for x in data['operationPassee'] ]
+curs.execute('''SELECT date FROM last''')
-balance = data['soldeDispo']
+row = curs.fetchone()
+startdate = datetime.strptime(row[0], '%d-%m-%Y') if row else datetime(2016, 10, 1)
+transactions = []
+while startdate <= datetime.now():
+enddate = startdate + relativedelta(months=2)
+recent = remote.recent(startdate, enddate)
+data = recent['listerOperations']['compte']
+transactions += [ Transaction.fromjson(x) for x in data['operationPassee'] ]
+balance = data['soldeDispo']
+startdate = enddate
+transactions.sort(key=lambda x: datetime.strptime(x.date, '%d-%m-%Y'))
 curs = db.cursor()
 unseen = []
 for t in transactions:
 curs.execute('''SELECT COUNT(*) FROM transactions WHERE id = ?''', (t.id, ))
 message['Date'] = format_datetime(localtime())
 sendmail(message)
 curs.executemany('''INSERT INTO transactions (id) VALUES (?)''', ((x.id, ) for x in unseen))
+curs.execute('''INSERT OR REPLACE INTO 'last' (date) VALUES (?)''', (transactions[-1].date, ))
 db.commit()
 def messages(conf):
 db = sqlite3.connect(conf['DATABASE'])

Mercurial > hg > bnpparibas

comparison bnpparibas.py @ 17:80648a7a119c default tip