import json import logging import os import sys from pathlib import Path from typing import Any, Dict, List, Optional from pydantic import ValidationError from toldg.models import Config, Mapping, Transaction def get_files(directory: Path, ending="") -> List[Path]: """Gets files from directory recursively in lexigraphic order.""" return [ Path(os.path.join(subdir, f)) for subdir, _, files in os.walk(directory) for f in files if f.endswith(ending) ] def get_csv_files(directory: Path) -> List[Path]: return get_files(directory, ".csv") def get_ldg_files(directory: Path) -> List[Path]: return get_files(directory, ".ldg") def load_config() -> Config: try: config_file = Path(sys.argv[1]) except IndexError: logging.critical("Provide configuration file as first argument.") sys.exit(1) try: with open(config_file, "r") as f: config = Config(**json.load(f)) except ValidationError as e: logging.critical(f"Could not validate {config_file}.") logging.info(e) sys.exit(1) except FileNotFoundError: logging.critical(f"Could not find {config_file}.") sys.exit(1) return config def category_to_bean(c: str) -> str: sections = map(list, c.split(":")) new_sections = [] for section in sections: section[0] = section[0].upper() new_sections.append("".join(section)) return ":".join(new_sections) def write_meta(config: Config): with open(config.output_file, "a") as f: for category in config.categories: f.write(f"2017-01-01 open {category_to_bean(category)}\n") f.write("\n") f.write('option "operating_currency" "USD"\n\n') def write_mappings(transactions: List[Transaction], mappings_file: Path): """Write transactions to the mappings file.""" mappings = read_mappings(mappings_file) for t in transactions: if t.key() in mappings: pass else: mapping = Mapping( **{ "account2": t.account2.strip(), "narration": t.description, } ) mappings[t.key()] = mapping mappings = {k: v.model_dump(exclude_none=True) for k, v in mappings.items()} with open(mappings_file, "w") as f: json.dump(mappings, f, indent=2) def read_mappings(mappings_file: Path) -> Dict[str, Mapping]: """Read mappings from file.""" with open(mappings_file, "r") as f: data = json.load(f) for key, value in data.items(): data[key] = Mapping(**value) return data def remove_if_exists(output_file: Path): try: os.remove(output_file) except OSError: pass