Skip to content

Instantly share code, notes, and snippets.



Last active Sep 22, 2019
What would you like to do?
Simple api to fetch accounts, balances and transactions from Caixa Geral de Depósitos's ( CGD ) API used by their Windows 8 application.
# gist:
# Usage:
# session = cgd.CgdSession(uid, password)
# session.login()
# session.load_latest_transactions(account_key)
# 'session.known_accounts' is now populated with the initial accounts taken from the login response,
# and the data for the 'account_key' account.
# session.load_latest_transactions(account_key) loads the latest transactions and balances for a given account.
# The CGD API is very "screen-driven" for their app, so this ends up being a rather convoluted way to get data.
# But it works for me, so it might work for someone else.
import requests
from functools import wraps
from pprint import pformat
from decimal import Decimal
from datetime import datetime, timedelta
from os import path
import itertools
import json
def with_base(f):
def wrapper(*args, **kwds):
return ""+f(*args, **kwds)
return wrapper
class Urls:
def authentication(user_id):
return "/apps/r/co/li?u="+user_id
def account_balance(account_key):
return "/apps/r/cnt/sm/"+account_key
def account_details(account_key):
return "/apps/r/cnt/dc/"+account_key
def account_more_details():
return "/apps/r/cnt/dc/m"
class CgdException(RuntimeError):
class CgdCommunicationException(CgdException):
"""There was an error communicating with the CGD API"""
class InternalCgdException(CgdException):
"""Something went wrong inside this library"""
def populate_object_from_mapping_and_conversions(obj, json, mapping, conversions = {}):
errors = []
for k,v in mapping.items():
if not json.has_key(k):
errors.append("missing key '%s' corresponding to property '%s'" % (k,v))
value = json[k]
if conversions.has_key(k):
value = conversions[k](value)
setattr(obj, v, value)
if any(errors):
message = """
Errors while parsing from
\t%s""" % (pformat(json), "\n\t".join(errors))
raise CgdCommunicationException(message)
return obj
def serialize_date(date):
return date.strftime("%Y-%m-%dT%H:%M:%S%z")
def deserialize_date(date_string):
return datetime.strptime(date_string, "%Y-%m-%d %H:%M:%S")
def deserialize_money(money_int):
return money_int/Decimal(100)
def deserialize_transaction_type(type_string):
l = type_string.lower()
if l == 'c':
return 'credit'
if l == 'd':
return 'debit'
return 'unknown transaction type '+type_string
def create_requests_session():
proxies = {
#"https": "",
headers = {
return requests.session(headers = headers, proxies = proxies)
class Transaction:
def from_account_details_transactions(cls, transaction_json):
mapping = {
'moeo': 'original_currency',
'mon': 'amount',
'ndc': 'document_number',
'nmv': 'movement_number',
'saps': 'balance_after',
'sdaps': 'available_balance_after',
'tpm': 'transaction_type',
conversions = {
'tpm': deserialize_transaction_type
return populate_object_from_mapping_and_conversions(Transaction(), transaction_json, mapping, conversions)
class Balance:
def from_balance_movements(cls, balance_movements_slds):
mapping = {
'moe': 'currency',
'ldd': 'available_overdraft',
'ldde': 'available_overdraft_euros',
'scne': 'balance_euros',
'scnt': 'balance_currency',
'sdse': 'available_balance_euros',
'sdsp': 'available_balance_currency',
conversions = {
return populate_object_from_mapping_and_conversions(Balance(), balance_movements_slds, mapping, conversions)
def from_account_details_balance(cls, details_slds):
mapping = {
'acb': 'awaiting_charge',
'ctv': 'captive_amount',
'moe': 'currency',
'ldd': 'available_overdraft',
'ldde': 'available_overdraft_euros',
'ldn': 'negotiated_overdraft',
'ldu': 'available_overdraft',
'scne': 'balance_euros',
'scnt': 'balance_currency',
'sdse': 'available_balance_euros',
'sdsp': 'available_balance_currency',
'ti': 'unavailable_amount',
conversions = {
'acb': deserialize_money,
'ctv': deserialize_money,
'ldd': deserialize_money,
'ldde': deserialize_money,
'ldn': deserialize_money,
'ldu': deserialize_money,
'scne': deserialize_money,
'scnt': deserialize_money,
'sdse': deserialize_money,
'sdsp': deserialize_money,
'ti': deserialize_money,
return populate_object_from_mapping_and_conversions(Balance(), details_slds, mapping, conversions)
class Account:
def from_auth_result(cls, auth_result_lcnt):
mapping = {
account = Account()
account.transactions = []
return populate_object_from_mapping_and_conversions(account, auth_result_lcnt, mapping)
class CgdSession:
def __init__(self, userid, password):
self._s = create_requests_session()
self._auth = (userid, password)
def login(self):
auth_response = self._s.get(Urls.authentication(self._auth[0]), auth = self._auth)
if auth_response.status_code != 200:
message = auth_response.content
message = auth_response.json["m"]
raise CgdCommunicationException("Authentication failed: " + message)
if not auth_response.json.has_key("lcnt"):
raise CgdCommunicationException("Missing key 'lcnt' in authentication response")
self.known_accounts = { account.key : account for account in [Account.from_auth_result(account_json) for account_json in auth_response.json['lcnt']]}
def load_account_balance(self, account_key):
if not self.known_accounts.has_key(account_key):
# we can avoid this when we figure out a way to just load the account info
# so far this is always a side effect of loading something else
raise InternalCgdException("Account key %s wasn't a known account" % account_key)
response = self._s.get(Urls.account_balance(account_key))
account = self.known_accounts[account_key]
account.balance = Balance.from_balance_movements(response.json["slds"])
return account.balance
def load_latest_transactions(self, account_key, page_size = 30, first_movement_number = -1):
response = self._s.get(Urls.account_details(account_key))
for lcnt in response.json["lcnt"]:
possible_new_account = Account.from_auth_result(lcnt)
if not self.known_accounts.has_key(possible_new_account.key):
self.known_accounts[possible_new_account.key] = possible_new_account
account = self.known_accounts[account_key]
account.balance = Balance.from_account_details_balance(response.json["slds"])
transactions_bag = list(Transaction.from_account_details_transactions(lmov) for lmov in response.json["lmov"])
page_key = response.json["pkl"]
while(len(transactions_bag) <= page_size and min(t.movement_number for t in transactions_bag) > first_movement_number):
until = datetime.strptime(page_key[1], "%Y-%m-%d")
more_details_response = self._grab_more_details(self._s, account_key, until-timedelta(days = 30), until, page_key)
if not more_details_response.json.has_key("lmov") or len(more_details_response.json["lmov"]) == 0:
page_key = response.json["pkl"]
transactions_bag.extend(Transaction.from_account_details_transactions(lmov) for lmov in more_details_response.json["lmov"])
seen = set()
account.transactions_bag = transactions_bag
account.transactions.extend(t for t in transactions_bag if t.movement_number not in seen and not seen.add(t.movement_number))
def _grab_more_details(self, s, account_key, _from, until, page_key):
data = {
headers = {'Content-Type': 'application/json', 'Expect':'100-continue'}
return, data=json.dumps(data), headers = headers)

This comment has been minimized.

Copy link

@Hurleyking Hurleyking commented Sep 21, 2019

Boas bruno, so possivel correr script no windows? Como podemos obter uid? E possivel passar credenciais e obter dados? Obrigado


This comment has been minimized.

Copy link
Owner Author

@brunomlopes brunomlopes commented Sep 22, 2019


1- Tanto quanto sei isto funciona em qualquer SO.
2- Uid é o user id.
3- Diria que sim

Atenção que isto é de 2013. Não sei se a API continua igual, nem tenho forma de testar

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.