From e589dcdaf44adb5f5c5840a7e99a339ea67fde37 Mon Sep 17 00:00:00 2001 From: Ali <010109ali@gmail.com> Date: Tue, 12 May 2026 12:22:47 +0000 Subject: [PATCH] feat: fuzzy purpose matching in BRT Create & Reconcile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _find_mapping_for_txn now tries exact-substring purpose matches first (unchanged), then falls back to a partial-ratio fuzzy match against similarity_threshold_purpose — so a short/imprecise purpose keyword can match somewhere inside a long statement description. Order: purpose+ counterparty (exact) > purpose-only (exact) > purpose+counterparty (fuzzy) > purpose-only (fuzzy) > counterparty-only > fallback. Set the threshold to 100 to keep the old substring-only behaviour. This makes similarity_threshold_purpose meaningful in the BRT path (previously it only affected the API-driven import). Co-Authored-By: Claude Opus 4.7 (1M context) --- kapital_bank/mapping.py | 77 ++++++++++++++++++++++++++++++++++------- 1 file changed, 65 insertions(+), 12 deletions(-) diff --git a/kapital_bank/mapping.py b/kapital_bank/mapping.py index 03ee90c..09c697c 100644 --- a/kapital_bank/mapping.py +++ b/kapital_bank/mapping.py @@ -1,9 +1,31 @@ import json +from difflib import SequenceMatcher import frappe from frappe.utils import cint, flt +def _partial_ratio(needle, haystack): + """Best fuzzy ratio of `needle` against any equal-length window of `haystack` + (like fuzzywuzzy.partial_ratio) — lets a short purpose keyword fuzzily match + somewhere inside a long statement description. Inputs lowercased/stripped.""" + needle = (needle or "").strip() + haystack = (haystack or "").strip() + if not needle or not haystack: + return 0.0 + if len(needle) >= len(haystack): + return SequenceMatcher(None, needle, haystack).ratio() + best = 0.0 + span = len(needle) + for i in range(0, len(haystack) - span + 1): + r = SequenceMatcher(None, needle, haystack[i:i + span]).ratio() + if r > best: + best = r + if best >= 0.9999: + break + return best + + def _extract_messages(message_log): """Extract plain text from frappe.local.message_log entries.""" parts = [] @@ -145,6 +167,7 @@ def create_purpose_mappings(transactions, paid_from=None, paid_to=None, document if do_documents: txn_mappings = settings.transaction_mappings if mode == "Documents & Reconcile" else None + purpose_threshold = flt(settings.similarity_threshold_purpose or 70) / 100.0 if mode == "Documents & Reconcile": purpose_text_cache = {} @@ -164,7 +187,7 @@ def create_purpose_mappings(transactions, paid_from=None, paid_to=None, document continue if mode == "Documents & Reconcile": - mapping_row = _find_mapping_for_txn(txn, txn_mappings, purpose_text_cache) + mapping_row = _find_mapping_for_txn(txn, txn_mappings, purpose_text_cache, purpose_threshold) if not mapping_row: errors.append({ "reference_number": txn.get("reference_number") or "", @@ -267,11 +290,13 @@ def create_purpose_mappings(transactions, paid_from=None, paid_to=None, document return {"success": False, "message": str(e)} -def _find_mapping_for_txn(txn, transaction_mappings, purpose_text_cache): +def _find_mapping_for_txn(txn, transaction_mappings, purpose_text_cache, purpose_threshold): """Find the best matching transaction mapping row for a BRT txn. - Priority: rows matching both purpose+counterparty > purpose-only > counterparty-only. - Returns the first matching row or None. + Priority: purpose+counterparty (exact) > purpose-only (exact) > + purpose+counterparty (fuzzy) > purpose-only (fuzzy) > counterparty-only > + fallback by counterparty_type. "Exact" = the keyword is a substring of the + transaction's description; "fuzzy" = partial-ratio >= purpose_threshold. """ payment_type = "Pay" if txn.get("drcr") == "D" else "Receive" txn_purpose = (txn.get("purpose") or "").lower() @@ -310,26 +335,54 @@ def _find_mapping_for_txn(txn, transaction_mappings, purpose_text_cache): elif row.counterparty_type: fallback_rules.append(row) - def _purpose_matches(row): - keyword_text = purpose_text_cache.get(row.purpose_keyword, "") - return keyword_text and keyword_text in txn_purpose + def _kw(row): + return purpose_text_cache.get(row.purpose_keyword, "") + + def _purpose_exact(row): + kw = _kw(row) + return bool(kw) and kw in txn_purpose + + def _purpose_fuzzy(row): + kw = _kw(row) + return _partial_ratio(kw, txn_purpose) if kw else 0.0 def _counterparty_matches(row): return row.counterparty.lower() == txn_party + def _best_fuzzy(rules, check_cp): + best = None + best_score = 0.0 + for row in rules: + if check_cp and not _counterparty_matches(row): + continue + score = _purpose_fuzzy(row) + if score >= purpose_threshold and score > best_score: + best_score = score + best = row + return best + + # 1. purpose + counterparty — exact for row in both_rules: - if _purpose_matches(row) and _counterparty_matches(row): + if _purpose_exact(row) and _counterparty_matches(row): return row - + # 2. purpose-only — exact for row in purpose_only_rules: - if _purpose_matches(row): + if _purpose_exact(row): return row - + # 3. purpose + counterparty — fuzzy + hit = _best_fuzzy(both_rules, check_cp=True) + if hit: + return hit + # 4. purpose-only — fuzzy + hit = _best_fuzzy(purpose_only_rules, check_cp=False) + if hit: + return hit + # 5. counterparty-only for row in counterparty_only_rules: if _counterparty_matches(row): return row - # Phase 4: fallback by counterparty_type + # 6. fallback by counterparty_type inferred_cp_type = "Customer" if payment_type == "Receive" else "Supplier" for row in fallback_rules: if row.counterparty_type == inferred_cp_type: