impl/python.py

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import List, Optional, Sequence

from .dates_days_between import days_between  ← from dates.days-between ^1.0.0 · built alongside by fune
from .finance_bank_reconcile_match_types import ReconcileLine, ReconcileMatch, Reconciliation
from .math_round_div import round_div  ← from math.round-div ^1.0.0 · built alongside by fune


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)