budget/pfbudget/runnable.py
Luís Murta 3e20ae97db
parser command now writes do DB
`parse_data` from parser.py now takes a `DBManager` and runs
`insert_transactions` after parsing.
`DBManager` now takes `Transaction` type and returns the same on select
queries.
`DBManager` refactored with extensive type hinting.
`Transaction` updated to reflect use in `DBManager`.
2021-06-21 23:49:23 +01:00

183 lines
4.8 KiB
Python

from pathlib import Path
import argparse
import datetime as dt
from .categories import categorize_data
from .database import DBManager
from .graph import discrete, monthly
from .parsers import parse_data
from .transactions import load_transactions, save_transactions
from . import report
DEFAULT_DB = "data.db"
class PfBudgetInitialized(Exception):
pass
class PfBudgetNotInitialized(Exception):
pass
class DataFileMissing(Exception):
pass
def argparser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="does cool finance stuff")
parser.add_argument("--db", help="select current database", default=DEFAULT_DB)
parser.add_argument("-q", "--quiet", help="quiet")
parser.add_argument("--version")
subparsers = parser.add_subparsers(
dest="command", required=True, help="sub-command help"
)
"""
Init
"""
p_init = subparsers.add_parser("init", help="init help")
p_init.set_defaults(func=lambda args: DBManager(args.db))
"""
Exporting
"""
p_export = subparsers.add_parser("export", help="export help")
p_export.set_defaults(func=lambda args: DBManager(args.db).export())
"""
Parsing
"""
p_parse = subparsers.add_parser("parse", help="parse help")
p_parse.add_argument("path", nargs="+", type=str)
p_parse.add_argument("--bank", nargs=1, type=str)
p_parse.set_defaults(func=parse)
"""
Categorizing
"""
p_categorize = subparsers.add_parser("categorize", help="parse help")
p_categorize.set_defaults(func=categorize)
p_graph = subparsers.add_parser("graph", help="graph help")
p_report = subparsers.add_parser("report", help="report help")
p_status = subparsers.add_parser("status", help="status help")
p_graph.add_argument(
"option",
type=str,
choices=["monthly", "discrete"],
nargs="?",
default="monthly",
help="graph option help",
)
p_graph_interval = p_graph.add_mutually_exclusive_group()
p_graph_interval.add_argument(
"--interval", type=str, nargs=2, help="graph interval", metavar=("START", "END")
)
p_graph_interval.add_argument("--start", type=str, nargs=1, help="graph start date")
p_graph_interval.add_argument("--end", type=str, nargs=1, help="graph end date")
p_graph_interval.add_argument("--year", type=str, nargs=1, help="graph year")
p_status.set_defaults(func=status)
p_graph.set_defaults(func=graph)
p_report.set_defaults(func=f_report)
return parser
def parse(args, db):
"""Parser
Parses the contents of the raw directory into the data files, and
categorizes the transactions
Args:
args (dict): argparse variables
db (DBManager): db connection manager
"""
for path in args.path:
if (dir := Path(path)).is_dir():
for file in dir.iterdir():
parse_data(db, file, args.bank)
elif Path(path).is_file():
parse_data(db, path, args.bank)
else:
raise FileNotFoundError
def categorize(args, db):
"""Categorization
Automatically categorizes transactions based on the regex of each
category. Manually present the remaining to the user
Args:
args (dict): argparse variables
db (DBManager): db connection manager
"""
categorize_data(db)
def status(state, args):
"""Status
Prints the state file.
Args:
state (PFState): Internal state of the program
args (dict): argparse variables
"""
print(state)
def graph(state, args):
"""Graph
Plots the transactions over a period of time.
Args:
state (PFState): Internal state of the program
args (dict): argparse variables
"""
start, end = None, None
if args.start or args.interval:
start = dt.datetime.strptime(args.start[0], "%Y/%m/%d").date()
if args.end or args.interval:
end = dt.datetime.strptime(args.end[0], "%Y/%m/%d").date()
if args.interval:
start = dt.datetime.strptime(args.interval[0], "%Y/%m/%d").date()
end = dt.datetime.strptime(args.interval[1], "%Y/%m/%d").date()
if args.year:
start = dt.datetime.strptime(args.year[0], "%Y").date()
end = dt.datetime.strptime(
str(int(args.year[0]) + 1), "%Y"
).date() - dt.timedelta(days=1)
if args.option == "monthly":
monthly(state, start, end)
elif args.option == "discrete":
discrete(state, start, end)
def f_report(state, args):
"""Report
Prints a detailed report of the transactions over a period of time.
Args:
state (PFState): Internal state of the program
args (dict): argparse variables
"""
report.net(state)
def run():
db = DBManager("transactions.db")
args = argparser().parse_args()
args.func(args, db)