794 lines
27 KiB
Python
794 lines
27 KiB
Python
import json
|
|
from difflib import SequenceMatcher
|
|
|
|
import frappe
|
|
from frappe.utils import cint, flt
|
|
|
|
|
|
# Same translit table used in kapital_bank.bank_api._AZERI_MAP; duplicated here
|
|
# to keep this module self-contained.
|
|
_AZERI_MAP = str.maketrans("ƏəÜüÖöĞğİıÇ窺", "EeUuOoGgIiCcSs")
|
|
|
|
|
|
def _translit_az(text):
|
|
return (text or "").translate(_AZERI_MAP)
|
|
|
|
|
|
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 = []
|
|
for m in message_log:
|
|
if isinstance(m, dict):
|
|
parts.append(m.get("message") or m.get("msg") or "")
|
|
else:
|
|
parts.append(str(m))
|
|
return " | ".join(p for p in parts if p)
|
|
|
|
|
|
def _effective_case_insensitive(row, global_default):
|
|
"""Per-row Case Mode overrides the global Ignore Case flag."""
|
|
mode = (getattr(row, "case_mode", None) or "").strip()
|
|
if mode == "Ignore Case":
|
|
return True
|
|
if mode == "Case Sensitive":
|
|
return False
|
|
return global_default
|
|
|
|
|
|
def _effective_azeri(row, global_default):
|
|
"""Per-row Azərbaycan Translit mode overrides the global Consider Azərbaycan Characters flag."""
|
|
mode = (getattr(row, "azeri_mode", None) or "").strip()
|
|
if mode == "Apply Translit":
|
|
return True
|
|
if mode == "Strict (No Translit)":
|
|
return False
|
|
return global_default
|
|
|
|
|
|
def _build_name_to_party(settings):
|
|
"""Counterparty text -> {"Customer": erp_customer} / {"Supplier": erp_supplier}.
|
|
|
|
Each row is indexed under up to four key variants depending on its effective
|
|
Case Mode and Azərbaycan Translit settings (both of which fall back to globals):
|
|
strict / lowercase / translit / translit + lowercase.
|
|
"""
|
|
global_ci = bool(getattr(settings, "case_insensitive_party_match", 1))
|
|
global_az = bool(getattr(settings, "consider_azeri_chars", 1))
|
|
idx = {}
|
|
|
|
def _add(name, party_type, erp_party, ci, az):
|
|
key = (name or "").strip()
|
|
if not key:
|
|
return
|
|
idx.setdefault(key, {})[party_type] = erp_party
|
|
if ci:
|
|
idx.setdefault(key.lower(), {})[party_type] = erp_party
|
|
if az:
|
|
translit = _translit_az(key)
|
|
if translit != key:
|
|
idx.setdefault(translit, {})[party_type] = erp_party
|
|
if ci:
|
|
idx.setdefault(translit.lower(), {})[party_type] = erp_party
|
|
|
|
for row in settings.customer_mappings:
|
|
if not row.erp_customer or not row.kb_customer_name:
|
|
continue
|
|
cust_name = frappe.db.get_value("Kapital Bank Customer", row.kb_customer_name, "customer_name")
|
|
if cust_name:
|
|
_add(
|
|
cust_name, "Customer", row.erp_customer,
|
|
_effective_case_insensitive(row, global_ci),
|
|
_effective_azeri(row, global_az),
|
|
)
|
|
for row in settings.supplier_mappings:
|
|
if not row.erp_supplier or not row.kb_supplier_name:
|
|
continue
|
|
supp_name = frappe.db.get_value("Kapital Bank Supplier", row.kb_supplier_name, "supplier_name")
|
|
if supp_name:
|
|
_add(
|
|
supp_name, "Supplier", row.erp_supplier,
|
|
_effective_case_insensitive(row, global_ci),
|
|
_effective_azeri(row, global_az),
|
|
)
|
|
return idx
|
|
|
|
|
|
def _resolve_party_from_name(name, name_to_party, payment_type):
|
|
"""(party_type, erp_party) for `name`, or (None, None). Tries up to four
|
|
lookup variants (strict / lower / translit / translit+lower); the index
|
|
encodes per-row case/azərbaycan decisions, so flags aren't needed here."""
|
|
key = (name or "").strip()
|
|
if not key:
|
|
return None, None
|
|
translit = _translit_az(key)
|
|
cands = (
|
|
name_to_party.get(key)
|
|
or name_to_party.get(key.lower())
|
|
or (name_to_party.get(translit) if translit != key else None)
|
|
or (name_to_party.get(translit.lower()) if translit != key else None)
|
|
)
|
|
if not cands:
|
|
return None, None
|
|
preferred = "Customer" if payment_type == "Receive" else "Supplier"
|
|
other = "Supplier" if preferred == "Customer" else "Customer"
|
|
if preferred in cands:
|
|
return preferred, cands[preferred]
|
|
if other in cands:
|
|
return other, cands[other]
|
|
return None, None
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_purpose_mappings(transactions, paid_from=None, paid_to=None, document_type="Payment Entry", mode="Both"):
|
|
"""Bulk-create transaction mappings and/or documents from selected BRT transactions."""
|
|
try:
|
|
txn_list = json.loads(transactions) if isinstance(transactions, str) else transactions
|
|
|
|
do_mappings = mode == "Mappings Only"
|
|
do_documents = mode in ("Both", "Documents & Reconcile")
|
|
|
|
created_mappings = 0
|
|
already_mapped = 0
|
|
skipped_no_data = 0
|
|
created_docs = 0
|
|
reconciled = 0
|
|
errors = []
|
|
|
|
needs_settings = do_mappings or do_documents
|
|
if needs_settings:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
|
|
# A Bank Transaction created before its counterparty was mapped has no
|
|
# party. Resolve it now, from the customer/supplier mappings, so both
|
|
# counterparty-based rule matching and the created PE/JE pick it up.
|
|
name_to_party = _build_name_to_party(settings)
|
|
for txn in txn_list:
|
|
if txn.get("party") or not txn.get("bank_transaction_name"):
|
|
continue
|
|
bp_name = frappe.db.get_value("Bank Transaction", txn["bank_transaction_name"], "bank_party_name")
|
|
payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
|
|
pt, ep = _resolve_party_from_name(bp_name, name_to_party, payment_type)
|
|
if pt and ep:
|
|
txn["party_type"] = pt
|
|
txn["party"] = ep
|
|
|
|
if mode in ("Both", "Mappings Only"):
|
|
existing_purpose = {
|
|
(row.purpose_keyword, row.payment_type, (row.currency or "").upper())
|
|
for row in settings.transaction_mappings
|
|
if row.purpose_keyword and row.payment_type
|
|
}
|
|
existing_party_only = {
|
|
(row.counterparty_type or "", row.counterparty or "", row.payment_type, (row.currency or "").upper())
|
|
for row in settings.transaction_mappings
|
|
if not row.purpose_keyword and row.counterparty and row.payment_type
|
|
}
|
|
|
|
if do_mappings:
|
|
# Deduplicate within batch; first occurrence wins for party info
|
|
unique_purposes = {}
|
|
unique_parties = {}
|
|
|
|
for txn in txn_list:
|
|
purpose = (txn.get("purpose") or "").strip()
|
|
party_type = (txn.get("party_type") or "").strip()
|
|
party = (txn.get("party") or "").strip()
|
|
payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
|
|
|
|
bt_name = txn.get("bank_transaction_name")
|
|
kb_currency = (frappe.db.get_value("Bank Transaction", bt_name, "kb_currency") or "").strip().upper() if bt_name else ""
|
|
|
|
if not purpose and not party:
|
|
skipped_no_data += 1
|
|
continue
|
|
|
|
if purpose:
|
|
key = (purpose, payment_type, kb_currency)
|
|
if key not in unique_purposes:
|
|
unique_purposes[key] = {
|
|
"count": 0,
|
|
# party_type is already "Customer"/"Supplier" from ERPNext BRT
|
|
"counterparty_type": party_type,
|
|
"counterparty": party,
|
|
}
|
|
unique_purposes[key]["count"] += 1
|
|
else:
|
|
# Party-only mapping (no purpose)
|
|
key = (party_type, party, payment_type, kb_currency)
|
|
if key not in unique_parties:
|
|
unique_parties[key] = {
|
|
"count": 0,
|
|
"counterparty_type": party_type,
|
|
"counterparty": party,
|
|
"payment_type": payment_type,
|
|
}
|
|
unique_parties[key]["count"] += 1
|
|
|
|
for (purpose_text, payment_type, kb_currency), data in unique_purposes.items():
|
|
purpose_name = _ensure_purpose(purpose_text)
|
|
is_foreign = kb_currency and kb_currency != "AZN"
|
|
|
|
if (purpose_name, payment_type, kb_currency) in existing_purpose:
|
|
already_mapped += 1
|
|
continue
|
|
|
|
settings.append("transaction_mappings", {
|
|
"purpose_keyword": purpose_name,
|
|
"payment_type": payment_type,
|
|
"paid_from": paid_from,
|
|
"paid_to": paid_to,
|
|
"document_type": document_type,
|
|
"counterparty_type": data["counterparty_type"] or None,
|
|
"counterparty": data["counterparty"] or None,
|
|
"currency": kb_currency or None,
|
|
"multi_currency": 1 if is_foreign else 0,
|
|
})
|
|
existing_purpose.add((purpose_name, payment_type, kb_currency))
|
|
created_mappings += 1
|
|
|
|
for (party_type_k, party_k, payment_type, kb_currency), data in unique_parties.items():
|
|
is_foreign = kb_currency and kb_currency != "AZN"
|
|
|
|
if (party_type_k, party_k, payment_type, kb_currency) in existing_party_only:
|
|
already_mapped += 1
|
|
continue
|
|
|
|
settings.append("transaction_mappings", {
|
|
"purpose_keyword": None,
|
|
"payment_type": payment_type,
|
|
"paid_from": paid_from,
|
|
"paid_to": paid_to,
|
|
"document_type": document_type,
|
|
"counterparty_type": data["counterparty_type"] or None,
|
|
"counterparty": data["counterparty"] or None,
|
|
"currency": kb_currency or None,
|
|
"multi_currency": 1 if is_foreign else 0,
|
|
})
|
|
existing_party_only.add((party_type_k, party_k, payment_type, kb_currency))
|
|
created_mappings += 1
|
|
|
|
if created_mappings > 0:
|
|
settings.save(ignore_permissions=True)
|
|
for (purpose_text, _payment_type, _kb_currency) in unique_purposes:
|
|
purpose_name = frappe.db.get_value(
|
|
"Kapital Bank Purpose", {"purpose_keyword": purpose_text}, "name"
|
|
)
|
|
if purpose_name:
|
|
frappe.db.set_value(
|
|
"Kapital Bank Purpose", purpose_name, "status", "Mapped",
|
|
update_modified=False,
|
|
)
|
|
frappe.db.commit()
|
|
|
|
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
|
|
az = bool(getattr(settings, "consider_azeri_chars", 1))
|
|
|
|
if mode == "Documents & Reconcile":
|
|
purpose_text_cache = {}
|
|
for row in settings.transaction_mappings:
|
|
if row.purpose_keyword and row.purpose_keyword not in purpose_text_cache:
|
|
text = frappe.db.get_value("Kapital Bank Purpose", row.purpose_keyword, "purpose_keyword") or ""
|
|
purpose_text_cache[row.purpose_keyword] = _norm_text(text, az)
|
|
else:
|
|
purpose_text_cache = {}
|
|
|
|
for txn in txn_list:
|
|
if not txn.get("bank_transaction_name"):
|
|
errors.append({
|
|
"reference_number": txn.get("reference_number") or "",
|
|
"message": "No bank transaction name",
|
|
})
|
|
continue
|
|
|
|
if mode == "Documents & Reconcile":
|
|
mapping_row = _find_mapping_for_txn(txn, txn_mappings, purpose_text_cache, purpose_threshold, az)
|
|
if not mapping_row:
|
|
errors.append({
|
|
"reference_number": txn.get("reference_number") or "",
|
|
"message": f"No mapping found for purpose: '{txn.get('purpose', '')!s:.60}', party: '{txn.get('party', '')}'",
|
|
})
|
|
continue
|
|
txn_paid_from = mapping_row.paid_from
|
|
txn_paid_to = mapping_row.paid_to
|
|
txn_doc_type = mapping_row.document_type or "Payment Entry"
|
|
txn_multi_currency = bool(mapping_row.multi_currency)
|
|
txn_mapping_currency = (mapping_row.currency or "").strip()
|
|
else:
|
|
txn_paid_from = paid_from
|
|
txn_paid_to = paid_to
|
|
txn_doc_type = document_type
|
|
txn_multi_currency = False
|
|
txn_mapping_currency = ""
|
|
|
|
result = _create_and_reconcile_doc(txn, txn_paid_from, txn_paid_to, txn_doc_type, txn_multi_currency, txn_mapping_currency)
|
|
if result.get("success"):
|
|
created_docs += 1
|
|
if result.get("reconciled"):
|
|
reconciled += 1
|
|
elif result.get("error"):
|
|
errors.append({
|
|
"reference_number": txn.get("reference_number") or "",
|
|
"message": result.get("error"),
|
|
})
|
|
|
|
# In "Both" mode: add mapping only after successful document creation
|
|
if mode == "Both":
|
|
purpose = (txn.get("purpose") or "").strip()
|
|
party_type = (txn.get("party_type") or "").strip()
|
|
party = (txn.get("party") or "").strip()
|
|
payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
|
|
bt_name = txn.get("bank_transaction_name")
|
|
kb_currency = (frappe.db.get_value("Bank Transaction", bt_name, "kb_currency") or "").strip().upper() if bt_name else ""
|
|
is_foreign = kb_currency and kb_currency != "AZN"
|
|
|
|
if purpose:
|
|
purpose_name = _ensure_purpose(purpose)
|
|
key = (purpose_name, payment_type, kb_currency)
|
|
if key not in existing_purpose:
|
|
settings.append("transaction_mappings", {
|
|
"purpose_keyword": purpose_name,
|
|
"payment_type": payment_type,
|
|
"paid_from": paid_from,
|
|
"paid_to": paid_to,
|
|
"document_type": document_type,
|
|
"counterparty_type": party_type or None,
|
|
"counterparty": party or None,
|
|
"currency": kb_currency or None,
|
|
"multi_currency": 1 if is_foreign else 0,
|
|
})
|
|
existing_purpose.add(key)
|
|
created_mappings += 1
|
|
elif party:
|
|
key = (party_type, party, payment_type, kb_currency)
|
|
if key not in existing_party_only:
|
|
settings.append("transaction_mappings", {
|
|
"purpose_keyword": None,
|
|
"payment_type": payment_type,
|
|
"paid_from": paid_from,
|
|
"paid_to": paid_to,
|
|
"document_type": document_type,
|
|
"counterparty_type": party_type or None,
|
|
"counterparty": party or None,
|
|
"currency": kb_currency or None,
|
|
"multi_currency": 1 if is_foreign else 0,
|
|
})
|
|
existing_party_only.add(key)
|
|
created_mappings += 1
|
|
else:
|
|
skipped_no_data += 1
|
|
else:
|
|
errors.append({
|
|
"reference_number": txn.get("reference_number") or "",
|
|
"message": result.get("error") or "Unknown error",
|
|
})
|
|
|
|
if mode == "Both" and created_mappings > 0:
|
|
settings.save(ignore_permissions=True)
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"created_mappings": created_mappings,
|
|
"already_mapped": already_mapped,
|
|
"skipped_no_data": skipped_no_data,
|
|
"created_docs": created_docs,
|
|
"reconciled": reconciled,
|
|
"errors": errors,
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"create_purpose_mappings: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Mapping",
|
|
)
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
def _norm_text(text, az):
|
|
"""Lowercased + (optionally) Azərbaycan-translit form for purpose/party comparison."""
|
|
s = (text or "").lower()
|
|
if az:
|
|
s = _translit_az(s)
|
|
return s
|
|
|
|
|
|
def _find_mapping_for_txn(txn, transaction_mappings, purpose_text_cache, purpose_threshold, az=False):
|
|
"""Find the best matching transaction mapping row for a BRT txn.
|
|
|
|
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.
|
|
|
|
`az` toggles Azərbaycan transliteration of purpose / party text before
|
|
comparison (driven by settings.consider_azeri_chars).
|
|
"""
|
|
payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
|
|
txn_purpose = _norm_text(txn.get("purpose"), az)
|
|
txn_party = _norm_text(txn.get("party"), az)
|
|
|
|
bt_name = txn.get("bank_transaction_name")
|
|
if bt_name:
|
|
_bt_cur = frappe.db.get_value("Bank Transaction", bt_name, ["kb_currency", "currency"], as_dict=True) or {}
|
|
txn_kb_currency = (_bt_cur.get("kb_currency") or _bt_cur.get("currency") or "").strip().upper()
|
|
else:
|
|
txn_kb_currency = ""
|
|
|
|
both_rules = []
|
|
purpose_only_rules = []
|
|
counterparty_only_rules = []
|
|
fallback_rules = []
|
|
|
|
for row in transaction_mappings:
|
|
if not row.paid_from or not row.paid_to:
|
|
continue
|
|
if row.payment_type and row.payment_type != payment_type:
|
|
continue
|
|
row_currency = (row.currency or "").strip().upper()
|
|
if row_currency and row_currency != txn_kb_currency:
|
|
continue
|
|
|
|
has_purpose = bool(row.purpose_keyword)
|
|
has_counterparty = bool(row.counterparty)
|
|
|
|
if has_purpose and has_counterparty:
|
|
both_rules.append(row)
|
|
elif has_purpose:
|
|
purpose_only_rules.append(row)
|
|
elif has_counterparty:
|
|
counterparty_only_rules.append(row)
|
|
elif row.counterparty_type:
|
|
fallback_rules.append(row)
|
|
|
|
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 _norm_text(row.counterparty, az) == 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_exact(row) and _counterparty_matches(row):
|
|
return row
|
|
# 2. purpose-only — exact
|
|
for row in purpose_only_rules:
|
|
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
|
|
|
|
# 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:
|
|
return row
|
|
|
|
return None
|
|
|
|
|
|
def _create_and_reconcile_doc(txn, paid_from, paid_to, document_type, multi_currency=False, mapping_currency=""):
|
|
"""Create a Payment Entry or Journal Entry and reconcile it with the bank transaction."""
|
|
bank_transaction_name = txn.get("bank_transaction_name")
|
|
try:
|
|
bank_txn = frappe.get_doc("Bank Transaction", bank_transaction_name)
|
|
|
|
if document_type == "Journal Entry":
|
|
result = _create_journal_entry_for_brt(txn, paid_from, paid_to, bank_txn, multi_currency, mapping_currency)
|
|
else:
|
|
result = _create_payment_entry_for_brt(txn, paid_from, paid_to, bank_txn)
|
|
|
|
if not result.get("success"):
|
|
return result
|
|
|
|
doc_name = result["doc_name"]
|
|
|
|
try:
|
|
from erpnext.accounts.doctype.bank_reconciliation_tool.bank_reconciliation_tool import (
|
|
reconcile_vouchers,
|
|
)
|
|
|
|
reconcile_vouchers(
|
|
bank_transaction_name,
|
|
json.dumps([{
|
|
"payment_doctype": document_type,
|
|
"payment_name": doc_name,
|
|
"amount": abs(float(txn.get("amount") or 0)),
|
|
}]),
|
|
)
|
|
return {"success": True, "doc_name": doc_name, "reconciled": True}
|
|
except Exception as re:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"reconcile_vouchers failed for {doc_name}: {re}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Mapping",
|
|
)
|
|
return {"success": False, "error": str(re)}
|
|
|
|
except Exception as e:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"_create_and_reconcile_doc failed for {bank_transaction_name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Mapping",
|
|
)
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
def _create_payment_entry_for_brt(txn, paid_from, paid_to, bank_txn):
|
|
"""Create and submit a Payment Entry for a BRT transaction."""
|
|
try:
|
|
amount = abs(float(txn.get("amount") or 0))
|
|
party_type = (txn.get("party_type") or "").strip()
|
|
party = (txn.get("party") or "").strip()
|
|
posting_date = bank_txn.date or txn.get("date")
|
|
|
|
pe = frappe.new_doc("Payment Entry")
|
|
pe.payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
|
|
pe.company = bank_txn.company
|
|
pe.posting_date = posting_date
|
|
pe.paid_from = paid_from
|
|
pe.paid_to = paid_to
|
|
pe.paid_amount = amount
|
|
pe.received_amount = amount
|
|
pe.reference_no = bank_txn.reference_number or txn.get("reference_number") or ""
|
|
pe.reference_date = posting_date
|
|
pe.remarks = bank_txn.description or txn.get("purpose") or ""
|
|
# Only attach a party if one of the GL accounts is a party account —
|
|
# otherwise ERPNext rejects the PE ("Party Account ... is required").
|
|
if party_type and party:
|
|
pf_type = frappe.get_cached_value("Account", paid_from, "account_type") or ""
|
|
pt_type = frappe.get_cached_value("Account", paid_to, "account_type") or ""
|
|
if pf_type in ("Receivable", "Payable") or pt_type in ("Receivable", "Payable"):
|
|
pe.party_type = party_type
|
|
pe.party = party
|
|
|
|
frappe.local.message_log = []
|
|
pe.insert(ignore_permissions=True)
|
|
pe.submit()
|
|
|
|
if frappe.local.message_log:
|
|
msgs = _extract_messages(frappe.local.message_log)
|
|
frappe.local.message_log = []
|
|
frappe.db.rollback()
|
|
return {"success": False, "error": msgs}
|
|
|
|
return {"success": True, "doc_name": pe.name}
|
|
|
|
except Exception as e:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"_create_payment_entry_for_brt failed: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Mapping",
|
|
)
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
def _create_journal_entry_for_brt(txn, paid_from, paid_to, bank_txn, multi_currency=False, mapping_currency=""):
|
|
"""Create and submit a Journal Entry for a BRT transaction."""
|
|
try:
|
|
import erpnext
|
|
from erpnext.setup.utils import get_exchange_rate
|
|
|
|
amount = abs(float(txn.get("amount") or 0))
|
|
party_type = (txn.get("party_type") or "").strip()
|
|
party = (txn.get("party") or "").strip()
|
|
is_pay = txn.get("drcr") == "D"
|
|
posting_date = bank_txn.date or txn.get("date")
|
|
ref_no = bank_txn.reference_number or txn.get("reference_number") or ""
|
|
|
|
company_currency = erpnext.get_company_currency(bank_txn.company)
|
|
|
|
# Get account currencies and types
|
|
from_account = frappe.get_cached_value("Account", paid_from, ["account_type", "account_currency"], as_dict=True) or {}
|
|
to_account = frappe.get_cached_value("Account", paid_to, ["account_type", "account_currency"], as_dict=True) or {}
|
|
paid_from_type = from_account.get("account_type") or ""
|
|
paid_to_type = to_account.get("account_type") or ""
|
|
from_currency = from_account.get("account_currency") or company_currency
|
|
to_currency = to_account.get("account_currency") or company_currency
|
|
|
|
# Auto-enable multi_currency if any account is in a foreign currency
|
|
is_multi = from_currency != company_currency or to_currency != company_currency
|
|
|
|
currency_precision = cint(frappe.db.get_single_value("System Settings", "currency_precision") or 2)
|
|
# txn_currency = the currency the BT amount is denominated in.
|
|
# mapping_currency is a FILTER (which transactions this rule matches), NOT the amount's currency.
|
|
# bank_txn.currency = GL account currency = currency of deposit/withdrawal = correct.
|
|
txn_currency = (
|
|
txn.get("currency")
|
|
or (getattr(bank_txn, "currency", None) or "")
|
|
or company_currency
|
|
).strip()
|
|
|
|
# Exchange rates: account_currency → company_currency
|
|
from_rate = 1 if from_currency == company_currency else get_exchange_rate(from_currency, company_currency, posting_date)
|
|
to_rate = 1 if to_currency == company_currency else get_exchange_rate(to_currency, company_currency, posting_date)
|
|
|
|
# Rate for the transaction currency itself
|
|
if txn_currency == from_currency:
|
|
txn_rate = from_rate
|
|
elif txn_currency == to_currency:
|
|
txn_rate = to_rate
|
|
elif txn_currency == company_currency:
|
|
txn_rate = 1
|
|
else:
|
|
txn_rate = get_exchange_rate(txn_currency, company_currency, posting_date)
|
|
|
|
amount_in_company = flt(amount * txn_rate, currency_precision)
|
|
|
|
# Amount in each account's own currency
|
|
if from_currency == txn_currency:
|
|
from_amount = amount
|
|
elif from_currency == company_currency:
|
|
from_amount = amount_in_company
|
|
else:
|
|
from_amount = flt(amount_in_company / from_rate, currency_precision) if from_rate else amount_in_company
|
|
|
|
if to_currency == txn_currency:
|
|
to_amount = amount
|
|
elif to_currency == company_currency:
|
|
to_amount = amount_in_company
|
|
else:
|
|
to_amount = flt(amount_in_company / to_rate, currency_precision) if to_rate else amount_in_company
|
|
|
|
# Recompute exact rates from rounded amounts so Frappe's debit == credit in company currency
|
|
if from_currency != company_currency and from_amount:
|
|
from_rate = amount_in_company / from_amount
|
|
if to_currency != company_currency and to_amount:
|
|
to_rate = amount_in_company / to_amount
|
|
|
|
je = frappe.new_doc("Journal Entry")
|
|
je.voucher_type = "Bank Entry"
|
|
je.posting_date = posting_date
|
|
je.company = bank_txn.company
|
|
je.cheque_no = ref_no
|
|
je.cheque_date = posting_date
|
|
je.user_remark = bank_txn.description or txn.get("purpose") or ""
|
|
je.expense_income_type = "Expense" if is_pay else "Income"
|
|
if is_multi:
|
|
je.multi_currency = 1
|
|
|
|
# paid_from is always credited, paid_to is always debited.
|
|
# For Pay: paid_from = bank/cash account, paid_to = expense/payable account.
|
|
# For Receive: paid_from = income/receivable account, paid_to = bank/cash account.
|
|
# Only set *_in_account_currency + exchange_rate; Frappe computes debit/credit itself.
|
|
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:
|
|
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,
|
|
}
|
|
|
|
if party_type and party:
|
|
if paid_from_type in ("Receivable", "Payable"):
|
|
from_row["party_type"] = party_type
|
|
from_row["party"] = party
|
|
if paid_to_type in ("Receivable", "Payable"):
|
|
to_row["party_type"] = party_type
|
|
to_row["party"] = party
|
|
|
|
je.append("accounts", from_row)
|
|
je.append("accounts", to_row)
|
|
|
|
frappe.local.message_log = []
|
|
je.insert(ignore_permissions=True)
|
|
je.submit()
|
|
|
|
if frappe.local.message_log:
|
|
msgs = _extract_messages(frappe.local.message_log)
|
|
frappe.local.message_log = []
|
|
frappe.db.rollback()
|
|
return {"success": False, "error": msgs}
|
|
|
|
return {"success": True, "doc_name": je.name}
|
|
|
|
except Exception as e:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"_create_journal_entry_for_brt failed: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Mapping",
|
|
)
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
def _ensure_purpose(purpose_text):
|
|
"""Return the name of an existing or newly created Kapital Bank Purpose."""
|
|
existing = frappe.db.get_value(
|
|
"Kapital Bank Purpose", {"purpose_keyword": purpose_text}, "name"
|
|
)
|
|
if existing:
|
|
return existing
|
|
|
|
try:
|
|
doc = frappe.new_doc("Kapital Bank Purpose")
|
|
doc.purpose_keyword = purpose_text
|
|
doc.direction = "Both"
|
|
doc.status = "New"
|
|
doc.insert(ignore_permissions=True)
|
|
return doc.name
|
|
except frappe.DuplicateEntryError:
|
|
return frappe.db.get_value(
|
|
"Kapital Bank Purpose", {"purpose_keyword": purpose_text}, "name"
|
|
)
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Insert purpose failed for {purpose_text}: {e}",
|
|
"Kapital Bank Purpose Loading",
|
|
)
|
|
raise
|