diff --git a/bin/file-to-account b/bin/file-to-account new file mode 100755 index 00000000..99d074a8 --- /dev/null +++ b/bin/file-to-account @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +"""Move a file to a uniquely matching account name under a directory hierarchy. + +You can use this script to move a filename under a particular account-named +directory in a directory hierarchy, like this: + + file-to-account --documents=DOCUMENTS_DIR ACCOUNT_PATTERN FILENAME... + +For example: + + file-to-account -d ~/my-documents bofa my-bank-download.csv + +If the -o flag is not provided, the environment variable +$BEANGULP_DOCUMENTS_DIR is used automatically. + +The script will list all the subdirectories under this root, attempt to uniquely +match (case insensitively) the pattern given to a single directory name (e.g., +"bofa" above), and if found, will move the file there. Warnings are issued for +files that could not be moved. + +This is intended to be a convenient, quick tool for filing away downloads +manually. +""" +__copyright__ = "Copyright (C) 2022 Martin Blais" +__license__ = "GNU GPLv2" + +from os import path +import argparse +import fnmatch +import functools +import logging +import os +import re +import shutil + + +def main(): + logging.basicConfig(level=logging.INFO, format="%(levelname)-8s: %(message)s") + parser = argparse.ArgumentParser(description=__doc__.strip()) + parser.add_argument("account_pattern", help="Account name globbing pattern") + parser.add_argument("filenames", nargs="+", help="List of filenames to process.") + parser.add_argument( + "-d", + "-o", + "--documents", + action="store", + default=os.environ.get("BEANGULP_DOCUMENTS_DIR", None), + help="Documents directory root", + ) + args = parser.parse_args() + + # Validate that we have an output. + if not args.documents: + parser.error( + "Please set environment variable BEANGULP_DOCUMENTS_DIR or " + "provide --documents." + ) + + # Get the list of all available directories. + all_dirs = [ + path.join(root, dirname) + for root, dirs, _ in os.walk(path.abspath(args.documents)) + for dirname in dirs + ] + + # Find a uniquely matching directory. + matches = [ + dirname + for dirname in all_dirs + if re.search(args.account_pattern, dirname, flags=re.I) + ] + if len(matches) == 0: + logging.error(f"No matches for pattern '{args.account_pattern}'") + elif len(matches) > 1: + logging.error(f"Ambiguous matches for pattern '{args.account_pattern}':") + for match in matches: + logging.error(f" '{match}'") + else: + outdir = matches[0] + undo_filename = "/tmp/file-to-account-undo.sh" + logging.info(f"Undo script is located in {undo_filename}") + with open(undo_filename, "w") as f: + os.chmod(undo_filename, 0o755) + pr = functools.partial(print, file=f) + pr("#!/bin/bash") + for src in args.filenames: + dst = path.join(outdir, path.basename(src)) + logging.info(f"Moving '{src}' to '{dst}'") + shutil.copy(src, dst) + os.remove(src) + pr(f"mv '{dst}' '{src}'") + + +if __name__ == "__main__": + main() diff --git a/bin/region-errors-to-table b/bin/region-errors-to-table new file mode 100755 index 00000000..68662dc3 --- /dev/null +++ b/bin/region-errors-to-table @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Given a region in a Beancount, extract incomplete transactions to a table. + +This script accepts a top-level filename, a filename and start:end locations. It +will look for transactions with a `!` flag either on the transaction or on any +of its postings, and convert those to a simple table. +""" +__copyright__ = "Copyright (C) 2022 Martin Blais" +__license__ = "GNU GPLv2" + +from itertools import chain +from os import path +from typing import Iterable, List, Tuple +import argparse +import fnmatch +import functools +import logging +import os +import re +import shutil + +import click +import petl + +from beancount import loader +from beancount.core import data +from beancount.core import flags +from beancount.parser import printer +from beancount.scripts import doctor + + +petl.config.look_style = "minimal" +petl.config.failonerror = True + + +@click.command() +@click.argument("filename", type=click.Path(resolve_path=True, exists=True)) +@click.argument("region", type=doctor.FileRegion()) +def region_to_table(filename: str, region: Tuple[str, int, int]): + entries, errors, options_map = loader.load_file(filename) + region_entries = doctor.resolve_region_to_entries(entries, filename, region) + rows = entries_to_table( + [txn for txn in data.filter_txns(region_entries) if is_error(txn)] + ) + petl.wrap(rows).tocsv(petl.StdoutSource()) + + +def is_error(txn: data.Transaction) -> bool: + return txn.flag == flags.FLAG_WARNING or any( + posting.flag == flags.FLAG_WARNING for posting in txn.postings + ) + + +def entries_to_table(txns: Iterable[data.Directives]) -> List[str]: + header = ["date", "flag", "payee", "narration", "links", "pflag", "account", "number", "currency"] + return chain( + [header], + [ + ( + txn.date, + txn.flag, + txn.payee, + txn.narration, + " ".join(sorted(txn.links)), + txn.postings[0].flag or "", + txn.postings[0].account, + txn.postings[0].units.number, + txn.postings[0].units.currency, + ) + for txn in txns + ], + ) + + +if __name__ == "__main__": + region_to_table()