Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 95 additions & 0 deletions bin/file-to-account
Original file line number Diff line number Diff line change
@@ -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()
76 changes: 76 additions & 0 deletions bin/region-errors-to-table
Original file line number Diff line number Diff line change
@@ -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()