pending shared expenses script

commands/pending-shared-expenses

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+from beancount import loader
+from beancount.query import query
+from beancount.core.data import Transaction
+from beancount.parser import printer
+import argparse
+from datetime import date, timedelta
+from tabulate import tabulate
+from collections import defaultdict
+
+
+class bcolors:
+    HEADER = '\033[95m'
+    OKBLUE = '\033[94m'
+    OKCYAN = '\033[96m'
+    OKGREEN = '\033[92m'
+    WARNING = '\033[93m'
+    FAIL = '\033[91m'
+    ENDC = '\033[0m'
+    BOLD = '\033[1m'
+    UNDERLINE = '\033[4m'
+
+
+def get_shared_expenses(entries, options, start_date, end_date):
+    expenses_query = f"""SELECT date, account, value(position) FROM
+date >= {start_date} AND
+date < {end_date.isoformat()}
+WHERE account ~ \"Expenses:Lloguer\" OR account ~ \"Expenses:FacturesUtilitats\""""
+    rtypes, rrows = query.run_query(
+        entries, options, expenses_query)
+    expenses = []
+    for row in rrows:
+        if row.value_position.number > 0:
+            expenses.append(row)
+    return expenses
+
+
+def get_paid_equities(entries, options, start_date, end_date):
+    expenses_query = f"""SELECT date, account, value(position) FROM
+date >= {start_date} AND
+date < {end_date.isoformat()}
+WHERE account ~ \"Equity:LloguerMiquel\" OR account ~ \"Equity:FacturesUtilitatsMiquel\""""
+    rtypes, rrows = query.run_query(
+        entries, options, expenses_query)
+    return rrows
+
+
+def find_matching_expense(expense, equities):
+    for equity in equities:
+        if equity.value_position == expense.value_position:
+            return True
+    return False
+
+
+def find_missing_shared_expenses(expenses, equities):
+    missing_shared_expenses = []
+    for expense in expenses:
+        if not find_matching_expense(expense, equities):
+            missing_shared_expenses.append(expense)
+    return missing_shared_expenses
+
+
+def print_report(missing_shared_expenses, start_date):
+    print("Missing shared expenses")
+    print(f"From {start_date}")
+    print(missing_shared_expenses)
+    report = {
+        "Date": [],
+        "Account": [],
+        "Amount": []
+    }
+    for expense in missing_shared_expenses:
+        report["Date"].append(expense.date)
+        report["Account"].append(expense.account)
+        report["Amount"].append(expense.value_position)
+    print(tabulate(report))
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description='Find pending shared expenses (checks last 12 months)')
+    args = parser.parse_args()
+
+    # Calculate date range: 1 year ago to today
+    end_date = date.today()
+    start_date = end_date - timedelta(days=365)
+
+    print(f"Checking period: {start_date.isoformat()} to {
+          end_date.isoformat()}")
+
+    filename = "ledger/main.beancount"
+    entries, errors, options = loader.load_file(filename)
+
+    if errors:
+        printer.print_errors(errors)
+        return
+
+    expenses = get_shared_expenses(entries, options, start_date, end_date)
+    equities = get_paid_equities(entries, options, start_date, end_date)
+
+    missing_shared_expenses = find_missing_shared_expenses(expenses, equities)
+
+    print_report(missing_shared_expenses, start_date)
+
+
+if __name__ == "__main__":
+    main()