2308 lines
89 KiB
Python
2308 lines
89 KiB
Python
import json
|
|
import re
|
|
import frappe
|
|
from datetime import datetime
|
|
from frappe.utils import cint, flt
|
|
from difflib import SequenceMatcher
|
|
from kapital_bank.api import BIRBankClient
|
|
|
|
|
|
def _mask(text):
|
|
"""Mask digit sequences longer than 6 chars: keep first 3 + *** + last 3."""
|
|
return re.sub(r'\d{7,}', lambda m: m.group()[:3] + '***' + m.group()[-3:], str(text))
|
|
|
|
|
|
|
|
def _api_error_message(resp):
|
|
"""Extract the bank's own error message from response, or fall back to HTTP status text."""
|
|
try:
|
|
body = resp.json()
|
|
msg = body.get("response", {}).get("message", "")
|
|
if msg:
|
|
return msg
|
|
except (ValueError, KeyError):
|
|
pass
|
|
return f"HTTP {resp.status_code}: {resp.text[:200]}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# REGISTRY LOADERS (fetch from Kapital Bank → save to local DocTypes)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def load_accounts_to_registry(login_name=None):
|
|
"""Fetch accounts from GET /accounts and upsert into Kapital Bank Account registry."""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
data = client.get_json("/accounts")
|
|
accounts = data.get("responseData", {}).get("accountsList", [])
|
|
|
|
created = updated = 0
|
|
for a in accounts:
|
|
iban = a.get("ibanAcNo", "").strip()
|
|
if not iban:
|
|
continue
|
|
|
|
values = {
|
|
"cust_ac_no": a.get("custAcNo", ""),
|
|
"currency": a.get("ccy", ""),
|
|
"account_desc": a.get("acDesc", ""),
|
|
"account_status": a.get("status", ""),
|
|
"branch_code": a.get("branchCode", ""),
|
|
"current_balance": flt(a.get("currAmt", 0)),
|
|
"planned_balance": flt(a.get("plannedAmt", 0)),
|
|
"hold": flt(a.get("hold", 0)),
|
|
}
|
|
|
|
if frappe.db.exists("Kapital Bank Account", iban):
|
|
frappe.db.set_value("Kapital Bank Account", iban, values, update_modified=False)
|
|
updated += 1
|
|
else:
|
|
doc = frappe.new_doc("Kapital Bank Account")
|
|
doc.iban = iban
|
|
doc.update(values)
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "created": created, "updated": updated, "total": len(accounts)}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"load_accounts_to_registry failed: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def load_cards_to_registry(login_name=None):
|
|
"""Fetch cards from GET /cards and upsert into Kapital Bank Card registry."""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
data = client.get_json("/cards")
|
|
cards = data.get("responseData", {}).get("cards", [])
|
|
|
|
created = updated = 0
|
|
for c in cards:
|
|
account_number = str(c.get("accountNumber", "")).strip()
|
|
if not account_number:
|
|
continue
|
|
|
|
values = {
|
|
"card_type": c.get("cardType", ""),
|
|
"pan": c.get("pan", ""),
|
|
"currency": c.get("currency", ""),
|
|
"balance": flt(c.get("balance", 0)),
|
|
"expiry_date": str(c.get("expiryDate", "")),
|
|
}
|
|
|
|
if frappe.db.exists("Kapital Bank Card", account_number):
|
|
frappe.db.set_value("Kapital Bank Card", account_number, values, update_modified=False)
|
|
updated += 1
|
|
else:
|
|
doc = frappe.new_doc("Kapital Bank Card")
|
|
doc.account_number = account_number
|
|
doc.update(values)
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "created": created, "updated": updated, "total": len(cards)}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"load_cards_to_registry failed: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def load_counterparties_from_statements(from_date, to_date, login_name=None):
|
|
"""
|
|
Fetch account statements for all registered accounts and extract counterparties
|
|
into Kapital Bank Customer / Kapital Bank Supplier registries based on transaction direction.
|
|
D (debit, money out) → Supplier, C (credit, money in) → Customer.
|
|
"""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
|
|
accounts = frappe.get_all("Kapital Bank Account", fields=["iban", "cust_ac_no"])
|
|
if not accounts:
|
|
return {"success": False, "message": "No accounts in registry. Load accounts first."}
|
|
|
|
customers_created = suppliers_created = skipped = 0
|
|
debug_log = []
|
|
|
|
for acc in accounts:
|
|
iban = acc.iban
|
|
account_number = acc.cust_ac_no or iban
|
|
debug_log.append(f"Account: {_mask(iban)} custAcNo={_mask(account_number)} ({from_date} → {to_date})")
|
|
|
|
try:
|
|
resp = client.get(
|
|
"/v2/statement/account",
|
|
params={"accountNumber": account_number, "fromDate": _fmt_date(from_date), "toDate": _fmt_date(to_date)}
|
|
)
|
|
debug_log.append(f" HTTP {resp.status_code}")
|
|
if not resp.ok:
|
|
debug_log.append(f" Error: {_api_error_message(resp)}")
|
|
continue
|
|
data = resp.json()
|
|
except Exception as e:
|
|
debug_log.append(f" Error: {e}")
|
|
frappe.log_error(f"Statement fetch failed for {_mask(iban)}: {e}", "Kapital Bank Load Counterparties")
|
|
continue
|
|
|
|
resp_obj = data.get("response", {})
|
|
resp_code = resp_obj.get("code", "?")
|
|
resp_msg = resp_obj.get("message", "")
|
|
debug_log.append(f" response.code={resp_code} message={resp_msg!r}")
|
|
|
|
if resp_code != "0":
|
|
debug_log.append(f" SKIPPED (non-zero response code)")
|
|
continue
|
|
|
|
statement_list = (
|
|
data.get("responseData", {})
|
|
.get("operations", {})
|
|
.get("statementList", [])
|
|
)
|
|
debug_log.append(f" statementList entries: {len(statement_list)}")
|
|
|
|
for entry in statement_list:
|
|
contr_raw = (entry.get("contrAccount") or "").strip()
|
|
if not contr_raw:
|
|
continue
|
|
|
|
parts = [p.strip() for p in contr_raw.split(" / ")]
|
|
if len(parts) < 2:
|
|
continue
|
|
|
|
contr_iban = parts[0]
|
|
contr_name = parts[1]
|
|
contr_voen = parts[2] if len(parts) >= 3 else ""
|
|
|
|
if not contr_name:
|
|
continue
|
|
|
|
# Determine direction: D = money out = Supplier, C = money in = Customer
|
|
dr_cr = (entry.get("drcrInd") or "").upper()
|
|
|
|
if dr_cr == "D":
|
|
# Supplier
|
|
doctype = "Kapital Bank Supplier"
|
|
name_field = "supplier_name"
|
|
elif dr_cr == "C":
|
|
# Customer
|
|
doctype = "Kapital Bank Customer"
|
|
name_field = "customer_name"
|
|
else:
|
|
skipped += 1
|
|
continue
|
|
|
|
# Deduplicate: by VÖEN if present, else by name
|
|
if contr_voen and frappe.db.exists(doctype, {"tax_id": contr_voen}):
|
|
skipped += 1
|
|
continue
|
|
if not contr_voen and frappe.db.exists(doctype, {name_field: contr_name}):
|
|
skipped += 1
|
|
continue
|
|
|
|
doc = frappe.new_doc(doctype)
|
|
doc.update({
|
|
name_field: contr_name,
|
|
"tax_id": contr_voen or None,
|
|
"iban": contr_iban or None,
|
|
"status": "New",
|
|
})
|
|
try:
|
|
doc.insert(ignore_permissions=True)
|
|
if dr_cr == "D":
|
|
suppliers_created += 1
|
|
else:
|
|
customers_created += 1
|
|
except frappe.DuplicateEntryError:
|
|
skipped += 1
|
|
except Exception as ins_e:
|
|
frappe.log_error(f"Insert counterparty failed for {_mask(contr_raw)}: {ins_e}", "Kapital Bank Load Counterparties")
|
|
|
|
frappe.db.commit()
|
|
frappe.log_error("\n".join(debug_log), "Kapital Bank Load Counterparties Debug")
|
|
|
|
return {
|
|
"success": True,
|
|
"customers_created": customers_created,
|
|
"suppliers_created": suppliers_created,
|
|
"skipped": skipped,
|
|
"debug": debug_log,
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"load_counterparties_from_statements: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# REFERENCE DATA SUMMARY & LISTS (for Settings data tabs)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def get_kb_reference_data_summary():
|
|
"""Return counts for the Load Data dialog summary."""
|
|
try:
|
|
def counts(doctype, mapped_filter):
|
|
total = frappe.db.count(doctype)
|
|
mapped = frappe.db.count(doctype, filters=mapped_filter)
|
|
return {"total": total, "mapped": mapped}
|
|
|
|
summary = {
|
|
"accounts": counts("Kapital Bank Account", {"status": "Mapped"}),
|
|
"cards": counts("Kapital Bank Card", {"status": "Mapped"}),
|
|
"customers": counts("Kapital Bank Customer", {"status": "Mapped"}),
|
|
"suppliers": counts("Kapital Bank Supplier", {"status": "Mapped"}),
|
|
"purposes": counts("Kapital Bank Purpose", {"status": "Mapped"}),
|
|
}
|
|
return {"success": True, "summary": summary}
|
|
except Exception as e:
|
|
frappe.log_error(str(e), "Kapital Bank Reference Data Summary")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_kb_reference_data_list(data_type, limit=100, offset=0):
|
|
"""Return paginated list of registry records for HTML tab rendering."""
|
|
limit = cint(limit)
|
|
offset = cint(offset)
|
|
|
|
try:
|
|
if data_type == "accounts":
|
|
total = frappe.db.count("Kapital Bank Account")
|
|
data = frappe.get_all(
|
|
"Kapital Bank Account",
|
|
fields=["name", "iban", "currency", "current_balance", "planned_balance",
|
|
"account_status", "status", "bank_account"],
|
|
limit=limit, start=offset,
|
|
order_by="iban asc"
|
|
)
|
|
|
|
elif data_type == "cards":
|
|
total = frappe.db.count("Kapital Bank Card")
|
|
data = frappe.get_all(
|
|
"Kapital Bank Card",
|
|
fields=["name", "account_number", "card_type", "pan", "currency",
|
|
"balance", "expiry_date", "status", "bank_account"],
|
|
limit=limit, start=offset,
|
|
order_by="card_type asc"
|
|
)
|
|
|
|
elif data_type == "customers":
|
|
total = frappe.db.count("Kapital Bank Customer")
|
|
data = frappe.get_all(
|
|
"Kapital Bank Customer",
|
|
fields=["name", "customer_name", "tax_id", "iban", "bank_code",
|
|
"status", "mapped_customer", "creation"],
|
|
limit=limit, start=offset,
|
|
order_by="customer_name asc"
|
|
)
|
|
|
|
elif data_type == "suppliers":
|
|
total = frappe.db.count("Kapital Bank Supplier")
|
|
data = frappe.get_all(
|
|
"Kapital Bank Supplier",
|
|
fields=["name", "supplier_name", "tax_id", "iban", "bank_code",
|
|
"status", "mapped_supplier", "creation"],
|
|
limit=limit, start=offset,
|
|
order_by="supplier_name asc"
|
|
)
|
|
|
|
elif data_type == "purposes":
|
|
total = frappe.db.count("Kapital Bank Purpose")
|
|
data = frappe.get_all(
|
|
"Kapital Bank Purpose",
|
|
fields=["name", "purpose_keyword", "direction", "status"],
|
|
limit=limit, start=offset,
|
|
order_by="purpose_keyword asc"
|
|
)
|
|
|
|
else:
|
|
return {"success": False, "message": f"Unknown data_type: {data_type}"}
|
|
|
|
return {
|
|
"success": True,
|
|
"data": data,
|
|
"total_count": total,
|
|
"has_more": (offset + limit) < total
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"get_kb_reference_data_list({data_type}): {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# UNMAPPED GETTERS (for "Add unmapped" buttons)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def get_unmapped_accounts():
|
|
"""Return accounts not yet present in account_mappings child table."""
|
|
try:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
already_mapped = {row.iban for row in settings.account_mappings if row.iban}
|
|
|
|
all_accounts = frappe.get_all(
|
|
"Kapital Bank Account",
|
|
fields=["name", "iban", "account_desc", "currency", "current_balance", "account_status"]
|
|
)
|
|
unmapped = [a for a in all_accounts if a.iban not in already_mapped]
|
|
return {"success": True, "accounts": unmapped}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(str(e), "Kapital Bank get_unmapped_accounts")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_unmapped_cards():
|
|
"""Return cards not yet present in card_mappings child table."""
|
|
try:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
already_mapped = {row.account_number for row in settings.card_mappings if row.account_number}
|
|
|
|
all_cards = frappe.get_all(
|
|
"Kapital Bank Card",
|
|
fields=["name", "account_number", "pan", "card_type", "currency", "balance"]
|
|
)
|
|
unmapped = [c for c in all_cards if c.account_number not in already_mapped]
|
|
return {"success": True, "cards": unmapped}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(str(e), "Kapital Bank get_unmapped_cards")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_unmapped_customers():
|
|
"""Return Kapital Bank Customers with status='New' not yet in customer_mappings."""
|
|
try:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
already_mapped = {row.kb_customer_name for row in settings.customer_mappings if row.kb_customer_name}
|
|
|
|
customers = frappe.get_all(
|
|
"Kapital Bank Customer",
|
|
filters={"status": "New"},
|
|
fields=["name", "customer_name", "tax_id", "iban"]
|
|
)
|
|
unmapped = [c for c in customers if c.name not in already_mapped]
|
|
return {"success": True, "customers": unmapped}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(str(e), "Kapital Bank get_unmapped_customers")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_unmapped_suppliers():
|
|
"""Return Kapital Bank Suppliers with status='New' not yet in supplier_mappings."""
|
|
try:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
already_mapped = {row.kb_supplier_name for row in settings.supplier_mappings if row.kb_supplier_name}
|
|
|
|
suppliers = frappe.get_all(
|
|
"Kapital Bank Supplier",
|
|
filters={"status": "New"},
|
|
fields=["name", "supplier_name", "tax_id", "iban"]
|
|
)
|
|
unmapped = [s for s in suppliers if s.name not in already_mapped]
|
|
return {"success": True, "suppliers": unmapped}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(str(e), "Kapital Bank get_unmapped_suppliers")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_unmapped_purposes():
|
|
"""Return Kapital Bank Purposes with status='New' not yet in transaction_mappings."""
|
|
try:
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
already_mapped = {row.purpose_keyword for row in settings.transaction_mappings if row.purpose_keyword}
|
|
|
|
purposes = frappe.get_all(
|
|
"Kapital Bank Purpose",
|
|
filters={"status": "New"},
|
|
fields=["name", "purpose_keyword", "direction"]
|
|
)
|
|
unmapped = [p for p in purposes if p.name not in already_mapped]
|
|
return {"success": True, "purposes": unmapped}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(str(e), "Kapital Bank get_unmapped_purposes")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# CUSTOMER MATCHING & CREATION
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def match_similar_customers():
|
|
"""Auto-match Kapital Bank Customers to ERPNext Customers by name similarity."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
threshold = flt(doc.similarity_threshold_customers or 90) / 100.0
|
|
consider_azeri = bool(doc.consider_azeri_chars)
|
|
|
|
new_customers = frappe.get_all(
|
|
"Kapital Bank Customer",
|
|
filters={"status": "New"},
|
|
fields=["name", "customer_name", "tax_id"]
|
|
)
|
|
if not new_customers:
|
|
return {"success": True, "matched_count": 0, "total_processed": 0}
|
|
|
|
erp_customers = frappe.get_all("Customer", fields=["name", "customer_name", "tax_id"])
|
|
|
|
# VÖEN lookup for exact matching
|
|
voen_index = {}
|
|
for c in erp_customers:
|
|
if c.tax_id:
|
|
voen_index[c.tax_id.strip()] = c.name
|
|
|
|
# Build dict of existing mappings: kb_customer_name → erp_customer
|
|
existing_mappings = {
|
|
row.kb_customer_name: row.erp_customer
|
|
for row in doc.customer_mappings if row.kb_customer_name
|
|
}
|
|
|
|
# Group 1: customers already in the table but with empty erp_customer
|
|
customers_with_empty_mapping = []
|
|
for kb_name, erp_customer in existing_mappings.items():
|
|
if not erp_customer:
|
|
for c in new_customers:
|
|
if c.name == kb_name:
|
|
customers_with_empty_mapping.append(c)
|
|
break
|
|
|
|
# Group 2: customers not in the table at all
|
|
customers_not_in_table = [c for c in new_customers if c.name not in existing_mappings]
|
|
|
|
customers_to_process = customers_not_in_table + customers_with_empty_mapping
|
|
|
|
matched = 0
|
|
for customer in customers_to_process:
|
|
best_party = None
|
|
|
|
# Priority 1: exact VÖEN match
|
|
if customer.tax_id and customer.tax_id.strip() in voen_index:
|
|
best_party = voen_index[customer.tax_id.strip()]
|
|
else:
|
|
# Priority 2: name similarity
|
|
kb_name = _normalize(customer.customer_name, consider_azeri)
|
|
best_score = 0
|
|
|
|
for c in erp_customers:
|
|
score = SequenceMatcher(None, kb_name, _normalize(c.customer_name, consider_azeri)).ratio()
|
|
if score > best_score:
|
|
best_score = score
|
|
best_party = c.name
|
|
|
|
if best_score < threshold:
|
|
best_party = None
|
|
|
|
if best_party:
|
|
# Update existing row if present, else append new row
|
|
existing_idx = None
|
|
for idx, row in enumerate(doc.customer_mappings):
|
|
if row.kb_customer_name == customer.name:
|
|
existing_idx = idx
|
|
break
|
|
|
|
if existing_idx is not None:
|
|
doc.customer_mappings[existing_idx].erp_customer = best_party
|
|
doc.customer_mappings[existing_idx].mapping_type = "Automatic"
|
|
else:
|
|
doc.append("customer_mappings", {
|
|
"kb_customer_name": customer.name,
|
|
"tax_id": customer.tax_id or None,
|
|
"erp_customer": best_party,
|
|
"mapping_type": "Automatic",
|
|
})
|
|
|
|
frappe.db.set_value("Kapital Bank Customer", customer.name, {
|
|
"mapped_customer": best_party,
|
|
"status": "Mapped",
|
|
}, update_modified=False)
|
|
matched += 1
|
|
|
|
if matched > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "matched_count": matched, "total_processed": len(customers_to_process)}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"match_similar_customers: {e}\n{frappe.get_traceback()}", "Kapital Bank Customer Matching")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def match_similar_suppliers():
|
|
"""Auto-match Kapital Bank Suppliers to ERPNext Suppliers by name similarity."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
threshold = flt(doc.similarity_threshold_suppliers or 80) / 100.0
|
|
consider_azeri = bool(doc.consider_azeri_chars)
|
|
|
|
new_suppliers = frappe.get_all(
|
|
"Kapital Bank Supplier",
|
|
filters={"status": "New"},
|
|
fields=["name", "supplier_name", "tax_id"]
|
|
)
|
|
if not new_suppliers:
|
|
return {"success": True, "matched_count": 0, "total_processed": 0}
|
|
|
|
erp_suppliers = frappe.get_all("Supplier", fields=["name", "supplier_name", "tax_id"])
|
|
|
|
# VÖEN lookup for exact matching
|
|
voen_index = {}
|
|
for s in erp_suppliers:
|
|
if s.tax_id:
|
|
voen_index[s.tax_id.strip()] = s.name
|
|
|
|
# Build dict of existing mappings: kb_supplier_name → erp_supplier
|
|
existing_mappings = {
|
|
row.kb_supplier_name: row.erp_supplier
|
|
for row in doc.supplier_mappings if row.kb_supplier_name
|
|
}
|
|
|
|
# Group 1: suppliers already in the table but with empty erp_supplier
|
|
suppliers_with_empty_mapping = []
|
|
for kb_name, erp_supplier in existing_mappings.items():
|
|
if not erp_supplier:
|
|
for s in new_suppliers:
|
|
if s.name == kb_name:
|
|
suppliers_with_empty_mapping.append(s)
|
|
break
|
|
|
|
# Group 2: suppliers not in the table at all
|
|
suppliers_not_in_table = [s for s in new_suppliers if s.name not in existing_mappings]
|
|
|
|
suppliers_to_process = suppliers_not_in_table + suppliers_with_empty_mapping
|
|
|
|
matched = 0
|
|
for supplier in suppliers_to_process:
|
|
best_party = None
|
|
|
|
# Priority 1: exact VÖEN match
|
|
if supplier.tax_id and supplier.tax_id.strip() in voen_index:
|
|
best_party = voen_index[supplier.tax_id.strip()]
|
|
else:
|
|
# Priority 2: name similarity
|
|
kb_name = _normalize(supplier.supplier_name, consider_azeri)
|
|
best_score = 0
|
|
|
|
for s in erp_suppliers:
|
|
score = SequenceMatcher(None, kb_name, _normalize(s.supplier_name, consider_azeri)).ratio()
|
|
if score > best_score:
|
|
best_score = score
|
|
best_party = s.name
|
|
|
|
if best_score < threshold:
|
|
best_party = None
|
|
|
|
if best_party:
|
|
existing_idx = None
|
|
for idx, row in enumerate(doc.supplier_mappings):
|
|
if row.kb_supplier_name == supplier.name:
|
|
existing_idx = idx
|
|
break
|
|
|
|
if existing_idx is not None:
|
|
doc.supplier_mappings[existing_idx].erp_supplier = best_party
|
|
doc.supplier_mappings[existing_idx].mapping_type = "Automatic"
|
|
else:
|
|
doc.append("supplier_mappings", {
|
|
"kb_supplier_name": supplier.name,
|
|
"tax_id": supplier.tax_id or None,
|
|
"erp_supplier": best_party,
|
|
"mapping_type": "Automatic",
|
|
})
|
|
|
|
frappe.db.set_value("Kapital Bank Supplier", supplier.name, {
|
|
"mapped_supplier": best_party,
|
|
"status": "Mapped",
|
|
}, update_modified=False)
|
|
matched += 1
|
|
|
|
if matched > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "matched_count": matched, "total_processed": len(suppliers_to_process)}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"match_similar_suppliers: {e}\n{frappe.get_traceback()}", "Kapital Bank Supplier Matching")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_unmapped_customers(settings_name=None):
|
|
"""Create Customer records for unmapped customers from Settings.customer_mappings table."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
|
|
unmapped = []
|
|
for mapping in doc.customer_mappings:
|
|
if not mapping.erp_customer and mapping.kb_customer_name:
|
|
if frappe.db.exists("Kapital Bank Customer", mapping.kb_customer_name):
|
|
kb_customer = frappe.get_doc("Kapital Bank Customer", mapping.kb_customer_name)
|
|
unmapped.append((mapping, kb_customer))
|
|
|
|
if not unmapped:
|
|
return {"success": True, "created_count": 0, "message": "No unmapped customers in table to create"}
|
|
|
|
created = 0
|
|
for mapping, kb_customer in unmapped:
|
|
try:
|
|
customer_name = kb_customer.customer_name
|
|
if frappe.db.exists("Customer", {"customer_name": customer_name}):
|
|
continue
|
|
|
|
party = frappe.new_doc("Customer")
|
|
party.customer_name = customer_name
|
|
party.customer_type = "Company"
|
|
|
|
customer_group = mapping.customer_group or doc.default_customer_group
|
|
if not customer_group or not frappe.db.exists("Customer Group", customer_group):
|
|
customer_group = "All Customer Groups"
|
|
party.customer_group = customer_group
|
|
|
|
territory = mapping.territory or doc.default_territory or "All Territories"
|
|
if not frappe.db.exists("Territory", territory):
|
|
territory = "All Territories"
|
|
party.territory = territory
|
|
|
|
if kb_customer.tax_id:
|
|
party.tax_id = kb_customer.tax_id
|
|
|
|
payment_terms = mapping.payment_terms or doc.default_payment_terms
|
|
if payment_terms:
|
|
party.payment_terms = payment_terms
|
|
|
|
party.insert(ignore_permissions=True)
|
|
|
|
# Update mapping row
|
|
mapping.erp_customer = party.name
|
|
mapping.mapping_type = "Automatic"
|
|
|
|
# Update KB Customer status
|
|
frappe.db.set_value("Kapital Bank Customer", kb_customer.name, {
|
|
"status": "Mapped",
|
|
"mapped_customer": party.name,
|
|
"customer_group": party.customer_group,
|
|
"territory": party.territory,
|
|
"payment_terms": payment_terms or None,
|
|
}, update_modified=False)
|
|
|
|
created += 1
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_customers: error for {kb_customer.customer_name}: {e}", "Kapital Bank Customer Creation")
|
|
|
|
if created > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "created_count": created}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_customers: {e}\n{frappe.get_traceback()}", "Kapital Bank Customer Creation")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_unmapped_suppliers(settings_name=None):
|
|
"""Create Supplier records for unmapped suppliers from Settings.supplier_mappings table."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
|
|
unmapped = []
|
|
for mapping in doc.supplier_mappings:
|
|
if not mapping.erp_supplier and mapping.kb_supplier_name:
|
|
if frappe.db.exists("Kapital Bank Supplier", mapping.kb_supplier_name):
|
|
kb_supplier = frappe.get_doc("Kapital Bank Supplier", mapping.kb_supplier_name)
|
|
unmapped.append((mapping, kb_supplier))
|
|
|
|
if not unmapped:
|
|
return {"success": True, "created_count": 0, "message": "No unmapped suppliers in table to create"}
|
|
|
|
created = 0
|
|
for mapping, kb_supplier in unmapped:
|
|
try:
|
|
supplier_name = kb_supplier.supplier_name
|
|
if frappe.db.exists("Supplier", {"supplier_name": supplier_name}):
|
|
continue
|
|
|
|
party = frappe.new_doc("Supplier")
|
|
party.supplier_name = supplier_name
|
|
party.supplier_type = "Company"
|
|
|
|
supplier_group = mapping.supplier_group or doc.default_supplier_group
|
|
if not supplier_group or not frappe.db.exists("Supplier Group", supplier_group):
|
|
supplier_group = "All Supplier Groups"
|
|
party.supplier_group = supplier_group
|
|
|
|
if kb_supplier.tax_id:
|
|
party.tax_id = kb_supplier.tax_id
|
|
|
|
payment_terms = mapping.payment_terms or doc.default_payment_terms
|
|
if payment_terms:
|
|
party.payment_terms = payment_terms
|
|
|
|
party.insert(ignore_permissions=True)
|
|
|
|
# Update mapping row
|
|
mapping.erp_supplier = party.name
|
|
mapping.mapping_type = "Automatic"
|
|
|
|
# Update KB Supplier status
|
|
frappe.db.set_value("Kapital Bank Supplier", kb_supplier.name, {
|
|
"status": "Mapped",
|
|
"mapped_supplier": party.name,
|
|
"supplier_group": party.supplier_group,
|
|
"payment_terms": payment_terms or None,
|
|
}, update_modified=False)
|
|
|
|
created += 1
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_suppliers: error for {kb_supplier.supplier_name}: {e}", "Kapital Bank Supplier Creation")
|
|
|
|
if created > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "created_count": created}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_suppliers: {e}\n{frappe.get_traceback()}", "Kapital Bank Supplier Creation")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# ACCOUNT / CARD → BANK ACCOUNT MATCHING & CREATION
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def match_similar_accounts():
|
|
"""Auto-match account mappings to ERPNext Bank Accounts by IBAN."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
|
|
erp_bank_accounts = frappe.get_all(
|
|
"Bank Account",
|
|
fields=["name", "bank_account_no", "iban"],
|
|
)
|
|
|
|
# Build lookup: iban/bank_account_no → (name, account)
|
|
iban_index = {}
|
|
for ba in erp_bank_accounts:
|
|
if ba.iban:
|
|
iban_index[ba.iban.strip()] = ba
|
|
if ba.bank_account_no:
|
|
iban_index[ba.bank_account_no.strip()] = ba
|
|
|
|
matched = 0
|
|
total_processed = 0
|
|
|
|
for row in doc.account_mappings:
|
|
if row.bank_account or not row.iban:
|
|
continue
|
|
total_processed += 1
|
|
|
|
match = iban_index.get(row.iban.strip())
|
|
if match:
|
|
row.bank_account = match.name
|
|
# Fetch GL account from Bank Account
|
|
gl_account = frappe.db.get_value("Bank Account", match.name, "account")
|
|
if gl_account:
|
|
row.gl_account = gl_account
|
|
|
|
frappe.db.set_value("Kapital Bank Account", row.iban, {
|
|
"status": "Mapped",
|
|
"bank_account": match.name,
|
|
}, update_modified=False)
|
|
matched += 1
|
|
|
|
if matched > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "matched_count": matched, "total_processed": total_processed}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"match_similar_accounts: {e}\n{frappe.get_traceback()}", "Kapital Bank Account Matching")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_unmapped_accounts(settings_name=None):
|
|
"""Create Bank Account records for unmapped accounts in account_mappings."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
|
|
unmapped = []
|
|
for row in doc.account_mappings:
|
|
if not row.bank_account and row.iban:
|
|
unmapped.append(row)
|
|
|
|
if not unmapped:
|
|
return {"success": True, "created_count": 0, "message": "No unmapped accounts to create"}
|
|
|
|
default_bank = doc.default_bank
|
|
default_company = doc.default_company
|
|
|
|
created = 0
|
|
for row in unmapped:
|
|
try:
|
|
# Check if Bank Account already exists for this IBAN
|
|
existing = frappe.db.get_value("Bank Account", {"bank_account_no": row.iban}, "name")
|
|
if existing:
|
|
row.bank_account = existing
|
|
gl_account = frappe.db.get_value("Bank Account", existing, "account")
|
|
if gl_account:
|
|
row.gl_account = gl_account
|
|
frappe.db.set_value("Kapital Bank Account", row.iban, {
|
|
"status": "Mapped",
|
|
"bank_account": existing,
|
|
}, update_modified=False)
|
|
created += 1
|
|
continue
|
|
|
|
account_name = row.account_label or row.iban
|
|
|
|
ba = frappe.new_doc("Bank Account")
|
|
ba.account_name = account_name
|
|
ba.bank_account_no = row.iban
|
|
if default_bank:
|
|
ba.bank = default_bank
|
|
if default_company:
|
|
ba.company = default_company
|
|
ba.insert(ignore_permissions=True)
|
|
|
|
row.bank_account = ba.name
|
|
gl_account = frappe.db.get_value("Bank Account", ba.name, "account")
|
|
if gl_account:
|
|
row.gl_account = gl_account
|
|
|
|
frappe.db.set_value("Kapital Bank Account", row.iban, {
|
|
"status": "Mapped",
|
|
"bank_account": ba.name,
|
|
}, update_modified=False)
|
|
created += 1
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_accounts: error for {row.iban}: {e}", "Kapital Bank Account Creation")
|
|
|
|
if created > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "created_count": created}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_accounts: {e}\n{frappe.get_traceback()}", "Kapital Bank Account Creation")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def match_similar_cards():
|
|
"""Auto-match card mappings to ERPNext Bank Accounts by account_number."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
|
|
erp_bank_accounts = frappe.get_all(
|
|
"Bank Account",
|
|
fields=["name", "bank_account_no"],
|
|
)
|
|
|
|
# Build lookup: bank_account_no → Bank Account
|
|
acno_index = {}
|
|
for ba in erp_bank_accounts:
|
|
if ba.bank_account_no:
|
|
acno_index[ba.bank_account_no.strip()] = ba
|
|
|
|
matched = 0
|
|
total_processed = 0
|
|
|
|
for row in doc.card_mappings:
|
|
if row.bank_account or not row.account_number:
|
|
continue
|
|
total_processed += 1
|
|
|
|
match = acno_index.get(row.account_number.strip())
|
|
if match:
|
|
row.bank_account = match.name
|
|
gl_account = frappe.db.get_value("Bank Account", match.name, "account")
|
|
if gl_account:
|
|
row.gl_account = gl_account
|
|
|
|
frappe.db.set_value("Kapital Bank Card", {"account_number": row.account_number}, {
|
|
"status": "Mapped",
|
|
"bank_account": match.name,
|
|
}, update_modified=False)
|
|
matched += 1
|
|
|
|
if matched > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "matched_count": matched, "total_processed": total_processed}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"match_similar_cards: {e}\n{frappe.get_traceback()}", "Kapital Bank Card Matching")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_unmapped_cards(settings_name=None):
|
|
"""Create Bank Account records for unmapped cards in card_mappings."""
|
|
try:
|
|
doc = frappe.get_doc("Kapital Bank Settings")
|
|
|
|
unmapped = []
|
|
for row in doc.card_mappings:
|
|
if not row.bank_account and row.account_number:
|
|
unmapped.append(row)
|
|
|
|
if not unmapped:
|
|
return {"success": True, "created_count": 0, "message": "No unmapped cards to create"}
|
|
|
|
default_bank = doc.default_bank
|
|
default_company = doc.default_company
|
|
|
|
created = 0
|
|
for row in unmapped:
|
|
try:
|
|
# Check if Bank Account already exists for this account_number
|
|
existing = frappe.db.get_value("Bank Account", {"bank_account_no": row.account_number}, "name")
|
|
if existing:
|
|
row.bank_account = existing
|
|
gl_account = frappe.db.get_value("Bank Account", existing, "account")
|
|
if gl_account:
|
|
row.gl_account = gl_account
|
|
frappe.db.set_value("Kapital Bank Card", {"account_number": row.account_number}, {
|
|
"status": "Mapped",
|
|
"bank_account": existing,
|
|
}, update_modified=False)
|
|
created += 1
|
|
continue
|
|
|
|
account_name = row.card_label or row.account_number
|
|
|
|
ba = frappe.new_doc("Bank Account")
|
|
ba.account_name = account_name
|
|
ba.bank_account_no = row.account_number
|
|
if default_bank:
|
|
ba.bank = default_bank
|
|
if default_company:
|
|
ba.company = default_company
|
|
ba.insert(ignore_permissions=True)
|
|
|
|
row.bank_account = ba.name
|
|
gl_account = frappe.db.get_value("Bank Account", ba.name, "account")
|
|
if gl_account:
|
|
row.gl_account = gl_account
|
|
|
|
frappe.db.set_value("Kapital Bank Card", {"account_number": row.account_number}, {
|
|
"status": "Mapped",
|
|
"bank_account": ba.name,
|
|
}, update_modified=False)
|
|
created += 1
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_cards: error for {row.account_number}: {e}", "Kapital Bank Card Creation")
|
|
|
|
if created > 0:
|
|
doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
return {"success": True, "created_count": created}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"create_unmapped_cards: {e}\n{frappe.get_traceback()}", "Kapital Bank Card Creation")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# PURPOSE LOADING FROM STATEMENTS
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def load_purposes_from_statements(from_date, to_date, login_name=None):
|
|
"""Load unique purposes from account statements into Kapital Bank Purpose registry."""
|
|
try:
|
|
# Get all registered accounts
|
|
accounts = frappe.get_all(
|
|
"Kapital Bank Account",
|
|
fields=["name", "cust_ac_no"],
|
|
)
|
|
|
|
if not accounts:
|
|
return {"success": True, "created": 0, "total_found": 0, "message": "No accounts in registry"}
|
|
|
|
client = BIRBankClient(login_name)
|
|
# purpose_text → set of directions ("D", "C")
|
|
purpose_directions = {}
|
|
|
|
for account in accounts:
|
|
if not account.cust_ac_no:
|
|
continue
|
|
try:
|
|
resp = client.get(
|
|
"/v2/statement/account",
|
|
params={
|
|
"accountNumber": account.cust_ac_no,
|
|
"fromDate": _fmt_date(from_date),
|
|
"toDate": _fmt_date(to_date),
|
|
}
|
|
)
|
|
if not resp.ok:
|
|
continue
|
|
|
|
data = resp.json()
|
|
resp_code = data.get("response", {}).get("code", "?")
|
|
if resp_code != "0":
|
|
continue
|
|
|
|
statement_list = (
|
|
data.get("responseData", {})
|
|
.get("operations", {})
|
|
.get("statementList", [])
|
|
)
|
|
|
|
for txn in statement_list:
|
|
purpose = (txn.get("purpose") or "").strip()
|
|
dr_cr = (txn.get("drcrInd") or "").upper()
|
|
if purpose:
|
|
if purpose not in purpose_directions:
|
|
purpose_directions[purpose] = set()
|
|
if dr_cr in ("D", "C"):
|
|
purpose_directions[purpose].add(dr_cr)
|
|
|
|
except Exception:
|
|
continue
|
|
|
|
# Create Kapital Bank Purpose records for new purposes
|
|
created = updated = 0
|
|
for purpose in sorted(purpose_directions):
|
|
dirs = purpose_directions[purpose]
|
|
if "D" in dirs and "C" in dirs:
|
|
direction = "Both"
|
|
elif "C" in dirs:
|
|
direction = "Receive"
|
|
else:
|
|
direction = "Pay"
|
|
|
|
existing = frappe.db.get_value("Kapital Bank Purpose", {"purpose_keyword": purpose}, ["name", "direction"], as_dict=True)
|
|
if existing:
|
|
# Update direction if it changed (e.g. was Pay, now seen as Both)
|
|
if existing.direction != direction and direction == "Both":
|
|
frappe.db.set_value("Kapital Bank Purpose", existing.name, "direction", "Both", update_modified=False)
|
|
updated += 1
|
|
else:
|
|
try:
|
|
doc = frappe.new_doc("Kapital Bank Purpose")
|
|
doc.purpose_keyword = purpose
|
|
doc.direction = direction
|
|
doc.status = "New"
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
except frappe.DuplicateEntryError:
|
|
pass
|
|
except Exception as ins_e:
|
|
frappe.log_error(f"Insert purpose failed for {purpose}: {ins_e}", "Kapital Bank Purpose Loading")
|
|
|
|
if created > 0:
|
|
frappe.db.commit()
|
|
|
|
return {"success": True, "created": created, "updated": updated, "total_found": len(purpose_directions)}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"load_purposes_from_statements: {e}\n{frappe.get_traceback()}", "Kapital Bank Purpose Loading")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# RAW API ACCESSORS (for reports / manual use)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@frappe.whitelist()
|
|
def get_accounts(login_name=None):
|
|
"""GET /api/b2b/accounts — raw account list (not saved to registry)."""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
data = client.get_json("/accounts")
|
|
accounts = data.get("responseData", {}).get("accountsList", [])
|
|
return {"success": True, "accounts": accounts}
|
|
except Exception as e:
|
|
frappe.log_error(f"get_accounts failed: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_account_statement(login_name=None, iban=None, from_date=None, to_date=None):
|
|
"""GET /api/b2b/v2/statement/account — account statement for a date range."""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
params = {"accountNumber": iban, "fromDate": _fmt_date(from_date), "toDate": _fmt_date(to_date)}
|
|
data = client.get_json("/v2/statement/account", params=params)
|
|
operations = data.get("responseData", {}).get("operations", {})
|
|
return {"success": True, "data": operations}
|
|
except Exception as e:
|
|
frappe.log_error(f"get_account_statement failed: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_cards(login_name=None):
|
|
"""GET /api/b2b/cards — raw card list (not saved to registry)."""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
data = client.get_json("/cards")
|
|
cards = data.get("responseData", {}).get("cards", [])
|
|
return {"success": True, "cards": cards}
|
|
except Exception as e:
|
|
frappe.log_error(f"get_cards failed: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_card_statement(login_name=None, account_no=None, from_date=None, to_date=None):
|
|
"""GET /api/b2b/v2/statement/card — card statement for a date range."""
|
|
try:
|
|
client = BIRBankClient(login_name)
|
|
params = {"accountNumber": account_no, "fromDate": str(from_date) if from_date else None, "toDate": str(to_date) if to_date else None}
|
|
data = client.get_json("/v2/statement/card", params=params)
|
|
operations = data.get("responseData", {}).get("operations", {})
|
|
return {"success": True, "data": operations}
|
|
except Exception as e:
|
|
frappe.log_error(f"get_card_statement failed: {e}\n{frappe.get_traceback()}", "Kapital Bank API")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# STATEMENT IMPORT (Kapital Bank → Payment Entry)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def _parse_txn_date(trn_dt):
|
|
"""Parse transaction date string into a date object. Returns None on failure."""
|
|
trn_dt = (trn_dt or "").strip()
|
|
if not trn_dt:
|
|
return None
|
|
for fmt in ("%b %d, %Y", "%b %d,%Y"):
|
|
try:
|
|
return datetime.strptime(trn_dt, fmt).date()
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _parse_card_txn_date(date_str):
|
|
"""Parse card transaction date DD.MM.YYYY into a date object."""
|
|
date_str = (date_str or "").strip()
|
|
if not date_str:
|
|
return None
|
|
try:
|
|
return datetime.strptime(date_str, "%d.%m.%Y").date()
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _parse_counterparty(contr_raw):
|
|
"""Parse contrAccount string → (name, voen)."""
|
|
parts = [p.strip() for p in (contr_raw or "").split(" / ")]
|
|
contr_name = parts[1] if len(parts) > 1 else ""
|
|
contr_voen = parts[2] if len(parts) > 2 else ""
|
|
return contr_name, contr_voen
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_statement_transactions(from_date, to_date, account_iban, login_name=None):
|
|
"""Fetch bank statement and return parsed transactions for UI selection.
|
|
|
|
Filters out duplicates (already imported by reference_no).
|
|
Returns: {success, transactions: [...], total_fetched, skipped_duplicates}
|
|
"""
|
|
try:
|
|
account_iban = account_iban.strip()
|
|
|
|
# Find cust_ac_no for this IBAN
|
|
cust_ac_no = frappe.db.get_value("Kapital Bank Account", account_iban, "cust_ac_no")
|
|
if not cust_ac_no:
|
|
return {"success": False, "message": f"Account {account_iban} not found in registry."}
|
|
|
|
# Fetch statement
|
|
client = BIRBankClient(login_name)
|
|
resp = client.get(
|
|
"/v2/statement/account",
|
|
params={
|
|
"accountNumber": cust_ac_no,
|
|
"fromDate": _fmt_date(from_date),
|
|
"toDate": _fmt_date(to_date),
|
|
}
|
|
)
|
|
if not resp.ok:
|
|
return {"success": False, "message": _api_error_message(resp)}
|
|
|
|
data = resp.json()
|
|
resp_code = data.get("response", {}).get("code", "?")
|
|
if resp_code != "0":
|
|
resp_msg = data.get("response", {}).get("message", "")
|
|
return {"success": False, "message": resp_msg or f"API error code={resp_code}"}
|
|
|
|
statement_list = (
|
|
data.get("responseData", {})
|
|
.get("operations", {})
|
|
.get("statementList", [])
|
|
)
|
|
|
|
# Use the account's own currency for all transactions from this account
|
|
account_currency = frappe.db.get_value("Kapital Bank Account", account_iban, "currency") or "AZN"
|
|
|
|
transactions = []
|
|
skipped_duplicates = 0
|
|
|
|
for txn in statement_list:
|
|
trn_ref = (txn.get("trnRefNo") or "").strip()
|
|
|
|
# Deduplication via tracking doctype
|
|
if trn_ref and frappe.db.exists("Kapital Bank Transaction", {"reference_no": trn_ref}):
|
|
skipped_duplicates += 1
|
|
continue
|
|
|
|
# Parse date
|
|
parsed_date = _parse_txn_date(txn.get("trnDt"))
|
|
if not parsed_date:
|
|
continue
|
|
|
|
# Amount and currency (use account currency, not the counterparty/operation currency from API)
|
|
amount = abs(flt(txn.get("fcyAmount") or txn.get("lcyAmount") or 0))
|
|
currency = account_currency
|
|
|
|
# Direction
|
|
dr_cr = (txn.get("drcrInd") or "").upper()
|
|
|
|
# Counterparty
|
|
contr_raw = txn.get("contrAccount") or ""
|
|
contr_parts = [p.strip() for p in contr_raw.split(" / ")]
|
|
contr_iban = contr_parts[0] if contr_parts else ""
|
|
contr_name = contr_parts[1] if len(contr_parts) > 1 else ""
|
|
contr_voen = contr_parts[2] if len(contr_parts) > 2 else ""
|
|
|
|
# Purpose
|
|
purpose = (txn.get("purpose") or "").strip()
|
|
|
|
transactions.append({
|
|
"ref_no": trn_ref,
|
|
"date": str(parsed_date),
|
|
"counterparty": contr_name,
|
|
"contr_voen": contr_voen,
|
|
"amount": amount,
|
|
"currency": currency,
|
|
"kb_currency": (txn.get("acCcy") or "").strip(),
|
|
"drcr": dr_cr or "D",
|
|
"purpose": purpose,
|
|
"source_type": "account",
|
|
"source_label": account_iban,
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"transactions": transactions,
|
|
"total_fetched": len(statement_list),
|
|
"skipped_duplicates": skipped_duplicates,
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"get_statement_transactions: {e}\n{frappe.get_traceback()}", "Kapital Bank Import Statement")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_card_statement_transactions(from_date, to_date, card_account_number, login_name=None):
|
|
"""Fetch card statement and return parsed transactions for UI selection.
|
|
|
|
Uses /v2/statement/card with DD.MM.YYYY date format.
|
|
Filters out duplicates (already imported by reference_no).
|
|
Returns: {success, transactions: [...], total_fetched, skipped_duplicates}
|
|
"""
|
|
try:
|
|
card_account_number = card_account_number.strip()
|
|
|
|
# Verify card exists
|
|
if not frappe.db.exists("Kapital Bank Card", card_account_number):
|
|
return {"success": False, "message": f"Card {card_account_number} not found in registry."}
|
|
|
|
# Card statements are limited to 90 days by the API
|
|
from frappe.utils import date_diff, getdate
|
|
if date_diff(getdate(to_date), getdate(from_date)) > 90:
|
|
return {"success": False, "message": "Card statement date range cannot exceed 90 days."}
|
|
|
|
pan = frappe.db.get_value("Kapital Bank Card", card_account_number, "pan") or card_account_number
|
|
|
|
# Fetch card statement (date format: YYYY-MM-DD, raw Frappe date)
|
|
client = BIRBankClient(login_name)
|
|
resp = client.get(
|
|
"/v2/statement/card",
|
|
params={
|
|
"accountNumber": card_account_number,
|
|
"fromDate": str(from_date) if from_date else None,
|
|
"toDate": str(to_date) if to_date else None,
|
|
}
|
|
)
|
|
if not resp.ok:
|
|
return {"success": False, "message": _api_error_message(resp)}
|
|
|
|
data = resp.json()
|
|
resp_code = data.get("response", {}).get("code", "?")
|
|
if resp_code != "0":
|
|
resp_msg = data.get("response", {}).get("message", "")
|
|
return {"success": False, "message": resp_msg or f"API error code={resp_code}"}
|
|
|
|
# Card statements use responseData.operation[] (not operations.statementList[])
|
|
operation_list = data.get("responseData", {}).get("operation", [])
|
|
|
|
transactions = []
|
|
skipped_duplicates = 0
|
|
|
|
for txn in operation_list:
|
|
txn_id = str(txn.get("id") or "").strip()
|
|
|
|
# Deduplication via tracking doctype
|
|
if txn_id and frappe.db.exists("Kapital Bank Transaction", {"reference_no": txn_id}):
|
|
skipped_duplicates += 1
|
|
continue
|
|
|
|
# Parse date (DD.MM.YYYY format)
|
|
parsed_date = _parse_card_txn_date(txn.get("operationDate"))
|
|
if not parsed_date:
|
|
continue
|
|
|
|
# Amount and currency
|
|
amount = abs(flt(txn.get("originalAmount") or txn.get("amount") or 0))
|
|
currency = (txn.get("originalCurrencyStr") or txn.get("currencyStr") or "AZN").strip()
|
|
|
|
# Direction: DEBIT/CREDIT → D/C
|
|
direction = (txn.get("direction") or "").upper()
|
|
dr_cr = "D" if direction == "DEBIT" else "C"
|
|
|
|
# Purpose (operationName serves as both purpose and counterparty description)
|
|
operation_name = (txn.get("operationName") or "").strip()
|
|
|
|
transactions.append({
|
|
"ref_no": txn_id,
|
|
"date": str(parsed_date),
|
|
"counterparty": "",
|
|
"contr_voen": "",
|
|
"amount": amount,
|
|
"currency": currency,
|
|
"drcr": dr_cr,
|
|
"purpose": operation_name,
|
|
"source_type": "card",
|
|
"source_label": f"Card: {pan}",
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"transactions": transactions,
|
|
"total_fetched": len(operation_list),
|
|
"skipped_duplicates": skipped_duplicates,
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(f"get_card_statement_transactions: {e}\n{frappe.get_traceback()}", "Kapital Bank Import Card Statement")
|
|
return {"success": False, "message": str(e)}
|
|
|
|
|
|
def _build_import_lookups(settings):
|
|
"""Build lookup dicts once for bulk import."""
|
|
purpose_rules = []
|
|
for row in settings.transaction_mappings:
|
|
if not row.paid_from or not row.paid_to:
|
|
continue
|
|
if not row.purpose_keyword and not row.counterparty and not row.counterparty_type:
|
|
continue
|
|
|
|
keyword_lower = ""
|
|
if row.purpose_keyword:
|
|
keyword = frappe.db.get_value("Kapital Bank Purpose", row.purpose_keyword, "purpose_keyword") or ""
|
|
keyword_lower = keyword.lower()
|
|
|
|
cp_name = ""
|
|
cp_voen = ""
|
|
if row.counterparty and row.counterparty_type:
|
|
if row.counterparty_type == "Kapital Bank Customer":
|
|
cp_data = frappe.db.get_value("Kapital Bank Customer", row.counterparty, ["customer_name", "tax_id"], as_dict=True)
|
|
else:
|
|
cp_data = frappe.db.get_value("Kapital Bank Supplier", row.counterparty, ["supplier_name", "tax_id"], as_dict=True)
|
|
if cp_data:
|
|
cp_name = (cp_data.get("customer_name") or cp_data.get("supplier_name") or "").strip()
|
|
cp_voen = (cp_data.get("tax_id") or "").strip()
|
|
|
|
purpose_rules.append((
|
|
keyword_lower,
|
|
row.paid_from,
|
|
row.paid_to,
|
|
getattr(row, "payment_type", None) or "",
|
|
cp_name,
|
|
cp_voen,
|
|
getattr(row, "counterparty_type", None) or "",
|
|
getattr(row, "document_type", None) or "Payment Entry",
|
|
getattr(row, "cost_center", None) or "",
|
|
bool(getattr(row, "multi_currency", 0)),
|
|
))
|
|
|
|
# Each key maps to a dict with optional "Customer" and "Supplier" entries.
|
|
# This preserves both when the same name/VOEN exists in both registries,
|
|
# allowing payment_type-aware selection at import time.
|
|
voen_to_party = {}
|
|
name_to_party = {}
|
|
|
|
for row in settings.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:
|
|
kb_customer_name = frappe.db.get_value("Kapital Bank Customer", row.kb_customer_name, "customer_name")
|
|
if kb_customer_name:
|
|
name_to_party.setdefault(kb_customer_name.strip(), {})["Customer"] = entry
|
|
|
|
for row in settings.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:
|
|
kb_supplier_name = frappe.db.get_value("Kapital Bank Supplier", row.kb_supplier_name, "supplier_name")
|
|
if kb_supplier_name:
|
|
name_to_party.setdefault(kb_supplier_name.strip(), {})["Supplier"] = entry
|
|
|
|
return purpose_rules, voen_to_party, name_to_party
|
|
|
|
|
|
def _import_one_transaction(txn_data, settings, purpose_rules, purpose_threshold,
|
|
voen_to_party, name_to_party):
|
|
"""Import a single transaction using pre-built lookups. Returns result dict."""
|
|
ref_no = txn_data.get("ref_no", "")
|
|
parsed_date = txn_data.get("date")
|
|
amount = abs(flt(txn_data.get("amount", 0)))
|
|
dr_cr = (txn_data.get("drcr") or "D").upper()
|
|
payment_type = "Pay" if dr_cr == "D" else "Receive"
|
|
purpose = (txn_data.get("purpose") or "").strip()
|
|
contr_name = (txn_data.get("counterparty") or "").strip()
|
|
contr_voen = (txn_data.get("contr_voen") or "").strip()
|
|
source_type = txn_data.get("source_type", "account")
|
|
|
|
# Purpose mapping with fuzzy matching — priority: both > purpose-only > counterparty-only
|
|
purpose_lower = purpose.lower()
|
|
paid_from = paid_to = None
|
|
|
|
both_rules = []
|
|
purpose_only_rules = []
|
|
counterparty_only_rules = []
|
|
fallback_rules = []
|
|
for rule in purpose_rules:
|
|
keyword, pf, pt, rule_pt, cp_name, cp_voen, cp_type, doc_type, rule_cc, rule_mc = rule
|
|
has_keyword = bool(keyword)
|
|
has_cp = bool(cp_name or cp_voen)
|
|
if has_keyword and has_cp:
|
|
both_rules.append(rule)
|
|
elif has_keyword:
|
|
purpose_only_rules.append(rule)
|
|
elif has_cp:
|
|
counterparty_only_rules.append(rule)
|
|
elif cp_type:
|
|
fallback_rules.append(rule)
|
|
|
|
def _counterparty_matches(cp_name, cp_voen):
|
|
if cp_voen and contr_voen and cp_voen == contr_voen:
|
|
return True
|
|
if cp_name and contr_name and cp_name == contr_name:
|
|
return True
|
|
return False
|
|
|
|
def _match_purpose_rules(rules, check_counterparty):
|
|
best_score = 0
|
|
best_pf = best_pt = best_doc_type = best_cc = best_mc = None
|
|
for keyword, pf, pt, rule_pt, cp_name, cp_voen, cp_type, doc_type, rule_cc, rule_mc in rules:
|
|
if rule_pt and rule_pt != payment_type:
|
|
continue
|
|
if check_counterparty and not _counterparty_matches(cp_name, cp_voen):
|
|
continue
|
|
if keyword:
|
|
if keyword in purpose_lower:
|
|
return pf, pt, doc_type, rule_cc, rule_mc
|
|
score = SequenceMatcher(None, purpose_lower, keyword).ratio()
|
|
if score > best_score and score >= purpose_threshold:
|
|
best_score = score
|
|
best_pf, best_pt, best_doc_type, best_cc, best_mc = pf, pt, doc_type, rule_cc, rule_mc
|
|
else:
|
|
return pf, pt, doc_type, rule_cc, rule_mc
|
|
return best_pf, best_pt, best_doc_type, best_cc, best_mc
|
|
|
|
doc_type = "Payment Entry"
|
|
cost_center = ""
|
|
multi_currency = False
|
|
|
|
# Phase 1: rules with BOTH purpose keyword AND counterparty
|
|
paid_from, paid_to, matched_doc_type, matched_cc, matched_mc = _match_purpose_rules(both_rules, check_counterparty=True)
|
|
if matched_doc_type:
|
|
doc_type = matched_doc_type
|
|
cost_center = matched_cc or ""
|
|
multi_currency = matched_mc or False
|
|
|
|
# Phase 2: rules with only purpose keyword
|
|
if not paid_from or not paid_to:
|
|
paid_from, paid_to, matched_doc_type, matched_cc, matched_mc = _match_purpose_rules(purpose_only_rules, check_counterparty=False)
|
|
if matched_doc_type:
|
|
doc_type = matched_doc_type
|
|
cost_center = matched_cc or ""
|
|
multi_currency = matched_mc or False
|
|
|
|
# Phase 3: rules with only counterparty
|
|
if not paid_from or not paid_to:
|
|
paid_from, paid_to, matched_doc_type, matched_cc, matched_mc = _match_purpose_rules(counterparty_only_rules, check_counterparty=True)
|
|
if matched_doc_type:
|
|
doc_type = matched_doc_type
|
|
cost_center = matched_cc or ""
|
|
multi_currency = matched_mc or False
|
|
|
|
# Phase 4: fallback rules — counterparty_type only, no specific counterparty
|
|
if not paid_from or not paid_to:
|
|
inferred_cp_type = "Customer" if payment_type == "Receive" else "Supplier"
|
|
for keyword, pf, pt, rule_pt, cp_name, cp_voen, cp_type, doc_type, rule_cc, rule_mc in fallback_rules:
|
|
if rule_pt and rule_pt != payment_type:
|
|
continue
|
|
if cp_type and cp_type != inferred_cp_type:
|
|
continue
|
|
paid_from, paid_to = pf, pt
|
|
doc_type = doc_type
|
|
cost_center = rule_cc or ""
|
|
multi_currency = rule_mc or False
|
|
break
|
|
|
|
if not paid_from or not paid_to:
|
|
return {
|
|
"success": False,
|
|
"error_type": "unmapped_purpose",
|
|
"message": f"No purpose mapping for: {purpose}",
|
|
"unmatched_purpose": [{"purpose": purpose}],
|
|
"txn_data": txn_data,
|
|
}
|
|
|
|
# Party mapping: VOEN → name → None
|
|
# When a key maps to both Customer and Supplier, prefer the one matching payment_type.
|
|
def _resolve_party(candidates):
|
|
preferred = "Customer" if payment_type == "Receive" else "Supplier"
|
|
fallback = "Supplier" if preferred == "Customer" else "Customer"
|
|
entry = candidates.get(preferred) or candidates.get(fallback)
|
|
return (entry["party_type"], entry["erp_party"]) if entry else (None, None)
|
|
|
|
party_type = None
|
|
erp_party = None
|
|
if contr_voen and contr_voen in voen_to_party:
|
|
party_type, erp_party = _resolve_party(voen_to_party[contr_voen])
|
|
elif contr_name and contr_name in name_to_party:
|
|
party_type, erp_party = _resolve_party(name_to_party[contr_name])
|
|
|
|
if not party_type or not erp_party:
|
|
if source_type == "card" and not contr_name and not contr_voen:
|
|
return {
|
|
"success": False,
|
|
"error_type": "unmapped_party",
|
|
"message": f"Card transaction has no counterparty. Purpose: {purpose}",
|
|
"unmatched_party": [{"name": "(card transaction)", "voen": ""}],
|
|
"txn_data": txn_data,
|
|
}
|
|
return {
|
|
"success": False,
|
|
"error_type": "unmapped_party",
|
|
"message": f"No party mapping for: {contr_name} (VOEN: {contr_voen})",
|
|
"unmatched_party": [{"name": contr_name, "voen": contr_voen}],
|
|
"txn_data": txn_data,
|
|
}
|
|
|
|
if doc_type == "Journal Entry":
|
|
# Create Journal Entry
|
|
je = frappe.new_doc("Journal Entry")
|
|
je.voucher_type = "Bank Entry"
|
|
je.posting_date = parsed_date
|
|
je.company = settings.default_company
|
|
je.cheque_no = ref_no
|
|
je.cheque_date = parsed_date
|
|
je.user_remark = purpose
|
|
if multi_currency:
|
|
je.multi_currency = 1
|
|
|
|
# Party can only be set on Receivable/Payable accounts
|
|
party_accounts = {"Receivable", "Payable"}
|
|
pf_type = frappe.get_cached_value("Account", paid_from, "account_type") or ""
|
|
pt_type = frappe.get_cached_value("Account", paid_to, "account_type") or ""
|
|
pf_party = (party_type, erp_party) if pf_type in party_accounts else (None, None)
|
|
pt_party = (party_type, erp_party) if pt_type in party_accounts else (None, None)
|
|
|
|
je_row_extra = {}
|
|
if cost_center:
|
|
je_row_extra["cost_center"] = cost_center
|
|
|
|
if payment_type == "Pay":
|
|
je.append("accounts", {
|
|
"account": paid_from,
|
|
"credit_in_account_currency": amount,
|
|
"party_type": pf_party[0],
|
|
"party": pf_party[1],
|
|
**je_row_extra,
|
|
})
|
|
je.append("accounts", {
|
|
"account": paid_to,
|
|
"debit_in_account_currency": amount,
|
|
"party_type": pt_party[0],
|
|
"party": pt_party[1],
|
|
**je_row_extra,
|
|
})
|
|
else:
|
|
je.append("accounts", {
|
|
"account": paid_from,
|
|
"credit_in_account_currency": amount,
|
|
"party_type": pf_party[0],
|
|
"party": pf_party[1],
|
|
**je_row_extra,
|
|
})
|
|
je.append("accounts", {
|
|
"account": paid_to,
|
|
"debit_in_account_currency": amount,
|
|
"party_type": pt_party[0],
|
|
"party": pt_party[1],
|
|
**je_row_extra,
|
|
})
|
|
|
|
try:
|
|
je.insert(ignore_permissions=True)
|
|
je.submit()
|
|
except Exception as je_err:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"JE creation failed for ref {ref_no}: {je_err}\n{frappe.get_traceback()}",
|
|
"KB JE Import Error",
|
|
)
|
|
frappe.db.commit()
|
|
return {
|
|
"success": False,
|
|
"error_type": "import_error",
|
|
"message": str(je_err),
|
|
"txn_data": txn_data,
|
|
}
|
|
|
|
_create_transaction_record(
|
|
reference_no=ref_no,
|
|
posting_date=parsed_date,
|
|
amount=amount,
|
|
currency=txn_data.get("currency", "AZN"),
|
|
party=contr_name,
|
|
source_type="Account" if source_type == "account" else "Card",
|
|
source_label=txn_data.get("source_label", ""),
|
|
journal_entry=je.name,
|
|
commit=False,
|
|
)
|
|
|
|
return {"success": True, "journal_entry": je.name}
|
|
|
|
# Create Payment Entry
|
|
pe = frappe.new_doc("Payment Entry")
|
|
pe.payment_type = payment_type
|
|
pe.company = settings.default_company
|
|
pe.posting_date = parsed_date
|
|
pe.paid_from = paid_from
|
|
pe.paid_to = paid_to
|
|
pe.paid_amount = amount
|
|
pe.received_amount = amount
|
|
pe.reference_no = ref_no
|
|
pe.reference_date = parsed_date
|
|
pe.purpose = purpose
|
|
pe.remarks = purpose
|
|
pe.party_type = party_type
|
|
pe.party = erp_party
|
|
if cost_center:
|
|
pe.cost_center = cost_center
|
|
|
|
try:
|
|
pe.insert(ignore_permissions=True)
|
|
pe.submit()
|
|
except Exception as pe_err:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"PE creation failed for ref {ref_no}: {pe_err}\n{frappe.get_traceback()}",
|
|
"KB PE Import Error",
|
|
)
|
|
frappe.db.commit()
|
|
return {
|
|
"success": False,
|
|
"error_type": "import_error",
|
|
"message": str(pe_err),
|
|
"txn_data": txn_data,
|
|
}
|
|
|
|
# Create tracking record (no individual commit — bulk commit at the end)
|
|
_create_transaction_record(
|
|
reference_no=ref_no,
|
|
posting_date=parsed_date,
|
|
amount=amount,
|
|
currency=txn_data.get("currency", "AZN"),
|
|
party=contr_name,
|
|
source_type="Account" if source_type == "account" else "Card",
|
|
source_label=txn_data.get("source_label", ""),
|
|
payment_entry=pe.name,
|
|
commit=False,
|
|
)
|
|
|
|
return {"success": True, "payment_entry": pe.name}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def import_bulk_transactions(txn_list):
|
|
"""Enqueue bulk import as a background job for realtime progress via Socket.IO."""
|
|
if isinstance(txn_list, str):
|
|
txn_list = json.loads(txn_list)
|
|
|
|
user = frappe.session.user
|
|
|
|
frappe.enqueue(
|
|
_process_bulk_import,
|
|
txn_list=txn_list,
|
|
user=user,
|
|
queue="default",
|
|
timeout=600,
|
|
)
|
|
|
|
return {"success": True, "enqueued": True, "total": len(txn_list)}
|
|
|
|
|
|
def _process_bulk_import(txn_list, user):
|
|
"""Background job: process all transactions with realtime progress."""
|
|
frappe.log_error(f"PE import job started: {len(txn_list)} transactions, user={user}", "KB PE Import Start")
|
|
frappe.db.commit()
|
|
|
|
try:
|
|
frappe.set_user(user)
|
|
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
purpose_threshold = flt(settings.similarity_threshold_purpose or 70) / 100.0
|
|
purpose_rules, voen_to_party, name_to_party = _build_import_lookups(settings)
|
|
|
|
total = len(txn_list)
|
|
imported_count = 0
|
|
errors = []
|
|
|
|
for idx, txn_data in enumerate(txn_list):
|
|
frappe.publish_realtime(
|
|
"kb_import_progress",
|
|
{"current": idx + 1, "total": total},
|
|
user=user,
|
|
)
|
|
|
|
result = _import_one_transaction(
|
|
txn_data, settings, purpose_rules, purpose_threshold,
|
|
voen_to_party, name_to_party,
|
|
)
|
|
|
|
if result.get("success"):
|
|
imported_count += 1
|
|
else:
|
|
errors.append(result)
|
|
|
|
# Commit after each successful transaction so progress is durable
|
|
frappe.db.commit()
|
|
|
|
frappe.log_error(
|
|
f"PE import finished: {imported_count}/{total} imported, {len(errors)} errors",
|
|
"KB PE Import Complete",
|
|
)
|
|
frappe.db.commit()
|
|
|
|
frappe.publish_realtime(
|
|
"kb_import_complete",
|
|
{"total": total, "imported": imported_count, "errors": errors},
|
|
user=user,
|
|
)
|
|
except Exception:
|
|
frappe.db.rollback()
|
|
frappe.log_error(
|
|
f"Bulk PE import failed: {frappe.get_traceback()}",
|
|
"KB PE Import Error",
|
|
)
|
|
frappe.db.commit()
|
|
|
|
|
|
@frappe.whitelist()
|
|
def import_single_transaction(txn_data, account_iban, login_name=None):
|
|
"""Import a single transaction (legacy endpoint, kept for compatibility)."""
|
|
try:
|
|
if isinstance(txn_data, str):
|
|
txn_data = json.loads(txn_data)
|
|
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
purpose_threshold = flt(settings.similarity_threshold_purpose or 70) / 100.0
|
|
purpose_rules, voen_to_party, name_to_party = _build_import_lookups(settings)
|
|
|
|
result = _import_one_transaction(
|
|
txn_data, settings, purpose_rules, purpose_threshold,
|
|
voen_to_party, name_to_party,
|
|
)
|
|
if result.get("success"):
|
|
frappe.db.commit()
|
|
return result
|
|
|
|
except Exception as e:
|
|
frappe.local.message_log = []
|
|
frappe.log_error(
|
|
f"import_single_transaction failed: {e}\n{frappe.get_traceback()}",
|
|
"KB Single Import Error",
|
|
)
|
|
frappe.db.commit()
|
|
return {"success": False, "error_type": "import_error", "message": str(e)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# TRANSACTION TRACKING
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def _create_transaction_record(reference_no, posting_date, amount, currency,
|
|
party, source_type, source_label, payment_entry=None,
|
|
journal_entry=None, bank_transaction=None, commit=True):
|
|
"""Create a Kapital Bank Transaction record to track imported transaction."""
|
|
reference_no = str(reference_no).strip()
|
|
if not reference_no:
|
|
return
|
|
|
|
existing = frappe.db.get_value("Kapital Bank Transaction", {"reference_no": reference_no}, "name")
|
|
if existing:
|
|
return existing
|
|
|
|
try:
|
|
doc = frappe.new_doc("Kapital Bank Transaction")
|
|
doc.reference_no = reference_no
|
|
doc.posting_date = posting_date
|
|
doc.amount = amount
|
|
doc.currency = currency
|
|
doc.party = party
|
|
doc.source_type = source_type
|
|
doc.source_label = source_label
|
|
if payment_entry:
|
|
doc.payment_entry = payment_entry
|
|
if journal_entry:
|
|
doc.journal_entry = journal_entry
|
|
if bank_transaction:
|
|
doc.bank_transaction = bank_transaction
|
|
doc.status = "Imported"
|
|
doc.import_date = frappe.utils.now_datetime()
|
|
doc.insert(ignore_permissions=True)
|
|
if commit:
|
|
frappe.db.commit()
|
|
return doc.name
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"_create_transaction_record failed for {reference_no}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Tracking",
|
|
)
|
|
|
|
|
|
def on_delete_payment_entry(doc, method):
|
|
"""Delete linked Kapital Bank Transaction records when Payment Entry is deleted."""
|
|
try:
|
|
records = frappe.get_all(
|
|
"Kapital Bank Transaction",
|
|
filters={"payment_entry": doc.name},
|
|
fields=["name"],
|
|
)
|
|
for record in records:
|
|
frappe.delete_doc("Kapital Bank Transaction", record.name, ignore_permissions=True, force=True)
|
|
if records:
|
|
frappe.db.commit()
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting Kapital Bank Transaction for PE {doc.name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Cleanup",
|
|
)
|
|
|
|
|
|
def on_cancel_payment_entry(doc, method):
|
|
"""Delete linked Kapital Bank Transaction records when Payment Entry is cancelled."""
|
|
try:
|
|
records = frappe.get_all(
|
|
"Kapital Bank Transaction",
|
|
filters={"payment_entry": doc.name},
|
|
fields=["name"],
|
|
)
|
|
for record in records:
|
|
frappe.db.delete("Kapital Bank Transaction", {"name": record.name})
|
|
if records:
|
|
frappe.db.commit()
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting Kapital Bank Transaction for cancelled PE {doc.name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Cleanup",
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# JOURNAL ENTRY CLEANUP
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def on_delete_journal_entry(doc, method):
|
|
"""Delete linked Kapital Bank Transaction records when Journal Entry is deleted."""
|
|
try:
|
|
records = frappe.get_all(
|
|
"Kapital Bank Transaction",
|
|
filters={"journal_entry": doc.name},
|
|
fields=["name"],
|
|
)
|
|
for record in records:
|
|
frappe.delete_doc("Kapital Bank Transaction", record.name, ignore_permissions=True, force=True)
|
|
if records:
|
|
frappe.db.commit()
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting Kapital Bank Transaction for JE {doc.name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Cleanup",
|
|
)
|
|
|
|
|
|
def on_cancel_journal_entry(doc, method):
|
|
"""Delete linked Kapital Bank Transaction records when Journal Entry is cancelled."""
|
|
try:
|
|
records = frappe.get_all(
|
|
"Kapital Bank Transaction",
|
|
filters={"journal_entry": doc.name},
|
|
fields=["name"],
|
|
)
|
|
for record in records:
|
|
frappe.db.delete("Kapital Bank Transaction", {"name": record.name})
|
|
if records:
|
|
frappe.db.commit()
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting Kapital Bank Transaction for cancelled JE {doc.name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Cleanup",
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# BANK TRANSACTION CLEANUP
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def on_delete_bank_transaction(doc, method):
|
|
"""Delete linked Kapital Bank Transaction records when Bank Transaction is deleted."""
|
|
try:
|
|
records = frappe.get_all(
|
|
"Kapital Bank Transaction",
|
|
filters={"bank_transaction": doc.name},
|
|
fields=["name"],
|
|
)
|
|
for record in records:
|
|
frappe.delete_doc("Kapital Bank Transaction", record.name, ignore_permissions=True, force=True)
|
|
if records:
|
|
frappe.db.commit()
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting Kapital Bank Transaction for BT {doc.name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Cleanup",
|
|
)
|
|
|
|
|
|
def on_cancel_bank_transaction(doc, method):
|
|
"""Delete linked Kapital Bank Transaction records when Bank Transaction is cancelled."""
|
|
try:
|
|
records = frappe.get_all(
|
|
"Kapital Bank Transaction",
|
|
filters={"bank_transaction": doc.name},
|
|
fields=["name"],
|
|
)
|
|
for record in records:
|
|
frappe.db.delete("Kapital Bank Transaction", {"name": record.name})
|
|
if records:
|
|
frappe.db.commit()
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting Kapital Bank Transaction for cancelled BT {doc.name}: {e}\n{frappe.get_traceback()}",
|
|
"Kapital Bank Transaction Cleanup",
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# BANK TRANSACTION IMPORT (simpler path — no purpose/party mappings required)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def _build_bank_account_lookup(settings):
|
|
"""Build (IBAN/card-account-number, currency) → Bank Account lookup from settings mappings.
|
|
|
|
Also builds a fallback IBAN-only lookup for cases where currency is not specified in mapping.
|
|
Returns: {(source_id, currency): bank_account, ...}, {source_id: bank_account, ...}
|
|
"""
|
|
ba_lookup_by_currency = {}
|
|
ba_lookup_fallback = {}
|
|
|
|
for row in settings.account_mappings:
|
|
if row.iban and row.bank_account:
|
|
iban = row.iban.strip()
|
|
ccy = (row.currency or "").strip().upper()
|
|
if ccy:
|
|
ba_lookup_by_currency[(iban, ccy)] = row.bank_account
|
|
ba_lookup_fallback[iban] = row.bank_account
|
|
|
|
for row in settings.card_mappings:
|
|
if row.account_number and row.bank_account:
|
|
acc_no = row.account_number.strip()
|
|
ccy = (getattr(row, "currency", "") or "").strip().upper()
|
|
if ccy:
|
|
ba_lookup_by_currency[(acc_no, ccy)] = row.bank_account
|
|
ba_lookup_fallback[acc_no] = row.bank_account
|
|
|
|
return ba_lookup_by_currency, ba_lookup_fallback
|
|
|
|
|
|
def _import_one_bank_transaction(txn_data, settings, voen_to_party, name_to_party,
|
|
ba_lookup_by_currency, ba_lookup_fallback):
|
|
"""Import a single transaction as a Bank Transaction document. Returns result dict."""
|
|
ref_no = txn_data.get("ref_no", "")
|
|
parsed_date = txn_data.get("date")
|
|
amount = abs(flt(txn_data.get("amount", 0)))
|
|
dr_cr = (txn_data.get("drcr") or "D").upper()
|
|
purpose = (txn_data.get("purpose") or "").strip()
|
|
contr_name = (txn_data.get("counterparty") or "").strip()
|
|
contr_voen = (txn_data.get("contr_voen") or "").strip()
|
|
currency = (txn_data.get("currency") or "AZN").strip().upper()
|
|
source_type = txn_data.get("source_type", "account")
|
|
source_label = txn_data.get("source_label", "")
|
|
|
|
# Determine bank_account from source identifier + currency
|
|
bank_account = None
|
|
source_keys = []
|
|
if source_type == "account":
|
|
source_id = txn_data.get("source_id", "")
|
|
if source_id:
|
|
source_keys.append(source_id.strip())
|
|
if source_label:
|
|
source_keys.append(source_label.strip())
|
|
elif source_type == "card":
|
|
source_id = txn_data.get("source_id", "")
|
|
if source_id:
|
|
source_keys.append(source_id.strip())
|
|
if source_label:
|
|
card_num = source_label.replace("Card: ", "").strip()
|
|
source_keys.append(card_num)
|
|
|
|
# Try currency-specific lookup first, then fallback
|
|
for key in source_keys:
|
|
bank_account = ba_lookup_by_currency.get((key, currency))
|
|
if bank_account:
|
|
break
|
|
if not bank_account:
|
|
for key in source_keys:
|
|
bank_account = ba_lookup_fallback.get(key)
|
|
if bank_account:
|
|
break
|
|
|
|
if not bank_account:
|
|
source_display = source_keys[0] if source_keys else source_label
|
|
return {
|
|
"success": False,
|
|
"error_type": "unmapped_bank_account",
|
|
"message": f"No bank account mapping for {source_display}",
|
|
"txn_data": txn_data,
|
|
}
|
|
|
|
# Use the bank account's own currency to avoid mismatch errors in Frappe
|
|
gl_account = frappe.db.get_value("Bank Account", bank_account, "account")
|
|
if gl_account:
|
|
currency = frappe.db.get_value("Account", gl_account, "account_currency") or currency
|
|
|
|
# Optional party mapping: VÖEN → name → None
|
|
# C (credit, money in) → prefer Customer; D (debit, money out) → prefer Supplier
|
|
def _resolve_party(candidates):
|
|
preferred = "Customer" if dr_cr == "C" else "Supplier"
|
|
fallback = "Supplier" if preferred == "Customer" else "Customer"
|
|
entry = candidates.get(preferred) or candidates.get(fallback)
|
|
return (entry["party_type"], entry["erp_party"]) if entry else (None, None)
|
|
|
|
party_type = None
|
|
erp_party = None
|
|
if contr_voen and contr_voen in voen_to_party:
|
|
party_type, erp_party = _resolve_party(voen_to_party[contr_voen])
|
|
elif contr_name and contr_name in name_to_party:
|
|
party_type, erp_party = _resolve_party(name_to_party[contr_name])
|
|
|
|
# Create Bank Transaction
|
|
bt = frappe.new_doc("Bank Transaction")
|
|
bt.date = parsed_date
|
|
bt.deposit = amount if dr_cr == "C" else 0
|
|
bt.withdrawal = amount if dr_cr == "D" else 0
|
|
bt.currency = currency
|
|
bt.description = purpose
|
|
bt.reference_number = ref_no
|
|
bt.transaction_id = ref_no
|
|
bt.company = settings.default_company
|
|
bt.bank_party_name = contr_name or purpose
|
|
|
|
bt.bank_account = bank_account
|
|
|
|
kb_currency = (txn_data.get("kb_currency") or "").strip()
|
|
if kb_currency:
|
|
bt.kb_currency = kb_currency
|
|
|
|
if party_type and erp_party:
|
|
bt.party_type = party_type
|
|
bt.party = erp_party
|
|
|
|
try:
|
|
bt.insert(ignore_permissions=True)
|
|
bt.submit()
|
|
except Exception as bt_err:
|
|
frappe.db.rollback()
|
|
frappe.local.message_log = []
|
|
return {
|
|
"success": False,
|
|
"error_type": "import_error",
|
|
"message": str(bt_err),
|
|
"txn_data": txn_data,
|
|
}
|
|
|
|
# Create tracking record
|
|
_create_transaction_record(
|
|
reference_no=ref_no,
|
|
posting_date=parsed_date,
|
|
amount=amount,
|
|
currency=currency,
|
|
party=contr_name,
|
|
source_type="Account" if source_type == "account" else "Card",
|
|
source_label=source_label,
|
|
bank_transaction=bt.name,
|
|
commit=False,
|
|
)
|
|
|
|
return {"success": True, "bank_transaction": bt.name}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def import_bulk_bank_transactions(txn_list):
|
|
"""Enqueue bulk Bank Transaction import as a background job."""
|
|
if isinstance(txn_list, str):
|
|
txn_list = json.loads(txn_list)
|
|
|
|
user = frappe.session.user
|
|
|
|
frappe.enqueue(
|
|
_process_bulk_bt_import,
|
|
txn_list=txn_list,
|
|
user=user,
|
|
queue="default",
|
|
timeout=600,
|
|
)
|
|
|
|
return {"success": True, "enqueued": True, "total": len(txn_list)}
|
|
|
|
|
|
def _process_bulk_bt_import(txn_list, user):
|
|
"""Background job: import transactions as Bank Transaction with realtime progress."""
|
|
frappe.log_error(f"BT import job started: {len(txn_list)} transactions, user={user}", "KB BT Import Start")
|
|
frappe.db.commit()
|
|
|
|
try:
|
|
frappe.set_user(user)
|
|
|
|
settings = frappe.get_single("Kapital Bank Settings")
|
|
_purpose_rules, voen_to_party, name_to_party = _build_import_lookups(settings)
|
|
ba_lookup_by_currency, ba_lookup_fallback = _build_bank_account_lookup(settings)
|
|
|
|
total = len(txn_list)
|
|
imported_count = 0
|
|
errors = []
|
|
|
|
for idx, txn_data in enumerate(txn_list):
|
|
frappe.publish_realtime(
|
|
"kb_bt_import_progress",
|
|
{"current": idx + 1, "total": total},
|
|
user=user,
|
|
)
|
|
|
|
result = _import_one_bank_transaction(
|
|
txn_data, settings, voen_to_party, name_to_party,
|
|
ba_lookup_by_currency, ba_lookup_fallback,
|
|
)
|
|
|
|
if result.get("success"):
|
|
imported_count += 1
|
|
else:
|
|
errors.append(result)
|
|
|
|
frappe.db.commit()
|
|
|
|
error_sample = "\n".join(
|
|
f" [{e.get('error_type')}] {e.get('message', '')[:200]}"
|
|
for e in errors[:10]
|
|
)
|
|
frappe.log_error(
|
|
f"BT import finished: {imported_count}/{total} imported, {len(errors)} errors\n"
|
|
f"First errors:\n{error_sample}",
|
|
"KB BT Import Complete",
|
|
)
|
|
frappe.db.commit()
|
|
|
|
frappe.publish_realtime(
|
|
"kb_bt_import_complete",
|
|
{"total": total, "imported": imported_count, "errors": errors},
|
|
user=user,
|
|
)
|
|
except Exception:
|
|
frappe.db.rollback()
|
|
frappe.log_error(
|
|
f"Bulk BT import failed: {frappe.get_traceback()}",
|
|
"KB BT Import Error",
|
|
)
|
|
frappe.db.commit()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# INTERNAL HELPERS
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
_AZERI_MAP = str.maketrans("ƏəÜüÖöĞğİıÇ窺", "EeUuOoGgIiCcSs")
|
|
|
|
def _normalize(text, consider_azeri=True):
|
|
"""Normalize text for similarity comparison.
|
|
Optionally replaces Azerbaijani characters before lowercasing."""
|
|
if not text:
|
|
return ""
|
|
s = str(text)
|
|
if consider_azeri:
|
|
s = s.translate(_AZERI_MAP)
|
|
return " ".join(s.lower().split())
|
|
|
|
|
|
def _fmt_date(d, sep='-'):
|
|
"""Convert YYYY-MM-DD string or date object → DD{sep}MM{sep}YYYY.
|
|
|
|
Account statement → sep='-' → DD-MM-YYYY
|
|
Card statement → sep='.' → DD.MM.YYYY
|
|
"""
|
|
if not d:
|
|
return None
|
|
if hasattr(d, 'strftime'):
|
|
return d.strftime(f'%d{sep}%m{sep}%Y')
|
|
parts = str(d).split('-')
|
|
if len(parts) == 3:
|
|
return f"{parts[2]}{sep}{parts[1]}{sep}{parts[0]}"
|
|
return str(d)
|