feat(bank-integration): fuzzy purpose matching in BRT Create & Reconcile

_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 bank-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 actually do something
in the BRT path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ali 2026-05-12 12:22:40 +00:00
parent cd162c1f89
commit 42cac5a006
1 changed files with 65 additions and 11 deletions

View File

@ -9,6 +9,7 @@ Bank Integration source. Three modes: "Both" / "Mappings Only" /
""" """
import json import json
from difflib import SequenceMatcher
import frappe import frappe
from frappe.utils import cint, flt from frappe.utils import cint, flt
@ -16,6 +17,27 @@ from frappe.utils import cint, flt
PURPOSE_DOCTYPE = "Bank Integration Purpose" PURPOSE_DOCTYPE = "Bank Integration Purpose"
def _partial_ratio(needle, haystack):
"""Best fuzzy ratio of `needle` against any equal-length window of `haystack`
(like fuzzywuzzy.partial_ratio). Lets a short keyword fuzzily match somewhere
inside a long bank-statement description. Inputs should be 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): def _extract_messages(message_log):
parts = [] parts = []
for m in message_log: for m in message_log:
@ -150,6 +172,7 @@ def create_purpose_mappings(transactions, bank_integration, paid_from=None, paid
frappe.db.commit() frappe.db.commit()
if do_documents: if do_documents:
purpose_threshold = flt(settings.similarity_threshold_purpose or 70) / 100.0
if mode == "Documents & Reconcile": if mode == "Documents & Reconcile":
purpose_text_cache = {} purpose_text_cache = {}
for row in settings.transaction_mappings: for row in settings.transaction_mappings:
@ -170,7 +193,7 @@ def create_purpose_mappings(transactions, bank_integration, paid_from=None, paid
continue continue
if mode == "Documents & Reconcile": 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: if not mapping_row:
errors.append({ errors.append({
"reference_number": txn.get("reference_number") or "", "reference_number": txn.get("reference_number") or "",
@ -272,10 +295,13 @@ def create_purpose_mappings(transactions, bank_integration, paid_from=None, paid
return {"success": False, "message": str(e)} 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. """Find the best matching transaction mapping row for a BRT txn.
Priority: rows matching both purpose+counterparty > purpose-only > counterparty-only > fallback. 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" payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
txn_purpose = (txn.get("purpose") or "").lower() txn_purpose = (txn.get("purpose") or "").lower()
@ -310,25 +336,53 @@ def _find_mapping_for_txn(txn, transaction_mappings, purpose_text_cache):
elif row.counterparty_type: elif row.counterparty_type:
fallback_rules.append(row) fallback_rules.append(row)
def _purpose_matches(row): def _kw(row):
keyword_text = purpose_text_cache.get(row.purpose_keyword, "") return purpose_text_cache.get(row.purpose_keyword, "")
return keyword_text and keyword_text in txn_purpose
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): def _counterparty_matches(row):
return row.counterparty.lower() == txn_party 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: for row in both_rules:
if _purpose_matches(row) and _counterparty_matches(row): if _purpose_exact(row) and _counterparty_matches(row):
return row return row
# 2. purpose-only — exact
for row in purpose_only_rules: for row in purpose_only_rules:
if _purpose_matches(row): if _purpose_exact(row):
return 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: for row in counterparty_only_rules:
if _counterparty_matches(row): if _counterparty_matches(row):
return row return row
# 6. fallback by counterparty_type
inferred_cp_type = "Customer" if payment_type == "Receive" else "Supplier" inferred_cp_type = "Customer" if payment_type == "Receive" else "Supplier"
for row in fallback_rules: for row in fallback_rules:
if row.counterparty_type == inferred_cp_type: if row.counterparty_type == inferred_cp_type: