1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
#!/usr/bin/python3
import configparser
import argparse
import datetime
import logging
from fints.client import FinTS3PinTanClient
import peewee as p
from playhouse.fields import PickledField
from matrix_client.client import MatrixClient
logging.basicConfig(level=logging.INFO)
parser = argparse.ArgumentParser()
parser.add_argument('--config', nargs=1, required=True, help='Account .ini config file')
parser.add_argument('--db', nargs=1, default='log.db', help='Database')
parser.add_argument('--fetch-last', metavar='DAYS', default=None, type=int, help='Fetch all last DAYS')
args = parser.parse_args()
config = configparser.ConfigParser()
config.read(args.config)
db = p.SqliteDatabase(args.db)
class Transaction(p.Model):
created_date = p.DateField(default=datetime.date.today)
accountnumber = p.CharField()
identifier = p.CharField()
sepa_entry_date = p.DateField()
details = PickledField()
def format_message_html(self):
return """\
{accountnumber} <b>{details[posting_text]}</b>:
<code>{details[amount].amount}</code> {details[amount].currency}
{details[applicant_name]} <i>{details[purpose]} {details[end_to_end_reference]}</i>
""".format(**self._data)
class Meta:
database = db
indexes = ((('accountnumber', 'identifier'), True),)
db.create_tables([Transaction], True)
if args.fetch_last is None:
# Look for last date of a saved transaction or fallback default timesamp
last_transaction_date = Transaction \
.select(p.fn.max(Transaction.sepa_entry_date)) \
.where(Transaction.accountnumber == config['bank']['accountnumber']) \
.scalar(convert=True) or datetime.date.today() - datetime.timedelta(30)
else:
last_transaction_date = datetime.date.today() - datetime.timedelta(args.fetch_last)
f = FinTS3PinTanClient(config['bank']['blz'], config['bank']['username'], config['bank']['password'],
config['bank']['url'])
accounts = f.get_sepa_accounts()
account = list(filter(lambda a: a.accountnumber == config['bank']['accountnumber'], accounts))[0]
logging.info("select transactions on account %s starting from %s", account.accountnumber, last_transaction_date)
transactions = f.get_transactions(account, last_transaction_date, datetime.date.today())
new_transactions = []
for t in transactions:
date = datetime.date.fromordinal(t.data['entry_date'].toordinal()),
identifier = "{applicant_bin}-{applicant_iban}-{mydate}".format(mydate=date, **t.data)
filter_pred = {'accountnumber': config['bank']['accountnumber'], 'identifier': identifier}
data = {'details': t.data, 'sepa_entry_date': date}
try:
record = Transaction.get(**filter_pred)
record.update(data)
logging.info("Updated %s", record.id)
except Transaction.DoesNotExist:
data.update(filter_pred)
record = Transaction.create(**data)
new_transactions += [record]
logging.info("Created %s", record.id)
record.save()
if len(new_transactions) > 0:
client = MatrixClient(config['matrix']['host'])
token = client.login(config['matrix']['user'], config['matrix']['pass'])
room = client.join_room(config['matrix']['room'])
for t in new_transactions:
message = t.format_message_html()
room.send_html(message)
client.logout()
|
