felixm/ledgerpy
1
0
Fork 0
Code Pull Requests Releases Activity

Update readme. Refactor toldg a little. Provide example files for testing by the user.

Browse Source
This commit is contained in:
Felix Martin
2020-08-11 16:30:30 -04:00
parent 3138be8d17
commit 367c38592b
13 changed files with 337 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
toldg.py
View File

@@ -62,7 +62,7 @@ class CsvMapping:
@dataclass
class LdgTransaction:
class Transaction:
"""
Class for ledger transaction to render into ldg file.
"""
@@ -129,47 +129,64 @@ def get_mappings(mappings_directory: str) -> List[CsvMapping]:
for m in get_mappings_from_file(f)]
def get_transactions(csv_file, config: CsvConfig, mappings: List[CsvMapping]):
def date_to_date(date):
def get_transactions(csv_file: str, config: CsvConfig) -> List[Transaction]:
def date_to_date(date: str) -> str:
d = datetime.datetime.strptime(date, config.input_date_format)
return d.strftime(config.output_date_format)
def flip_sign(amount):
if amount.startswith("-"):
return amount[1:]
return "-" + amount
def flip_sign(amount: str) -> str:
return amount[1:] if amount.startswith("-") else "-" + amount
def row_to_transaction(row, fields):
""" The user can configure the mapping of CSV fields to the three
required fields date, amount and description via the CsvConfig. """
t = {field: row[index] for index, field in fields}
amount = t['amount']
return Transaction(config.currency, flip_sign(amount), amount,
date_to_date(t['date']), config.account1,
"account2", t['description'], csv_file, ", ".join(row))
fields = [(i, f) for i, f in enumerate(config.fields) if f]
with open(csv_file, 'r') as f:
reader = csv.reader(f, delimiter=config.delimiter,
quotechar=config.quotechar)
for _ in range(config.skip):
next(reader)
transactions = [row_to_transaction(row, fields)
for row in reader if row]
return transactions
def apply_mappings(transactions: List[Transaction], mappings: List[CsvMapping]):
def make_equal_len(str_1, str_2):
max_len = max(len(str_1), len(str_2))
str_1 += " " * (max_len - len(str_1))
str_2 += " " * (max_len - len(str_2))
return (str_1, str_2)
def get_account2(transaction):
def get_matching_mappings(transaction):
t = transaction
matching_mappings = []
for mapping in mappings:
pattern = mapping.description_pattern
if type(pattern) is str and pattern == transaction.description:
pass
elif type(pattern) is re.Pattern and pattern.match(t.description):
pass
else:
if type(pattern) is str and pattern != transaction.description:
continue
elif type(pattern) is re.Pattern and not pattern.match(t.description):
continue
specifiers_match = True
for attr, value in mapping.specifiers:
if getattr(t, attr) != value:
specifiers_match = False
if not specifiers_match:
continue
matching_mappings.append(mapping)
return matching_mappings
if specifiers_match:
matching_mappings.append(mapping)
def get_account2(transaction):
matching_mappings = get_matching_mappings(transaction)
if not matching_mappings:
logging.info(f"No match for {transaction}.")
e = f"expenses,{t.description},credit={t.credit};date={t.date}\n"
unmatched_expenses.append(e)
return "expenses"
return ""
elif len(matching_mappings) == 1:
return matching_mappings[0].account2
else:
@@ -179,38 +196,23 @@ def get_transactions(csv_file, config: CsvConfig, mappings: List[CsvMapping]):
logging.info(f" {m}")
return matching_mappings[0].account2
def row_to_transaction(row):
t = {field: row[index] for index, field in fields}
amount = t['amount']
t = LdgTransaction(config.currency, flip_sign(amount), amount,
date_to_date(t['date']), config.account1,
"", t['description'], csv_file, ", ".join(row))
t.account1, t.account2 = make_equal_len(t.account1, get_account2(t))
return t
fields = [(index, field)
for index, field in enumerate(config.fields) if field]
unmatched_expenses = []
with open(csv_file, 'r') as f:
reader = csv.reader(f, delimiter=config.delimiter,
quotechar=config.quotechar)
[next(reader) for _ in range(config.skip)]
transactions = [t
for row in reader
if row
if (t := row_to_transaction(row))
]
return transactions, unmatched_expenses
for t in transactions:
account2 = get_account2(t)
if not account2:
unmatched_expenses.append(t)
account2 = "expenses"
t.account1, t.account2 = make_equal_len(t.account1, account2)
return unmatched_expenses
def render_to_file(transactions, csv_file, ledger_file, template_file=""):
def render_to_file(transactions: List[Transaction], csv_file: str, ledger_file: str):
content = "".join([LEDGER_TRANSACTION_TEMPLATE.format(t=t)
for t in transactions])
status = "no change"
if not os.path.isfile(ledger_file):
with open(ledger_file, 'w') as f:
f.write(new_content)
f.write(content)
status = "new"
else:
with open(ledger_file, 'r') as f:
@@ -223,9 +225,25 @@ def render_to_file(transactions, csv_file, ledger_file, template_file=""):
logging.info(f"{csv_file:30} -> {ledger_file:30} | {status}")
def main(config):
def file_age(file):
return time.time() - os.path.getmtime(file)
def write_mappings(unmatched_transactions: List[Transaction], mappings_directory: str):
""" Write mappings for unmatched expenses for update by the user. """
if not unmatched_transactions:
return
fn = os.path.join(mappings_directory, "unmatched.csv")
with open(fn, 'a') as f:
writer = csv.writer(f)
for t in unmatched_transactions:
e = ["expenses", t.description,
f"credit={t.credit};date={t.date}"]
writer.writerow(e)
def process_csv_file(csv_file, mappings: List[CsvMapping], config: Config):
def csv_to_ldg_filename(csv_file: str, config: Config) -> str :
r = csv_file
r = r.replace(config.input_directory, config.output_directory)
r = r.replace(".csv", ".ldg")
return r
def get_csv_config(csv_file: str, csv_configs: List[CsvConfig]) -> CsvConfig:
cs = [c for c in csv_configs
@@ -236,41 +254,28 @@ def main(config):
raise Exception(f"More than one config for {csv_file=}.")
return cs[0]
def write_unmatched_expenses(unmatched_expenses, mappings_directory):
if not unmatched_expenses:
return
fn = os.path.join(mappings_directory, "unmatched.csv")
with open(fn, 'a') as f:
for e in unmatched_expenses:
f.write(e)
ledger_file = csv_to_ldg_filename(csv_file, config)
csv_config = get_csv_config(csv_file, config.csv_configs)
transactions = get_transactions(csv_file, csv_config)
unmatched_transactions = apply_mappings(transactions, mappings)
write_mappings(unmatched_transactions, config.mappings_directory)
render_to_file(transactions, csv_file, ledger_file)
def csv_to_ldg_filename(csv_file: str, config: Config):
r = csv_file
r = r.replace(config.input_directory, config.output_directory)
r = r.replace(".csv", ".ldg")
return r
def process_csv_file(csv_file, mappings: List[CsvMapping], config: Config):
ledger_file = csv_to_ldg_filename(csv_file, config)
csv_config = get_csv_config(csv_file, config.csv_configs)
def process_ldg_file(ldg_file: str, config: Config):
file_age = lambda file: time.time() - os.path.getmtime(file)
dest_file = ldg_file.replace(config.input_directory, config.output_directory)
status = "no change"
if not os.path.isfile(dest_file):
status = "new"
shutil.copy(ldg_file, dest_file)
if file_age(dest_file) > file_age(ldg_file):
shutil.copy(ldg_file, dest_file)
status = "update"
logging.info(f"{ldg_file:30} -> {dest_file:30} | {status}")
transactions, unmatched = get_transactions(
csv_file, csv_config, mappings)
write_unmatched_expenses(unmatched, config.mappings_directory)
render_to_file(transactions, csv_file, ledger_file)
def process_ldg_file(ldg_file: str, config: Config):
dest_file = ldg_file.replace(
config.input_directory, config.output_directory)
status = "no change"
if not os.path.isfile(dest_file):
status = "new"
shutil.copy(ldg_file, dest_file)
if file_age(dest_file) > file_age(ldg_file):
shutil.copy(ldg_file, dest_file)
status = "update"
logging.info(f"{ldg_file:30} -> {dest_file:30} | {status}")
def main(config):
input_files = get_files(config.input_directory)
config.csv_configs = [CsvConfig(**c) for c in config.csv_configs]
mappings = get_mappings(config.mappings_directory)
@@ -286,7 +291,7 @@ def main(config):
if __name__ == "__main__":
logging.basicConfig(stream=sys.stdout,
level=logging.DEBUG,
level=logging.INFO,
format='%(message)s')
try:
config_file = sys.argv[1]