Luís Murta
8fe0ecc597
Categorizer will work directly on ORM classes, which will cleanup the code, since changes will automatically be persisted when change the objects. Adds wrapper session class inside the DbClient for the manager to use. The manager will have to have some DB session knowledge, which adds some unfortunate coupling. Removes some unnecessary relations between tables that were added by mistake. category CLI option now uses the manager.
322 lines
9.4 KiB
Python
322 lines
9.4 KiB
Python
from pathlib import Path
|
|
import argparse
|
|
import re
|
|
|
|
from pfbudget.core.categories import categorize_data
|
|
from pfbudget.core.manager import Manager
|
|
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(manager: Manager) -> 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"
|
|
)
|
|
|
|
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(func=lambda args: manager.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(func=lambda args: parse(manager, args))
|
|
|
|
"""
|
|
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(
|
|
func=lambda args: manager.categorize(vars(args))
|
|
)
|
|
|
|
"""
|
|
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(func=lambda args: manager.register(vars(args)))
|
|
|
|
"""
|
|
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(func=lambda args: manager.unregister(vars(args)))
|
|
|
|
"""
|
|
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(func=lambda args: NordigenInput(manager).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(
|
|
func=lambda args: NordigenInput(manager).requisition(
|
|
args.name[0], args.country[0]
|
|
)
|
|
)
|
|
|
|
"""
|
|
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(func=lambda args: download(manager, args))
|
|
|
|
"""
|
|
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)))
|
|
)
|
|
|
|
return parser
|
|
|
|
|
|
def parse(manager: 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: Manager, args):
|
|
start, end = pfbudget.utils.parse_args_period(args)
|
|
manager.parser(NordigenInput(manager, vars(args), start, end))
|
|
|
|
|
|
def run():
|
|
manager = Manager(DEFAULT_DB)
|
|
args = argparser(manager).parse_args()
|
|
args.func(args)
|