Last active
October 5, 2020 15:22
-
-
Save CRamsan/e07202eae0db2bea802228c8c1a14d03 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python2.7 | |
# Thanks to haochi for developing the personalcapital pip | |
# https://github.com/haochi/personalcapital | |
# Most of the work on this script is based on the main file from that repo. | |
from datetime import datetime, timedelta | |
from personalcapital import PersonalCapital, RequireTwoFactorException, TwoFactorVerificationModeEnum | |
import getpass | |
import json | |
import logging | |
import os | |
import sqlite3 | |
TABLE_NAME = 'transactions' | |
# List of tuples that define the table schema | |
# The first tuple will be used as the primary key | |
# The following parrameters are ignored: | |
# - currency | |
# - originalDescription | |
# - accountName | |
COLUMN_MATRIX = (('userTransactionId', 'INTEGER'), | |
('symbol', 'TEXT'), | |
('isInterest', 'INTEGER'), | |
('netCost', 'REAL'), | |
('cusipNumber', 'TEXT'), | |
('description', 'TEXT'), | |
('memo', 'TEXT'), | |
('isCredit', 'INTEGER'), | |
('isEditable', 'INTEGER'), | |
('isCashOut', 'INTEGER'), | |
('merchantId', 'TEXT'), | |
('price', 'REAL'), | |
('holdingType', 'TEXT'), | |
('lotHandling', 'TEXT'), | |
('customReason', 'TEXT'), | |
('isDuplicate', 'INTEGER'), | |
('resultType', 'TEXT'), | |
('isSpending', 'INTEGER'), | |
('amount', 'REAL'), | |
('checkNumber ', 'TEXT'), | |
('quantity', ' REAL'), | |
('transactionTypeId', 'INTEGER'), | |
('isIncome', 'INTEGER'), | |
('includeInCashManager', 'INTEGER'), | |
('merchant', 'TEXT'), | |
('isNew', 'INTEGER'), | |
('isCashIn', 'INTEGER'), | |
('transactionDate', 'TEXT'), | |
('transactionType', 'TEXT'), | |
('accountId', 'TEXT'), | |
('originalAmount', 'REAL'), | |
('isCost', 'INTEGER'), | |
('userAccountId', 'INTEGER'), | |
('simpleDescription', 'TEXT'), | |
('investmentType', 'TEXT'), | |
('runningBalance', 'REAL'), | |
('hasViewed', 'INTEGER'), | |
('caytegoryId', 'INTEGER'), | |
('status', 'TEXT')) | |
''' | |
Function to generate the comand to create a table. This function will generate the string based on the COLUMN_MATRIX. The first tuple will be used as the primary key. | |
At the moment there are no foreign key, although that is planned in the future. | |
''' | |
def table_create_command(): | |
command = 'CREATE TABLE IF NOT EXISTS ' + TABLE_NAME + ' (' + COLUMN_MATRIX[0][0] + ' ' + COLUMN_MATRIX[0][1] + ' PRIMARY KEY, ' | |
column_count = len(COLUMN_MATRIX) | |
for index in range(1,column_count): | |
command += COLUMN_MATRIX[index][0] + ' ' + COLUMN_MATRIX[index][1] | |
if index < (column_count - 1): | |
command += ', ' | |
command += ')' | |
return command | |
''' | |
Helper method to generate the list of column names. | |
''' | |
def generate_column_names(): | |
column_count = len(COLUMN_MATRIX) | |
command = '(' | |
for index in range(column_count): | |
command += COLUMN_MATRIX[index][0] | |
if index < (column_count - 1): | |
command += ', ' | |
command += ')' | |
return command | |
''' | |
In order to keep the SQL commands out of the main function, | |
I am just moving them here. This function does not commit changes. | |
''' | |
def create_table_if_needed(cursor): | |
cursor.execute(table_create_command()) | |
''' | |
This function takes a transaction dictionary and a cursor as input. | |
If the transaction already exists, then the new value will replace | |
the existing one. | |
''' | |
def insert_or_update_record(transaction, cursor): | |
column_count = len(COLUMN_MATRIX) | |
found_properties = {} | |
for index in range(column_count): | |
key = COLUMN_MATRIX[index][0] | |
if key in transaction: | |
found_properties[key] = transaction[key] | |
found_columns = found_properties.keys() | |
command = 'INSERT OR REPLACE INTO ' + TABLE_NAME + ' (' + ', '.join(found_columns) + ') values (' | |
column_count = len(found_columns) | |
for index in range(column_count): | |
value = found_properties[found_columns[index]] | |
if isinstance(value, basestring): | |
value = '\'' + value.replace('\'', '') + '\'' | |
elif isinstance(value, bool): | |
if value: | |
value = '1' | |
else: | |
value = '0' | |
else: | |
value = str(value) | |
command += value | |
if index < (column_count - 1): | |
command += ', ' | |
command += ');' | |
try: | |
cursor.execute(command) | |
except sqlite3.OperationalError: | |
print (command) | |
# Python 2 and 3 compatibility | |
if hasattr(__builtins__, 'raw_input'): | |
input = raw_input | |
class PewCapital(PersonalCapital): | |
""" | |
Extends PersonalCapital to save and load session | |
So that it doesn't require 2-factor auth every time | |
""" | |
def __init__(self): | |
PersonalCapital.__init__(self) | |
self.__session_file = 'session.json' | |
def load_session(self): | |
try: | |
with open(self.__session_file) as data_file: | |
cookies = {} | |
try: | |
cookies = json.load(data_file) | |
except ValueError as err: | |
logging.error(err) | |
self.set_session(cookies) | |
except IOError as err: | |
logging.error(err) | |
def save_session(self): | |
with open(self.__session_file, 'w') as data_file: | |
data_file.write(json.dumps(self.get_session())) | |
''' | |
Helper function to retrieve the email from the enviroment variable | |
''' | |
def get_email(): | |
email = os.getenv('PEW_EMAIL') | |
if not email: | |
print('You can set the environment variables for PEW_EMAIL and PEW_PASSWORD so the prompts don\'t come up every time') | |
return input('Enter email:') | |
return email | |
''' | |
XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX | |
Helper function to retrieve your password from the enviroment variable. | |
USE WITH CARE! YOU DONT WANT YOUR PASSWORD LYING AROUND YOUR HISTORY OR | |
SCRIPTS!! | |
XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX | |
''' | |
def get_password(): | |
password = os.getenv('PEW_PASSWORD') | |
if not password: | |
return getpass.getpass('Enter password:') | |
return password | |
def main(): | |
email, password = get_email(), get_password() | |
pc = PewCapital() | |
pc.load_session() | |
try: | |
pc.login(email, password) | |
except RequireTwoFactorException: | |
# For some reason doing EMAIL TFA is not working. So I have to rely on SMS. | |
pc.two_factor_challenge(TwoFactorVerificationModeEnum.SMS) | |
pc.two_factor_authenticate(TwoFactorVerificationModeEnum.SMS, input('code: ')) | |
pc.authenticate_password(password) | |
accounts_response = pc.fetch('/newaccount/getAccounts') | |
now = datetime.now() | |
date_format = '%Y-%m-%d' | |
days = 360 | |
start_date = (now - (timedelta(days=days+1))).strftime(date_format) | |
end_date = (now - (timedelta(days=1))).strftime(date_format) | |
transactions_response = pc.fetch('/transaction/getUserTransactions', { | |
'sort_cols': 'transactionTime', | |
'sort_rev': 'true', | |
'page': '0', | |
'rows_per_page': '100', | |
'startDate': start_date, | |
'endDate': end_date, | |
'component': 'DATAGRID' | |
}) | |
pc.save_session() | |
accounts = accounts_response.json()['spData'] | |
print('Networth: {0}'.format(accounts['networth'])) | |
transactions = transactions_response.json()['spData'] | |
transaction_list = transactions['transactions'] | |
print('Number of transactions between {0} and {1}: {2}'.format(transactions['startDate'], transactions['endDate'], len(transaction_list))) | |
db = sqlite3.connect('data.db') | |
cursor = db.cursor() | |
# Create the table if it does not exist | |
create_table_if_needed(cursor) | |
db.commit() | |
for transaction in transaction_list: | |
insert_or_update_record(transaction, cursor) | |
db.commit() | |
db.close() | |
if __name__ == '__main__': | |
main() |
Do you mean into https://github.com/haochi/personalcapital ? I have not looked into trying to merge it.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Did this ever get merged into the original personal capital project? The additions here are super helpful