from typing import List, Optional, Sequence from .dates_days_between import days_between from .finance_bank_reconcile_match_types import ReconcileLine, ReconcileMatch, Reconciliation from .math_round_div import round_div def _reduce(reference: str) -> str: out = [] for ch in reference: if "0" <= ch <= "9" or "A" <= ch <= "Z": out.append(ch) elif "a" <= ch <= "z": out.append(chr(ord(ch) - 32)) return "".join(out) def _bigrams(text: str) -> List[str]: return [text[i : i + 2] for i in range(len(text) - 1)] def _similarity(a: str, b: str) -> int: """Sørensen-Dice on adjacent character pairs, 0 to 10000, rounded down.""" x = _reduce(a) y = _reduce(b) if len(x) > 0 and x == y: return 10000 left = _bigrams(x) right = _bigrams(y) if len(left) == 0 or len(right) == 0: return 0 used = [False] * len(right) shared = 0 for pair in left: for i, other in enumerate(right): if not used[i] and other == pair: used[i] = True shared += 1 break return round_div(2 * shared * 10000, len(left) + len(right), "down") def _check_ids(lines: Sequence[ReconcileLine], side: str) -> None: seen = set() for line in lines: if line.id in seen: raise ValueError('duplicate %s id "%s"' % (side, line.id)) seen.add(line.id) def match_bank_lines( bank_lines: Sequence[ReconcileLine], ledger_entries: Sequence[ReconcileLine], date_window_days: int, min_similarity: int, ) -> Reconciliation: """Pair bank lines with ledger entries: equal amount, dates within the window, references alike enough, best pairs first and ties broken by input order. """ if isinstance(date_window_days, bool) or not isinstance(date_window_days, int) or date_window_days < 0: raise ValueError("dateWindowDays must be 0 or more, received %s" % (date_window_days,)) if ( isinstance(min_similarity, bool) or not isinstance(min_similarity, int) or min_similarity < 0 or min_similarity > 10000 ): raise ValueError("minSimilarity must be 0 to 10000, received %s" % (min_similarity,)) _check_ids(bank_lines, "bank line") _check_ids(ledger_entries, "ledger entry") candidates = [] for b, bank in enumerate(bank_lines): for l, entry in enumerate(ledger_entries): days = days_between(entry.date, bank.date) if bank.amount.currency != entry.amount.currency or bank.amount.minor != entry.amount.minor: continue if abs(days) > date_window_days: continue score = _similarity(bank.reference, entry.reference) if score < min_similarity: continue candidates.append((-score, abs(days), b, l, days, score)) candidates.sort() bank_match: List[Optional[ReconcileMatch]] = [None] * len(bank_lines) ledger_used = [False] * len(ledger_entries) for _, _, b, l, days, score in candidates: if bank_match[b] is not None or ledger_used[l]: continue bank_match[b] = ReconcileMatch( bank_id=bank_lines[b].id, ledger_id=ledger_entries[l].id, day_difference=days, similarity=score ) ledger_used[l] = True matches = [m for m in bank_match if m is not None] unmatched_bank = [bank.id for b, bank in enumerate(bank_lines) if bank_match[b] is None] unmatched_ledger = [entry.id for l, entry in enumerate(ledger_entries) if not ledger_used[l]] return Reconciliation(matches=matches, unmatched_bank=unmatched_bank, unmatched_ledger=unmatched_ledger)