Luís Murta
9d33df78a8
Implements the argument parser, the manager logic and the DB client methods. Encapsulates the DbClient connection under the _db attribute on the manager. Adds verbose option to enable ORM increased logging.
369 lines
11 KiB
Python
369 lines
11 KiB
Python
from pathlib import Path
|
|
import argparse
|
|
import re
|
|
|
|
from pfbudget.common.types import Command, Operation
|
|
from pfbudget.core.categories import categorize_data
|
|
from pfbudget.input.json import JsonParser
|
|
from pfbudget.input.nordigen import NordigenInput
|
|
from pfbudget.db.sqlite import DatabaseClient
|
|
import pfbudget.reporting.graph
|
|
import pfbudget.reporting.report
|
|
import pfbudget.utils
|
|
|
|
|
|
DEFAULT_DB = "data.db"
|
|
|
|
|
|
class PfBudgetInitialized(Exception):
|
|
pass
|
|
|
|
|
|
class PfBudgetNotInitialized(Exception):
|
|
pass
|
|
|
|
|
|
class DataFileMissing(Exception):
|
|
pass
|
|
|
|
|
|
def argparser() -> argparse.ArgumentParser:
|
|
|
|
help = argparse.ArgumentParser(add_help=False)
|
|
help.add_argument(
|
|
"-db",
|
|
"--database",
|
|
nargs="?",
|
|
help="select current database",
|
|
default=DEFAULT_DB,
|
|
)
|
|
help.add_argument(
|
|
"-q", "--quiet", action="store_true", help="reduces the amount of verbose"
|
|
)
|
|
help.add_argument(
|
|
"-v", "--verbose", action="store_true", help="increases the amount of verbose"
|
|
)
|
|
|
|
period = argparse.ArgumentParser(add_help=False).add_mutually_exclusive_group()
|
|
period.add_argument(
|
|
"--interval", type=str, nargs=2, help="graph interval", metavar=("START", "END")
|
|
)
|
|
period.add_argument("--start", type=str, nargs=1, help="graph start date")
|
|
period.add_argument("--end", type=str, nargs=1, help="graph end date")
|
|
period.add_argument("--year", type=str, nargs=1, help="graph year")
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="does cool finance stuff",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
parser.add_argument(
|
|
"--version",
|
|
action="version",
|
|
version=re.search(
|
|
r'__version__\s*=\s*[\'"]([^\'"]*)[\'"]',
|
|
open("pfbudget/__init__.py").read(),
|
|
).group(1),
|
|
)
|
|
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
"""
|
|
Init
|
|
"""
|
|
p_init = subparsers.add_parser(
|
|
"init",
|
|
description="Initializes the SQLite3 database",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_init.set_defaults(command=Command.Init)
|
|
|
|
"""
|
|
Exporting
|
|
"""
|
|
p_export = subparsers.add_parser(
|
|
"export",
|
|
description="Exports the selected database to a .csv file",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_export.set_defaults(func=lambda args: DatabaseClient(args.database).export())
|
|
|
|
"""
|
|
Parsing
|
|
"""
|
|
p_parse = subparsers.add_parser(
|
|
"parse",
|
|
description="Parses and adds the requested transactions into the selected database",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_parse.add_argument("path", nargs="+", type=str)
|
|
p_parse.add_argument("--bank", nargs=1, type=str)
|
|
p_parse.add_argument("--creditcard", nargs=1, type=str)
|
|
p_parse.add_argument("--category", nargs=1, type=int)
|
|
p_parse.set_defaults(command=Command.Parse)
|
|
|
|
"""
|
|
Categorizing
|
|
"""
|
|
p_categorize = subparsers.add_parser(
|
|
"categorize",
|
|
description="Categorizes the transactions in the selected database",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categorize.set_defaults(command=Command.Categorize)
|
|
|
|
"""
|
|
Graph
|
|
"""
|
|
p_graph = subparsers.add_parser(
|
|
"graph",
|
|
description="Graph of the transactions",
|
|
parents=[help, period],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_graph.add_argument(
|
|
"option",
|
|
type=str,
|
|
choices=["monthly", "discrete", "networth"],
|
|
nargs="?",
|
|
default="monthly",
|
|
help="graph option help",
|
|
)
|
|
p_graph.add_argument("--save", action="store_true")
|
|
p_graph.set_defaults(func=graph)
|
|
|
|
"""
|
|
Report
|
|
"""
|
|
p_report = subparsers.add_parser(
|
|
"report",
|
|
description="Prints report of transaction groups",
|
|
parents=[help, period],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_report.add_argument(
|
|
"option",
|
|
type=str,
|
|
choices=["net", "detailed"],
|
|
nargs="?",
|
|
default="net",
|
|
help="report option help",
|
|
)
|
|
p_report.set_defaults(func=report)
|
|
|
|
"""
|
|
Register bank
|
|
"""
|
|
p_register = subparsers.add_parser(
|
|
"register",
|
|
description="Register a bank",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_register.add_argument("bank", type=str, nargs=1, help="bank option help")
|
|
p_register.add_argument(
|
|
"--requisition", type=str, nargs=1, help="requisition option help"
|
|
)
|
|
p_register.add_argument("--invert", action="store_true")
|
|
p_register.set_defaults(command=Command.Register)
|
|
|
|
"""
|
|
Unregister bank
|
|
"""
|
|
p_register = subparsers.add_parser(
|
|
"unregister",
|
|
description="Unregister a bank",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_register.add_argument("bank", type=str, nargs=1, help="bank option help")
|
|
p_register.set_defaults(command=Command.Unregister)
|
|
|
|
"""
|
|
Nordigen API
|
|
"""
|
|
p_nordigen_access = subparsers.add_parser(
|
|
"token",
|
|
description="Get new access token",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_nordigen_access.set_defaults(command=Command.Token)
|
|
|
|
"""
|
|
(Re)new bank requisition ID
|
|
"""
|
|
p_nordigen_access = subparsers.add_parser(
|
|
"renew",
|
|
description="(Re)new the Bank requisition ID",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_nordigen_access.add_argument("name", nargs=1, type=str)
|
|
p_nordigen_access.add_argument("country", nargs=1, type=str)
|
|
p_nordigen_access.set_defaults(command=Command.Renew)
|
|
|
|
"""
|
|
Downloading through Nordigen API
|
|
"""
|
|
p_nordigen_download = subparsers.add_parser(
|
|
"download",
|
|
description="Downloads transactions using Nordigen API",
|
|
parents=[help, period],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_nordigen_download.add_argument("--id", nargs="+", type=str)
|
|
p_nordigen_download.add_argument("--name", nargs="+", type=str)
|
|
p_nordigen_download.add_argument("--all", action="store_true")
|
|
p_nordigen_download.set_defaults(command=Command.Download)
|
|
|
|
# """
|
|
# List available banks on Nordigen API
|
|
# """
|
|
# p_nordigen_list = subparsers.add_parser(
|
|
# "list",
|
|
# description="Lists banks in {country}",
|
|
# parents=[help],
|
|
# formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
# )
|
|
# p_nordigen_list.add_argument("country", nargs=1, type=str)
|
|
# p_nordigen_list.set_defaults(func=lambda args: nordigen_banks(manager, args))
|
|
|
|
# """
|
|
# Nordigen JSONs
|
|
# """
|
|
# p_nordigen_json = subparsers.add_parser(
|
|
# "json",
|
|
# description="",
|
|
# parents=[help],
|
|
# formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
# )
|
|
# p_nordigen_json.add_argument("json", nargs=1, type=str)
|
|
# p_nordigen_json.add_argument("bank", nargs=1, type=str)
|
|
# p_nordigen_json.add_argument("--invert", action=argparse.BooleanOptionalAction)
|
|
# p_nordigen_json.set_defaults(
|
|
# func=lambda args: manager.parser(JsonParser(vars(args)))
|
|
# )
|
|
|
|
# Add category
|
|
p_categories = subparsers.add_parser(
|
|
"category",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categories_commands = p_categories.add_subparsers(dest="command", required=True)
|
|
p_categories_add = p_categories_commands.add_parser(
|
|
"add",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categories_add.add_argument("category", nargs="+", type=str)
|
|
p_categories_add.add_argument("--group", nargs="?", type=str)
|
|
p_categories_add.set_defaults(command=Command.Category, op=Operation.Add)
|
|
|
|
p_categories_remove = p_categories_commands.add_parser(
|
|
"remove",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categories_remove.add_argument("category", nargs="+", type=str)
|
|
p_categories_remove.add_argument("--group", nargs="?", type=str)
|
|
p_categories_remove.set_defaults(command=Command.Category, op=Operation.Remove)
|
|
|
|
p_categories_addgroup = p_categories_commands.add_parser(
|
|
"addgroup",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categories_addgroup.add_argument("group", nargs="+", type=str)
|
|
p_categories_addgroup.set_defaults(command=Command.Category, op=Operation.AddGroup)
|
|
|
|
p_categories_removegroup = p_categories_commands.add_parser(
|
|
"removegroup",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categories_removegroup.add_argument("group", nargs="+", type=str)
|
|
p_categories_removegroup.set_defaults(command=Command.Category, op=Operation.RemoveGroup)
|
|
|
|
p_categories_updategroup = p_categories_commands.add_parser(
|
|
"updategroup",
|
|
parents=[help],
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
)
|
|
p_categories_updategroup.add_argument("category", nargs="+", type=str)
|
|
p_categories_updategroup.add_argument("--group", nargs=1, type=str)
|
|
p_categories_updategroup.set_defaults(command=Command.Category, op=Operation.UpdateGroup)
|
|
|
|
return parser
|
|
|
|
|
|
def parse(manager, args):
|
|
"""Parses the contents of the path in args to the selected database.
|
|
|
|
Args:
|
|
args (dict): argparse variables
|
|
"""
|
|
for path in args.path:
|
|
if (dir := Path(path)).is_dir():
|
|
for file in dir.iterdir():
|
|
manager.parse(file, vars(args))
|
|
elif Path(path).is_file():
|
|
manager.parse(path, vars(args))
|
|
else:
|
|
raise FileNotFoundError
|
|
|
|
|
|
def graph(args):
|
|
"""Plots the transactions over a period of time.
|
|
|
|
Args:
|
|
args (dict): argparse variables
|
|
"""
|
|
start, end = pfbudget.utils.parse_args_period(args)
|
|
if args.option == "monthly":
|
|
pfbudget.reporting.graph.monthly(
|
|
DatabaseClient(args.database), vars(args), start, end
|
|
)
|
|
elif args.option == "discrete":
|
|
pfbudget.reporting.graph.discrete(
|
|
DatabaseClient(args.database), vars(args), start, end
|
|
)
|
|
elif args.option == "networth":
|
|
pfbudget.reporting.graph.networth(
|
|
DatabaseClient(args.database), vars(args), start, end
|
|
)
|
|
|
|
|
|
def report(args):
|
|
"""Prints a detailed report of the transactions over a period of time.
|
|
|
|
Args:
|
|
args (dict): argparse variables
|
|
"""
|
|
start, end = pfbudget.utils.parse_args_period(args)
|
|
if args.option == "net":
|
|
pfbudget.reporting.report.net(DatabaseClient(args.database), start, end)
|
|
elif args.option == "detailed":
|
|
pfbudget.reporting.report.detailed(DatabaseClient(args.database), start, end)
|
|
|
|
|
|
# def nordigen_banks(manager: Manager, args):
|
|
# input = NordigenInput(manager)
|
|
# input.list(vars(args)["country"][0])
|
|
|
|
|
|
def download(manager, args: dict):
|
|
start, end = pfbudget.utils.parse_args_period(args)
|
|
manager.parser(NordigenInput(manager, args, start, end))
|
|
|
|
|
|
def run():
|
|
args = vars(argparser().parse_args())
|
|
assert "command" in args, "No command selected"
|
|
return args["command"], args
|