refactor(bank-integration): cleanup + robust amount parsing
- Delete mapping_resolver.py: dead code (nothing imports it; create_reconcile
builds the indexes it needs directly).
- Simplify _create_journal_entry_for_brt: the if is_pay / else branches were
identical because the matching transaction mapping is scoped by payment_type,
so paid_from / paid_to already encode direction (from = credit side, to =
debit side for both Pay and Receive). Removed the dead conditional, kept a
comment explaining why no swap is needed.
- Add _parse_amount helper in excel_parser: handles currency text in the cell
('AZN 250', '250 AZN'), comma decimals ('250,00'), mixed thousands+decimal
('1,200.50' / '1.200,50'), accounting-style brackets ('(45.50)' → -45.50),
and trailing minus ('250-'). Used for amount/debit/credit cells.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
e8e9359cb5
commit
5dd1296221
|
|
@ -645,20 +645,10 @@ def _create_journal_entry_for_brt(txn, paid_from, paid_to, bank_txn, multi_curre
|
|||
if is_multi:
|
||||
je.multi_currency = 1
|
||||
|
||||
if is_pay:
|
||||
from_row = {
|
||||
"account": paid_from,
|
||||
"exchange_rate": from_rate,
|
||||
"credit_in_account_currency": from_amount,
|
||||
"debit_in_account_currency": 0,
|
||||
}
|
||||
to_row = {
|
||||
"account": paid_to,
|
||||
"exchange_rate": to_rate,
|
||||
"debit_in_account_currency": to_amount,
|
||||
"credit_in_account_currency": 0,
|
||||
}
|
||||
else:
|
||||
# paid_from / paid_to come from the matching transaction mapping (which is
|
||||
# scoped by payment_type), so direction is already baked in: paid_from is
|
||||
# always the credit side, paid_to is always the debit side. No swap needed
|
||||
# between Pay and Receive at this layer.
|
||||
from_row = {
|
||||
"account": paid_from,
|
||||
"exchange_rate": from_rate,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""Parse Excel bank statements using Bank Integration Excel Preset configuration."""
|
||||
|
||||
import re
|
||||
from datetime import datetime, date
|
||||
|
||||
import frappe
|
||||
|
|
@ -160,6 +161,68 @@ def _split_aliases(raw):
|
|||
return {part.strip().upper() for part in str(raw).split(",") if part.strip()}
|
||||
|
||||
|
||||
_AMOUNT_STRIP_RE = re.compile(r"[^\d,.\-]")
|
||||
|
||||
|
||||
def _parse_amount(cell):
|
||||
"""Convert a money-like cell value into a float.
|
||||
|
||||
Handles:
|
||||
- numeric cells (int/float) directly
|
||||
- currency text prefix/suffix: 'AZN 250.00', '250.00 AZN', 'USD 1,200.50'
|
||||
- comma decimal separator: '250,00' → 250.00
|
||||
- thousands + decimal: '1,200.50' / '1.200,50'
|
||||
- accounting-style negative in brackets: '(45.50)' → -45.50
|
||||
|
||||
Returns float, or None if the cell is empty / unparseable.
|
||||
"""
|
||||
if cell is None:
|
||||
return None
|
||||
if isinstance(cell, bool):
|
||||
return None
|
||||
if isinstance(cell, (int, float)):
|
||||
return float(cell)
|
||||
s = str(cell).strip()
|
||||
if not s:
|
||||
return None
|
||||
|
||||
negate = False
|
||||
if s.startswith("(") and s.endswith(")"):
|
||||
negate = True
|
||||
s = s[1:-1].strip()
|
||||
if s.endswith("-"):
|
||||
negate = not negate
|
||||
s = s[:-1].strip()
|
||||
|
||||
# Drop currency text and stray symbols, keep digits, separators, minus.
|
||||
s = _AMOUNT_STRIP_RE.sub("", s)
|
||||
if s.count("-") > 1:
|
||||
s = s.replace("-", "")
|
||||
negate = not negate
|
||||
elif s.startswith("-"):
|
||||
negate = not negate
|
||||
s = s[1:]
|
||||
|
||||
# Decimal separator detection. If both ',' and '.' appear, the rightmost is
|
||||
# the decimal one and the other is a thousands grouping. If only ',' appears,
|
||||
# assume it's the decimal separator.
|
||||
if "," in s and "." in s:
|
||||
if s.rfind(",") > s.rfind("."):
|
||||
s = s.replace(".", "").replace(",", ".")
|
||||
else:
|
||||
s = s.replace(",", "")
|
||||
elif "," in s:
|
||||
s = s.replace(",", ".")
|
||||
|
||||
if not s or s in (".", "-"):
|
||||
return None
|
||||
try:
|
||||
v = float(s)
|
||||
except ValueError:
|
||||
return None
|
||||
return -v if negate else v
|
||||
|
||||
|
||||
def _normalize_header(value):
|
||||
if value is None:
|
||||
return ""
|
||||
|
|
@ -212,11 +275,11 @@ def _parse_row(raw_row, idx_to_standard, custom_format, amount_mode,
|
|||
elif std == "counterparty_iban":
|
||||
record["contr_iban"] = str(cell).strip()
|
||||
elif std == "amount":
|
||||
amount_val = flt(cell)
|
||||
amount_val = _parse_amount(cell)
|
||||
elif std == "debit":
|
||||
debit_val = flt(cell)
|
||||
debit_val = _parse_amount(cell)
|
||||
elif std == "credit":
|
||||
credit_val = flt(cell)
|
||||
credit_val = _parse_amount(cell)
|
||||
elif std == "direction":
|
||||
direction_val = str(cell).strip().upper()
|
||||
elif std == "currency":
|
||||
|
|
|
|||
|
|
@ -1,194 +0,0 @@
|
|||
"""Resolve a unified mappings dict for a given Bank Account.
|
||||
|
||||
Reads either a Bank Integration record or Kapital Bank Settings (Single)
|
||||
based on the BA's hidden bank_integration_type / bank_integration fields,
|
||||
and returns a uniform shape consumable by create_reconcile.
|
||||
"""
|
||||
|
||||
import frappe
|
||||
from frappe.utils import flt
|
||||
|
||||
|
||||
def resolve_mappings_for_bank_account(bank_account_name):
|
||||
"""Returns the unified mappings dict, or None if BA is not configured."""
|
||||
ba = frappe.get_doc("Bank Account", bank_account_name)
|
||||
bi_type = ba.get("bank_integration_type")
|
||||
bi_name = ba.get("bank_integration")
|
||||
|
||||
if bi_type == "Bank Integration" and bi_name:
|
||||
if not frappe.db.exists("Bank Integration", bi_name):
|
||||
return None
|
||||
bi = frappe.get_doc("Bank Integration", bi_name)
|
||||
return _resolve_bank_integration(bi)
|
||||
|
||||
if bi_type == "Kapital Bank Settings":
|
||||
if not frappe.db.exists("DocType", "Kapital Bank Settings"):
|
||||
return None
|
||||
try:
|
||||
kb = frappe.get_single("Kapital Bank Settings")
|
||||
except Exception:
|
||||
return None
|
||||
return _resolve_kb_settings(kb)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _resolve_bank_integration(bi):
|
||||
purpose_rules = []
|
||||
for row in bi.transaction_mappings:
|
||||
if not row.paid_from or not row.paid_to:
|
||||
continue
|
||||
keyword_text = ""
|
||||
if row.purpose_keyword:
|
||||
keyword_text = frappe.db.get_value(
|
||||
"Bank Integration Purpose", row.purpose_keyword, "purpose_keyword"
|
||||
) or ""
|
||||
cp_name = ""
|
||||
cp_voen = ""
|
||||
if row.counterparty and row.counterparty_type:
|
||||
source_doctype = (
|
||||
"Bank Integration Customer" if row.counterparty_type == "Customer"
|
||||
else "Bank Integration Supplier"
|
||||
)
|
||||
name_field = "customer_name" if row.counterparty_type == "Customer" else "supplier_name"
|
||||
cp_data = frappe.db.get_value(source_doctype, row.counterparty, [name_field, "tax_id"], as_dict=True)
|
||||
if cp_data:
|
||||
cp_name = (cp_data.get(name_field) or "").strip()
|
||||
cp_voen = (cp_data.get("tax_id") or "").strip()
|
||||
|
||||
purpose_rules.append({
|
||||
"purpose_keyword": keyword_text.lower().strip(),
|
||||
"payment_type": row.payment_type or "",
|
||||
"counterparty_name": cp_name,
|
||||
"counterparty_voen": cp_voen,
|
||||
"counterparty_type": row.counterparty_type or "",
|
||||
"paid_from": row.paid_from,
|
||||
"paid_to": row.paid_to,
|
||||
"cost_center": row.cost_center or "",
|
||||
"multi_currency": bool(row.multi_currency),
|
||||
"currency": (row.currency or "").upper(),
|
||||
"document_type": row.document_type or "Payment Entry",
|
||||
})
|
||||
|
||||
voen_to_party, name_to_party = _build_party_indexes_bi(bi)
|
||||
|
||||
return {
|
||||
"source": "Bank Integration",
|
||||
"name": bi.name,
|
||||
"company": bi.default_company,
|
||||
"purpose_rules": purpose_rules,
|
||||
"voen_to_party": voen_to_party,
|
||||
"name_to_party": name_to_party,
|
||||
"similarity_threshold_purpose": flt(bi.similarity_threshold_purpose or 70) / 100.0,
|
||||
"consider_azeri_chars": bool(bi.consider_azeri_chars),
|
||||
}
|
||||
|
||||
|
||||
def _resolve_kb_settings(kb):
|
||||
purpose_rules = []
|
||||
for row in kb.transaction_mappings:
|
||||
if not row.paid_from or not row.paid_to:
|
||||
continue
|
||||
keyword_text = ""
|
||||
if row.purpose_keyword:
|
||||
keyword_text = frappe.db.get_value(
|
||||
"Kapital Bank Purpose", row.purpose_keyword, "purpose_keyword"
|
||||
) or ""
|
||||
cp_name = ""
|
||||
cp_voen = ""
|
||||
if row.counterparty and row.counterparty_type:
|
||||
source_doctype = row.counterparty_type
|
||||
name_field = "customer_name" if "Customer" in source_doctype else "supplier_name"
|
||||
cp_data = frappe.db.get_value(source_doctype, row.counterparty, [name_field, "tax_id"], as_dict=True)
|
||||
if cp_data:
|
||||
cp_name = (cp_data.get(name_field) or "").strip()
|
||||
cp_voen = (cp_data.get("tax_id") or "").strip()
|
||||
|
||||
cp_type_normalized = ""
|
||||
if row.counterparty_type:
|
||||
cp_type_normalized = "Customer" if "Customer" in row.counterparty_type else "Supplier"
|
||||
|
||||
purpose_rules.append({
|
||||
"purpose_keyword": keyword_text.lower().strip(),
|
||||
"payment_type": row.payment_type or "",
|
||||
"counterparty_name": cp_name,
|
||||
"counterparty_voen": cp_voen,
|
||||
"counterparty_type": cp_type_normalized,
|
||||
"paid_from": row.paid_from,
|
||||
"paid_to": row.paid_to,
|
||||
"cost_center": row.cost_center or "",
|
||||
"multi_currency": bool(getattr(row, "multi_currency", 0)),
|
||||
"currency": (getattr(row, "currency", None) or "").upper(),
|
||||
"document_type": getattr(row, "document_type", None) or "Payment Entry",
|
||||
})
|
||||
|
||||
voen_to_party, name_to_party = _build_party_indexes_kb(kb)
|
||||
|
||||
return {
|
||||
"source": "Kapital Bank Settings",
|
||||
"name": "Kapital Bank Settings",
|
||||
"company": kb.default_company,
|
||||
"purpose_rules": purpose_rules,
|
||||
"voen_to_party": voen_to_party,
|
||||
"name_to_party": name_to_party,
|
||||
"similarity_threshold_purpose": flt(kb.similarity_threshold_purpose or 70) / 100.0,
|
||||
"consider_azeri_chars": bool(kb.consider_azeri_chars),
|
||||
}
|
||||
|
||||
|
||||
def _build_party_indexes_bi(bi):
|
||||
voen_to_party = {}
|
||||
name_to_party = {}
|
||||
|
||||
for row in bi.customer_mappings:
|
||||
if not row.erp_customer:
|
||||
continue
|
||||
entry = {"party_type": "Customer", "erp_party": row.erp_customer}
|
||||
if row.tax_id:
|
||||
voen_to_party.setdefault(row.tax_id.strip(), {})["Customer"] = entry
|
||||
if row.bi_customer_name:
|
||||
cust_name = frappe.db.get_value("Bank Integration Customer", row.bi_customer_name, "customer_name")
|
||||
if cust_name:
|
||||
name_to_party.setdefault(cust_name.strip(), {})["Customer"] = entry
|
||||
|
||||
for row in bi.supplier_mappings:
|
||||
if not row.erp_supplier:
|
||||
continue
|
||||
entry = {"party_type": "Supplier", "erp_party": row.erp_supplier}
|
||||
if row.tax_id:
|
||||
voen_to_party.setdefault(row.tax_id.strip(), {})["Supplier"] = entry
|
||||
if row.bi_supplier_name:
|
||||
supp_name = frappe.db.get_value("Bank Integration Supplier", row.bi_supplier_name, "supplier_name")
|
||||
if supp_name:
|
||||
name_to_party.setdefault(supp_name.strip(), {})["Supplier"] = entry
|
||||
|
||||
return voen_to_party, name_to_party
|
||||
|
||||
|
||||
def _build_party_indexes_kb(kb):
|
||||
voen_to_party = {}
|
||||
name_to_party = {}
|
||||
|
||||
for row in kb.customer_mappings:
|
||||
if not row.erp_customer:
|
||||
continue
|
||||
entry = {"party_type": "Customer", "erp_party": row.erp_customer}
|
||||
if row.tax_id:
|
||||
voen_to_party.setdefault(row.tax_id.strip(), {})["Customer"] = entry
|
||||
if row.kb_customer_name:
|
||||
cust_name = frappe.db.get_value("Kapital Bank Customer", row.kb_customer_name, "customer_name")
|
||||
if cust_name:
|
||||
name_to_party.setdefault(cust_name.strip(), {})["Customer"] = entry
|
||||
|
||||
for row in kb.supplier_mappings:
|
||||
if not row.erp_supplier:
|
||||
continue
|
||||
entry = {"party_type": "Supplier", "erp_party": row.erp_supplier}
|
||||
if row.tax_id:
|
||||
voen_to_party.setdefault(row.tax_id.strip(), {})["Supplier"] = entry
|
||||
if row.kb_supplier_name:
|
||||
supp_name = frappe.db.get_value("Kapital Bank Supplier", row.kb_supplier_name, "supplier_name")
|
||||
if supp_name:
|
||||
name_to_party.setdefault(supp_name.strip(), {})["Supplier"] = entry
|
||||
|
||||
return voen_to_party, name_to_party
|
||||
Loading…
Reference in New Issue