198 lines
8 KiB
Python
198 lines
8 KiB
Python
#!/usr/bin/env python3
|
|
# AUTHOR: Daesorin
|
|
# CREATED: 2026-07-27
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
# CONFIGURATION
|
|
# set default accounting buckets, fallback currency, and journal vault path
|
|
DEFAULT_EXPENSE_ACCOUNT = "Expenses:Uncategorised"
|
|
DEFAULT_INCOME_ACCOUNT = "Income:Uncategorised"
|
|
DEFAULT_PAYMENT_ACCOUNT = "Assets:Banks:OPay"
|
|
DEFAULT_CURRENCY = "NGN"
|
|
NOTES_DIR = Path("~/obsidian/finances").expanduser()
|
|
|
|
# ACCOUNT NORMALISATION
|
|
# intercept legacy account strings and fix misrouted categories at compile time
|
|
def normalize_account(account_str, currency, name):
|
|
clean_account = account_str.strip()
|
|
clean_name = name.lower()
|
|
|
|
# auto resolve generic raenest root postings to multi currency leaf accounts
|
|
if clean_account == "Assets:Banks:Raenest":
|
|
if currency == "USD":
|
|
return "Assets:Banks:Raenest:USD"
|
|
if currency == "GBP":
|
|
return "Assets:Banks:Raenest:GBP"
|
|
if currency == "EUR":
|
|
return "Assets:Banks:Raenest:EUR"
|
|
return "Assets:Banks:Raenest:NGN"
|
|
|
|
# intercept fee and charge keywords before portfolio redirects
|
|
if any(w in clean_name for w in ("charge", "fee", "loss", "tax", "commission")):
|
|
if "uncategorised" in clean_account.lower():
|
|
return "Expenses:Financial:BankLoss"
|
|
return clean_account
|
|
|
|
# redirect uncategorised investment transfers to asset portfolios
|
|
if "uncategorised" in clean_account.lower():
|
|
if "cowrywise" in clean_name:
|
|
return "Assets:Investments:Cowrywise"
|
|
if "bybit" in clean_name:
|
|
return "Assets:Investments:Bybit"
|
|
if "octafx" in clean_name:
|
|
return "Assets:Investments:OctaFX"
|
|
|
|
return clean_account
|
|
|
|
# FRONTMATTER PARSER
|
|
# extract yaml between the opening and closing dashes without external libraries
|
|
def parse_frontmatter(file_path):
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
except OSError as e:
|
|
print(f"warning: unable to read {file_path} ({e})", file=sys.stderr)
|
|
return None
|
|
|
|
if not content.startswith("---"):
|
|
return None
|
|
|
|
parts = content.split("---", 2)
|
|
if len(parts) < 3:
|
|
return None
|
|
|
|
yaml_text = parts[1]
|
|
data = {}
|
|
for line in yaml_text.strip().split("\n"):
|
|
# ignore empty lines and full line comments
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
|
|
if ":" in line:
|
|
key, val = line.split(":", 1)
|
|
# strip inline yaml comments and surrounding quotation marks
|
|
clean_val = val.split("#", 1)[0].strip().strip('\'"')
|
|
data[key.strip().lower()] = clean_val
|
|
|
|
return data
|
|
|
|
# JOURNAL GENERATOR
|
|
# convert parsed dictionary into an hledger double entry transaction string
|
|
def generate_entry(data, file_path):
|
|
try:
|
|
date = data["date"]
|
|
name = data.get("name", "untitled transaction")
|
|
tx_type = data.get("type", "expense").lower()
|
|
|
|
# read dynamic accounts or fallback to defaults
|
|
currency = data.get("currency", DEFAULT_CURRENCY).strip().upper()
|
|
raw_payment = data.get("payment_account", DEFAULT_PAYMENT_ACCOUNT)
|
|
raw_target = data.get("account", DEFAULT_EXPENSE_ACCOUNT if tx_type not in {"income", "liability", "borrowing", "inflow"} else DEFAULT_INCOME_ACCOUNT)
|
|
|
|
# normalise accounts against currency and description keywords
|
|
payment_account = normalize_account(raw_payment, currency, name)
|
|
target_account = normalize_account(raw_target, currency, name)
|
|
|
|
# use decimal for precise financial arithmetic
|
|
unit = Decimal(data.get("unit", "1"))
|
|
amount = Decimal(data["amount"])
|
|
total = unit * amount
|
|
|
|
# CHARGE RESOLUTION
|
|
# extract bank transfer fees and pos agent charges without multiplying by unit
|
|
transfer_charges = Decimal(data.get("transfer_charges", "0"))
|
|
pos_charges = Decimal(data.get("pos_charges", "0"))
|
|
|
|
# FOREX CONVERSIONS
|
|
# handle dual currency exchanges and correct backwards conversion frontmatter
|
|
if tx_type == "conversion":
|
|
source_currency = data.get("source_currency", "USD").strip().upper()
|
|
source_amount = Decimal(data.get("source_amount", "0")) * unit
|
|
total_source_outflow = source_amount
|
|
|
|
charge_postings = ""
|
|
if pos_charges > 0:
|
|
charge_postings += f" Expenses:Financial:POSFees {pos_charges:.2f} {source_currency}\n"
|
|
total_source_outflow += pos_charges
|
|
if transfer_charges > 0:
|
|
charge_postings += f" Expenses:Financial:BankLoss {transfer_charges:.2f} {source_currency}\n"
|
|
total_source_outflow += transfer_charges
|
|
|
|
# fix reversed raenest conversion where foreign currency was logged as spent instead of local currency
|
|
if ("raenest" in name.lower() or "raenest" in payment_account.lower()) and source_currency != DEFAULT_CURRENCY:
|
|
entry = (
|
|
f"{date} * {name}\n"
|
|
f" {payment_account} {source_amount:.2f} {source_currency}\n"
|
|
f" {target_account} -{total:.2f} {currency} @@ {source_amount:.2f} {source_currency}\n"
|
|
f"{charge_postings}\n"
|
|
)
|
|
return entry
|
|
|
|
entry = (
|
|
f"{date} * {name}\n"
|
|
f" {target_account} {total:.2f} {currency}\n"
|
|
f"{charge_postings}"
|
|
f" {payment_account} -{total_source_outflow:.2f} {source_currency} @@ {total:.2f} {currency}\n\n"
|
|
)
|
|
return entry
|
|
|
|
# CASH FLOW ROUTING
|
|
# route transactions based on whether value enters or leaves the payment account
|
|
inflow_types = {"income", "liability", "borrowing"}
|
|
|
|
# handle loans collected and standard inflows unless routed from equity windfalls
|
|
if tx_type == "inflow" and not payment_account.lower().startswith("equity"):
|
|
inflow_types.add("inflow")
|
|
|
|
if tx_type in inflow_types:
|
|
entry = (
|
|
f"{date} * {name}\n"
|
|
f" {payment_account} {currency} {total:.2f}\n"
|
|
f" {target_account} -{currency} {total:.2f}\n\n"
|
|
)
|
|
else:
|
|
charge_postings = ""
|
|
total_outflow = total
|
|
|
|
if pos_charges > 0:
|
|
charge_postings += f" Expenses:Financial:POSFees {currency} {pos_charges:.2f}\n"
|
|
total_outflow += pos_charges
|
|
|
|
if transfer_charges > 0:
|
|
charge_postings += f" Expenses:Financial:BankLoss {currency} {transfer_charges:.2f}\n"
|
|
total_outflow += transfer_charges
|
|
|
|
entry = (
|
|
f"{date} * {name}\n"
|
|
f" {target_account} {currency} {total:.2f}\n"
|
|
f"{charge_postings}"
|
|
f" {payment_account} -{currency} {total_outflow:.2f}\n\n"
|
|
)
|
|
|
|
return entry
|
|
except (KeyError, InvalidOperation, TypeError) as e:
|
|
# log malformed notes to standard error to prevent silent data loss
|
|
print(f"warning: skipping malformed note '{file_path}' ({type(e).__name__})", file=sys.stderr)
|
|
return ""
|
|
|
|
# MAIN EXECUTION
|
|
# scan the journal directory recursively and print compiled journal entries
|
|
if __name__ == "__main__":
|
|
if not NOTES_DIR.is_dir():
|
|
print(f"error: directory '{NOTES_DIR}' does not exist", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
entries = []
|
|
# rglob enables recursive scanning through nested folder structures
|
|
for file_path in sorted(NOTES_DIR.rglob("*.md")):
|
|
data = parse_frontmatter(file_path)
|
|
if data and "date" in data and "amount" in data:
|
|
entry = generate_entry(data, file_path)
|
|
if entry:
|
|
entries.append(entry)
|
|
|
|
print("".join(entries), end="")
|