generated from felixm/defaultpy
93 lines
2.4 KiB
Python
93 lines
2.4 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
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_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
|