budget/pfbudget/cli/runnable.py
Luís Murta c42a399d3d
Adds the import operation and a timer
to the categorization. We can now import transactions from a csv file,
and later automatically categorize them all.
2023-01-10 23:45:09 +00:00

380 lines
12 KiB
Python

from dotenv import load_dotenv
import argparse
import datetime as dt
import decimal
import os
import re
from pfbudget.common.types import Operation
from pfbudget.db.model import AccountType, Period
from pfbudget.db.sqlite import DatabaseClient
import pfbudget.reporting.graph
import pfbudget.reporting.report
import pfbudget.utils
load_dotenv()
DEFAULT_DB = os.environ.get("DEFAULT_DB")
class PfBudgetInitialized(Exception):
pass
class PfBudgetNotInitialized(Exception):
pass
class DataFileMissing(Exception):
pass
def argparser() -> argparse.ArgumentParser:
universal = argparse.ArgumentParser(add_help=False)
universal.add_argument(
"-db",
"--database",
nargs="?",
help="select current database",
default=DEFAULT_DB,
)
universal.add_argument("-v", "--verbose", action="count", default=0)
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=[universal],
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(required=True)
"""
Init
"""
p_init = subparsers.add_parser(
"init",
description="Initializes the SQLite3 database",
parents=[universal],
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p_init.set_defaults(command=Operation.Init)
# Exports transactions to .csv file
export = subparsers.add_parser("export", parents=[period])
export.set_defaults(op=Operation.Export)
export.add_argument("file", nargs=1, type=str)
export_banks = export.add_mutually_exclusive_group()
export_banks.add_argument("--all", action="store_true")
export_banks.add_argument("--banks", nargs="+", type=str)
pimport = subparsers.add_parser("import")
pimport.set_defaults(op=Operation.Import)
pimport.add_argument("file", nargs=1, type=str)
# Parse from .csv
parse = subparsers.add_parser("parse")
parse.set_defaults(op=Operation.Parse)
parse.add_argument("path", nargs="+", type=str)
parse.add_argument("--bank", nargs=1, type=str)
parse.add_argument("--creditcard", nargs=1, type=str)
# Automatic/manual categorization
categorize = subparsers.add_parser("categorize").add_subparsers(required=True)
categorize.add_parser("auto").set_defaults(op=Operation.Categorize)
categorize.add_parser("manual").set_defaults(op=Operation.ManualCategorization)
"""
Graph
"""
p_graph = subparsers.add_parser(
"graph",
description="Graph of the transactions",
parents=[universal, 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=[universal, 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)
# Banks
bank(subparsers.add_parser("bank"))
# Nordigen access token
subparsers.add_parser("token").set_defaults(op=Operation.Token)
# Nordigen requisition id
requisition = subparsers.add_parser("eua")
requisition.set_defaults(op=Operation.RequisitionId)
requisition.add_argument("name", nargs=1, type=str)
requisition.add_argument("country", nargs=1, type=str)
# Download through the Nordigen API
download = subparsers.add_parser("download", parents=[period])
download.set_defaults(op=Operation.Download)
download_banks = download.add_mutually_exclusive_group()
download_banks.add_argument("--all", action="store_true")
download_banks.add_argument("--banks", nargs="+", type=str)
download.add_argument("--dry-run", action="store_true")
# List available banks in country C
banks = subparsers.add_parser("banks")
banks.set_defaults(op=Operation.NordigenCountryBanks)
banks.add_argument("country", nargs=1, type=str)
# Categories
category(subparsers.add_parser("category"))
# Tag
tags(subparsers.add_parser("tag"))
# Link
link(subparsers.add_parser("link"))
return parser
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 bank(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.BankAdd)
add.add_argument("bank", nargs=1, type=str)
add.add_argument("bic", nargs=1, type=str)
add.add_argument("type", nargs=1, type=str, choices=[e.name for e in AccountType])
rem = commands.add_parser("del")
rem.set_defaults(op=Operation.BankDel)
rem.add_argument("bank", nargs="+", type=str)
mod = commands.add_parser("mod")
mod.set_defaults(op=Operation.BankMod)
mod.add_argument("bank", nargs=1, type=str)
mod.add_argument("--bic", nargs=1, type=str)
mod.add_argument("--type", nargs=1, type=str, choices=[e.name for e in AccountType])
mod.add_argument("--remove", nargs="*", default=[], type=str)
nordigen(commands.add_parser("nordigen"))
def nordigen(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.NordigenAdd)
add.add_argument("bank", nargs=1, type=str)
add.add_argument("--bank_id", nargs=1, type=str)
add.add_argument("--requisition_id", nargs=1, type=str)
add.add_argument("--invert", action="store_true")
rem = commands.add_parser("del")
rem.set_defaults(op=Operation.NordigenDel)
rem.add_argument("bank", nargs="+", type=str)
mod = commands.add_parser("mod")
mod.set_defaults(op=Operation.NordigenMod)
mod.add_argument("bank", nargs=1, type=str)
mod.add_argument("--bank_id", nargs=1, type=str)
mod.add_argument("--requisition_id", nargs=1, type=str)
mod.add_argument("--invert", action="store_true")
mod.add_argument("--remove", nargs="*", default=[], type=str)
def category(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.CategoryAdd)
add.add_argument("category", nargs="+", type=str)
add.add_argument("--group", nargs="?", type=str)
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.CategoryRemove)
remove.add_argument("category", nargs="+", type=str)
update = commands.add_parser("update")
update.set_defaults(op=Operation.CategoryUpdate)
update.add_argument("category", nargs="+", type=str)
update.add_argument("--group", nargs="?", type=str)
schedule = commands.add_parser("schedule")
schedule.set_defaults(op=Operation.CategorySchedule)
schedule.add_argument("category", nargs="+", type=str)
schedule.add_argument("period", nargs=1, choices=[e.value for e in Period])
schedule.add_argument("--frequency", nargs=1, default=[1], type=int)
rule = commands.add_parser("rule")
category_rule(rule)
group = commands.add_parser("group")
category_group(group)
def category_group(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.GroupAdd)
add.add_argument("group", nargs="+", type=str)
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.GroupRemove)
remove.add_argument("group", nargs="+", type=str)
def category_rule(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.RuleAdd)
add.add_argument("category", nargs="+", type=str)
rules(add)
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.RuleRemove)
remove.add_argument("id", nargs="+", type=int)
modify = commands.add_parser("modify")
modify.set_defaults(op=Operation.RuleModify)
modify.add_argument("id", nargs="+", type=int)
modify.add_argument("--category", nargs=1, type=str)
rules(modify)
modify.add_argument("--remove", nargs="*", default=[], type=str)
def tags(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.TagAdd)
add.add_argument("tag", nargs="+", type=str)
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.TagRemove)
remove.add_argument("tag", nargs="+", type=str)
rule = commands.add_parser("rule")
tag_rule(rule)
def tag_rule(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
add = commands.add_parser("add")
add.set_defaults(op=Operation.TagRuleAdd)
add.add_argument("tag", nargs="+", type=str)
rules(add)
remove = commands.add_parser("remove")
remove.set_defaults(op=Operation.TagRuleRemove)
remove.add_argument("id", nargs="+", type=int)
modify = commands.add_parser("modify")
modify.set_defaults(op=Operation.TagRuleModify)
modify.add_argument("id", nargs="+", type=int)
modify.add_argument("--tag", nargs=1, type=str)
rules(modify)
def rules(parser: argparse.ArgumentParser):
parser.add_argument("--date", nargs=1, type=dt.date.fromisoformat)
parser.add_argument("--description", nargs=1, type=str)
parser.add_argument("--regex", nargs=1, type=str)
parser.add_argument("--bank", nargs=1, type=str)
parser.add_argument("--min", nargs=1, type=decimal.Decimal)
parser.add_argument("--max", nargs=1, type=decimal.Decimal)
def link(parser: argparse.ArgumentParser):
commands = parser.add_subparsers(required=True)
forge = commands.add_parser("forge")
forge.set_defaults(op=Operation.Forge)
forge.add_argument("original", nargs=1, type=int)
forge.add_argument("links", nargs="+", type=int)
dismantle = commands.add_parser("dismantle")
dismantle.set_defaults(op=Operation.Dismantle)
dismantle.add_argument("original", nargs=1, type=int)
dismantle.add_argument("links", nargs="+", type=int)