542 lines
18 KiB
Python
542 lines
18 KiB
Python
import json
|
|
|
|
import frappe
|
|
from frappe.utils import cint, flt
|
|
|
|
|
|
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)
|
|
|
|
|
|
@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 in ("Both", "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 mode == "Documents & Reconcile"
|
|
if needs_settings:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
|
|
if do_mappings:
|
|
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
|
|
}
|
|
|
|
# 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
|
|
|
|
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] = text.lower()
|
|
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)
|
|
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)
|
|
else:
|
|
txn_paid_from = paid_from
|
|
txn_paid_to = paid_to
|
|
txn_doc_type = document_type
|
|
txn_multi_currency = False
|
|
|
|
result = _create_and_reconcile_doc(txn, txn_paid_from, txn_paid_to, txn_doc_type, txn_multi_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"),
|
|
})
|
|
else:
|
|
errors.append({
|
|
"reference_number": txn.get("reference_number") or "",
|
|
"message": result.get("error") or "Unknown error",
|
|
})
|
|
|
|
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 _find_mapping_for_txn(txn, transaction_mappings, purpose_text_cache):
|
|
"""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.
|
|
"""
|
|
payment_type = "Pay" if txn.get("drcr") == "D" else "Receive"
|
|
txn_purpose = (txn.get("purpose") or "").lower()
|
|
txn_party = (txn.get("party") or "").lower()
|
|
|
|
bt_name = txn.get("bank_transaction_name")
|
|
txn_kb_currency = (frappe.db.get_value("Bank Transaction", bt_name, "kb_currency") or "").strip().upper() if bt_name else ""
|
|
|
|
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 _purpose_matches(row):
|
|
keyword_text = purpose_text_cache.get(row.purpose_keyword, "")
|
|
return keyword_text and keyword_text in txn_purpose
|
|
|
|
def _counterparty_matches(row):
|
|
return row.counterparty.lower() == txn_party
|
|
|
|
for row in both_rules:
|
|
if _purpose_matches(row) and _counterparty_matches(row):
|
|
return row
|
|
|
|
for row in purpose_only_rules:
|
|
if _purpose_matches(row):
|
|
return row
|
|
|
|
for row in counterparty_only_rules:
|
|
if _counterparty_matches(row):
|
|
return row
|
|
|
|
# Phase 4: 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):
|
|
"""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)
|
|
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.log_error(
|
|
f"reconcile_vouchers failed for {doc_name}: {re}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Mapping",
|
|
)
|
|
return {"success": True, "doc_name": doc_name, "reconciled": 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 ""
|
|
if party_type and party:
|
|
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):
|
|
"""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 = (txn.get("currency") 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.kb_account_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
|