budget/publisher.py

import sys
import os
import os.path
import re
from datetime import datetime, timedelta
from mastodon import Mastodon
import psycopg2
import pdb

def get_bills():

    current_year = now.year

    bills = 0

    try:

        conn = None

        conn = psycopg2.connect(database = budget_db, user = budget_db_user, password = "", host = "/var/run/postgresql", port = "5432")

        cur = conn.cursor()

        cur.execute("select sum (domain) + sum(server) + sum(backup) + sum(fileserver) +sum(setup) from bills where date_part('year', datetime) = (%s)", (current_year,))

        row = cur.fetchone()

        if row[0] != None:

            bills = row[0]

        cur.close()

        return bills

    except (Exception, psycopg2.DatabaseError) as error:

        sys.exit(error)

    finally:

        if conn is not None:

            conn.close()

def year_donations():

    current_year = now.year

    donations = 0

    try:

        conn = None

        conn = psycopg2.connect(database = budget_db, user = budget_db_user, password = "", host = "/var/run/postgresql", port = "5432")

        cur = conn.cursor()

        cur.execute("select sum (donations) + sum(owner) from incomes where date_part('year', datetime) = (%s)", (current_year,))

        row = cur.fetchone()

        if row[0] != None:

            donations = row[0]

        else:

            donations = 0

        cur.close()

        return donations

    except (Exception, psycopg2.DatabaseError) as error:

        sys.exit(error)

    finally:

        if conn is not None:

            conn.close()

def month_donations():

    current_month = now.month

    donations = 0

    try:

        conn = None

        conn = psycopg2.connect(database = budget_db, user = budget_db_user, password = "", host = "/var/run/postgresql", port = "5432")

        cur = conn.cursor()

        cur.execute("select count(donations) from incomes where date_part('month', datetime) = (%s)", (current_month,))

        row = cur.fetchone()

        if row[0] != None:

            n_donations = row[0]

        else:

            n_donations = 0

        cur.execute("select sum (donations) + sum(owner) from incomes where date_part('month', datetime) = (%s)", (current_month,))

        row = cur.fetchone()

        if row[0] != None:

            donations = row[0]

        else:

            donations = 0

        cur.close()

        return (n_donations, donations)

    except (Exception, psycopg2.DatabaseError) as error:

        sys.exit(error)

    finally:

        if conn is not None:

            conn.close()

def get_fixed_fees():

    fees_desc_lst = []

    fees_types_lst = []

    fees_lst = []


    try:

        conn = None

        conn = psycopg2.connect(database = budget_db, user = budget_db_user, password = "", host = "/var/run/postgresql", port = "5432")

        cur = conn.cursor()

        cur.execute("select fee_description, payment_type, fee from fees")

        rows = cur.fetchall()

        for row in rows:

            fees_desc_lst.append(row[0])

            fees_types_lst.append(row[1])

            fees_lst.append(row[2])

        cur.close()

        return (fees_desc_lst, fees_types_lst, fees_lst)

    except (Exception, psycopg2.DatabaseError) as error:

        sys.exit(error)

    finally:

        if conn is not None:

            conn.close()


def mastodon():

    # Load secrets from secrets file
    secrets_filepath = "secrets/secrets.txt"
    uc_client_id     = get_parameter("uc_client_id",     secrets_filepath)
    uc_client_secret = get_parameter("uc_client_secret", secrets_filepath)
    uc_access_token  = get_parameter("uc_access_token",  secrets_filepath)

    # Load configuration from config file
    config_filepath = "config/config.txt"
    mastodon_hostname = get_parameter("mastodon_hostname", config_filepath)
    bot_username = get_parameter("bot_username", config_filepath)

    # Initialise Mastodon API
    mastodon = Mastodon(
        client_id = uc_client_id,
        client_secret = uc_client_secret,
        access_token = uc_access_token,
        api_base_url = 'https://' + mastodon_hostname,
    )

    # Initialise access headers
    headers={ 'Authorization': 'Bearer %s'%uc_access_token }

    return (mastodon, mastodon_hostname, bot_username)

def db_config():

    # Load db configuration from config file
    config_filepath = "config/db_config.txt"
    budget_db = get_parameter("budget_db", config_filepath)
    budget_db_user = get_parameter("budget_db_user", config_filepath)

    return (budget_db, budget_db_user)

def get_parameter( parameter, file_path ):

    if not os.path.isfile(file_path):
        print("File %s not found, exiting."%file_path)
        sys.exit(0)

    with open( file_path ) as f:
        for line in f:
            if line.startswith( parameter ):
                return line.replace(parameter + ":", "").strip()

    print(file_path + "  Missing parameter %s "%parameter)
    sys.exit(0)

###############################################################################
# main

if __name__ == '__main__':

    mastodon, mastodon_hostname, bot_username = mastodon()

    budget_db, budget_db_user = db_config()

    now = datetime.now()

    donations, sum_donations = month_donations()

    year_donations = year_donations()

    bills = get_bills()

    fees_desc_lst, fees_types_lst, fees_lst = get_fixed_fees()

    toot_text = 'Finances de ' + mastodon_hostname + '\n\n'

    toot_text += 'Any: ' + str(now.year) + '\n'

    toot_text += '\n'

    toot_text += 'Pagaments: ' + str(bills) + '€' + '\n'

    toot_text += 'Donacions: ' + str(year_donations) + '€' + '\n'

    toot_text += f'Balanç: {str(year_donations - bills)}€\n'

    if year_donations != 0 and bills != 0:

        toot_text += 'Objectiu assolit: ' + str(round((year_donations * 100)/bills,2)) + '%\n\n'

    i = 0

    toot_text += f'Donacions rebudes (mes actual): {str(donations)} / {sum_donations}€\n\n'

    while i < len(fees_lst):

        toot_text += 'Cost del ' + str(fees_desc_lst[i]) + ': ' + str(fees_lst[i]) + '€ (' + str(fees_types_lst[i]) + ')\n'

        i += 1

    print(toot_text)

    mastodon.status_post(toot_text, in_reply_to_id=None)