mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-29 06:38:24 +00:00
This commit is contained in:
mergify[bot]
@@ -27,6 +27,7 @@
|
||||
"column_break_12",
|
||||
"branch_code",
|
||||
"bank_account_no",
|
||||
"statement_password",
|
||||
"address_and_contact",
|
||||
"address_html",
|
||||
"column_break_13",
|
||||
@@ -149,6 +150,12 @@
|
||||
"label": "Bank Account No",
|
||||
"length": 30
|
||||
},
|
||||
{
|
||||
"description": "Password used to open password-protected PDF statements for this account. Stored encrypted.",
|
||||
"fieldname": "statement_password",
|
||||
"fieldtype": "Password",
|
||||
"label": "Statement PDF Password"
|
||||
},
|
||||
{
|
||||
"fieldname": "address_and_contact",
|
||||
"fieldtype": "Section Break",
|
||||
|
||||
@@ -41,6 +41,7 @@ class BankAccount(Document):
|
||||
mask: DF.Data | None
|
||||
party: DF.DynamicLink | None
|
||||
party_type: DF.Link | None
|
||||
statement_password: DF.Password | None
|
||||
# end: auto-generated types
|
||||
|
||||
def onload(self):
|
||||
|
||||
@@ -28,7 +28,8 @@
|
||||
"detected_transaction_starting_index",
|
||||
"detected_transaction_ending_index",
|
||||
"section_break_yulq",
|
||||
"column_mapping"
|
||||
"column_mapping",
|
||||
"pdf_tables"
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
@@ -128,6 +129,13 @@
|
||||
"label": "Column Mapping",
|
||||
"options": "Bank Statement Import Log Column Map"
|
||||
},
|
||||
{
|
||||
"description": "Per-table extraction data for PDF statements (rows, bbox, page image, column mapping). Edited via the banking app.",
|
||||
"fieldname": "pdf_tables",
|
||||
"fieldtype": "JSON",
|
||||
"label": "PDF Tables",
|
||||
"read_only": 1
|
||||
},
|
||||
{
|
||||
"default": "Not Started",
|
||||
"fieldname": "status",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -7,7 +7,18 @@ from frappe.utils import getdate
|
||||
|
||||
from erpnext.accounts.doctype.bank_statement_import_log.bank_statement_import_log import (
|
||||
BankStatementImportLog,
|
||||
build_table_transactions,
|
||||
detect_column_mapping,
|
||||
detect_header_row,
|
||||
extract_pdf_tables,
|
||||
get_float_amount,
|
||||
get_statement_details,
|
||||
guess_column_mapping_by_content,
|
||||
reextract_pdf_table,
|
||||
set_header_index,
|
||||
set_pdf_table_header,
|
||||
update_column_mapping,
|
||||
update_pdf_tables,
|
||||
)
|
||||
from erpnext.accounts.test.accounts_mixin import AccountsTestMixin
|
||||
from erpnext.tests.utils import ERPNextTestSuite
|
||||
@@ -113,6 +124,346 @@ class TestBankStatementImportLog(ERPNextTestSuite, AccountsTestMixin):
|
||||
self.assertIsNone(get_float_amount("ABCD"))
|
||||
self.assertIsNone(get_float_amount("****"))
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# PDF statement import
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
@staticmethod
|
||||
def _make_pdf(html: str) -> bytes:
|
||||
import pdfkit
|
||||
|
||||
return pdfkit.from_string(html, False)
|
||||
|
||||
@staticmethod
|
||||
def _encrypt(pdf_bytes: bytes, password: str) -> bytes:
|
||||
import io
|
||||
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
|
||||
reader = PdfReader(io.BytesIO(pdf_bytes))
|
||||
writer = PdfWriter()
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
writer.encrypt(password)
|
||||
buffer = io.BytesIO()
|
||||
writer.write(buffer)
|
||||
return buffer.getvalue()
|
||||
|
||||
@staticmethod
|
||||
def _auto_map(table: dict) -> dict:
|
||||
"""Mimic prepare_pdf_tables' best-effort mapping for a single extracted table."""
|
||||
header_index, score = detect_header_row(table["rows"])
|
||||
if score >= 2:
|
||||
table["header_index"] = header_index
|
||||
table["column_mapping"] = detect_column_mapping(table["rows"][header_index])
|
||||
else:
|
||||
table["header_index"] = None
|
||||
table["column_mapping"] = guess_column_mapping_by_content(table["rows"])
|
||||
table["included"] = True
|
||||
return table
|
||||
|
||||
def test_pdf_multi_page_kept_separate_and_unioned(self):
|
||||
"""Tables on separate pages must NOT be merged; transactions are the union."""
|
||||
html = """
|
||||
<html><body>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Withdrawal</th><th>Deposit</th><th>Balance</th></tr>
|
||||
<tr><td>01/04/2024</td><td>UPI PAYMENT</td><td>500.00</td><td></td><td>9500.00</td></tr>
|
||||
<tr><td>03/04/2024</td><td>SALARY</td><td></td><td>20000.00</td><td>29500.00</td></tr></table>
|
||||
<div style="page-break-before: always"></div>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Withdrawal</th><th>Deposit</th><th>Balance</th></tr>
|
||||
<tr><td>05/04/2024</td><td>ATM WDL</td><td>2000.00</td><td></td><td>27500.00</td></tr></table>
|
||||
</body></html>
|
||||
"""
|
||||
tables = extract_pdf_tables(self._make_pdf(html))
|
||||
|
||||
# Two separate tables, one per page
|
||||
self.assertEqual(len(tables), 2)
|
||||
self.assertEqual(sorted(t["page"] for t in tables), [1, 2])
|
||||
for table in tables:
|
||||
self.assertIn("bbox", table)
|
||||
self.assertEqual(len(table["bbox"]), 4)
|
||||
|
||||
union = []
|
||||
for table in tables:
|
||||
final, _df, _af = build_table_transactions(self._auto_map(table))
|
||||
union.extend(final)
|
||||
|
||||
self.assertEqual(len(union), 3)
|
||||
self.assertEqual(sorted(t["date"] for t in union), ["2024-04-01", "2024-04-03", "2024-04-05"])
|
||||
|
||||
def test_pdf_junk_table_excluded(self):
|
||||
"""A non-transactions table (ad/summary) should yield zero transactions."""
|
||||
ad_table = self._auto_map({"rows": [["Open a new account!", "Call 1800-XYZ"]]})
|
||||
final, _df, _af = build_table_transactions(ad_table)
|
||||
self.assertEqual(final, [])
|
||||
|
||||
def test_headerless_content_mapping(self):
|
||||
"""Without a header row, columns are guessed from their contents."""
|
||||
rows = [
|
||||
["01/04/2024", "UPI PAYMENT", "500.00"],
|
||||
["03/04/2024", "SALARY CREDIT", "20000.00"],
|
||||
]
|
||||
mapping = {
|
||||
c["maps_to"]: c["index"]
|
||||
for c in guess_column_mapping_by_content(rows)
|
||||
if c["maps_to"] != "Do not import"
|
||||
}
|
||||
self.assertEqual(mapping.get("Date"), 0)
|
||||
self.assertEqual(mapping.get("Description"), 1)
|
||||
self.assertEqual(mapping.get("Amount"), 2)
|
||||
|
||||
def test_pdf_password_protected(self):
|
||||
"""Encrypted PDFs error without a password and succeed with the right one."""
|
||||
html = """
|
||||
<html><body><table border="1">
|
||||
<tr><th>Date</th><th>Narration</th><th>Amount</th></tr>
|
||||
<tr><td>01/04/2024</td><td>UPI PAYMENT</td><td>500.00</td></tr></table></body></html>
|
||||
"""
|
||||
encrypted = self._encrypt(self._make_pdf(html), "secret123")
|
||||
|
||||
# No / wrong password -> recognizable error
|
||||
self.assertRaises(frappe.ValidationError, extract_pdf_tables, encrypted)
|
||||
self.assertRaises(frappe.ValidationError, extract_pdf_tables, encrypted, "wrong")
|
||||
|
||||
# Correct password -> extracts
|
||||
tables = extract_pdf_tables(encrypted, "secret123")
|
||||
self.assertTrue(tables)
|
||||
|
||||
def test_pdf_no_tables_detected(self):
|
||||
"""A PDF with no detectable tables raises a clear error (e.g. scanned PDFs)."""
|
||||
html = "<html><body><p>Just some prose with no tabular data at all.</p></body></html>"
|
||||
self.assertRaises(frappe.ValidationError, extract_pdf_tables, self._make_pdf(html))
|
||||
|
||||
def _create_pdf_import_log(self, html: str) -> BankStatementImportLog:
|
||||
pdf_bytes = self._make_pdf(html)
|
||||
file_doc = frappe.get_doc(
|
||||
{
|
||||
"doctype": "File",
|
||||
"file_name": f"test-statement-{frappe.generate_hash(length=8)}.pdf",
|
||||
"is_private": 1,
|
||||
"content": pdf_bytes,
|
||||
}
|
||||
).insert(ignore_permissions=True)
|
||||
|
||||
doc = frappe.get_doc(
|
||||
{
|
||||
"doctype": "Bank Statement Import Log",
|
||||
"name": f"test-pdf-{frappe.generate_hash(length=8)}",
|
||||
"bank_account": self.bank_account,
|
||||
"file": file_doc.file_url,
|
||||
}
|
||||
)
|
||||
return doc.insert()
|
||||
|
||||
def test_pdf_full_lifecycle(self):
|
||||
"""End-to-end doc lifecycle: insert -> rasterize -> preview -> edit -> import."""
|
||||
html = """
|
||||
<html><body>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Withdrawal</th><th>Deposit</th><th>Balance</th></tr>
|
||||
<tr><td>01/04/2024</td><td>UPI PAYMENT</td><td>500.00</td><td></td><td>9500.00</td></tr>
|
||||
<tr><td>03/04/2024</td><td>SALARY</td><td></td><td>20000.00</td><td>29500.00</td></tr></table>
|
||||
<div style="page-break-before: always"></div>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Withdrawal</th><th>Deposit</th><th>Balance</th></tr>
|
||||
<tr><td>05/04/2024</td><td>ATM WDL</td><td>2000.00</td><td></td><td>27500.00</td></tr></table>
|
||||
</body></html>
|
||||
"""
|
||||
doc = self._create_pdf_import_log(html)
|
||||
|
||||
# before_insert populated the per-table JSON, page images and the union summary
|
||||
tables = doc.get_pdf_tables()
|
||||
self.assertEqual(len(tables), 2)
|
||||
for table in tables:
|
||||
self.assertTrue(table.get("page_image"))
|
||||
self.assertIn("bbox", table)
|
||||
# Page-image File must be attached to the final docname, not the client's temp id
|
||||
attached_to = frappe.db.get_value("File", {"file_url": table["page_image"]}, "attached_to_name")
|
||||
self.assertEqual(attached_to, doc.name)
|
||||
self.assertEqual(doc.number_of_transactions, 3)
|
||||
self.assertEqual(doc.total_debit_transactions, 2)
|
||||
self.assertEqual(doc.total_credit_transactions, 1)
|
||||
|
||||
# get_statement_details returns the union and the per-table data for the editor
|
||||
details = get_statement_details(doc.name)
|
||||
self.assertEqual(len(details["final_transactions"]), 3)
|
||||
self.assertEqual(details["raw_data"], [])
|
||||
self.assertEqual(len(details["pdf_tables"]), 2)
|
||||
|
||||
# Excluding the second table (page 2) drops its single transaction
|
||||
tables[1]["included"] = False
|
||||
update_pdf_tables(doc.name, tables)
|
||||
doc.reload()
|
||||
self.assertEqual(doc.number_of_transactions, 2)
|
||||
|
||||
# Re-include and import; transactions are created for the union
|
||||
tables[1]["included"] = True
|
||||
update_pdf_tables(doc.name, tables)
|
||||
doc.reload()
|
||||
doc.insert_transactions()
|
||||
doc.reload()
|
||||
self.assertEqual(doc.status, "Completed")
|
||||
|
||||
created = frappe.get_all(
|
||||
"Bank Transaction", filters={"bank_account": self.bank_account, "docstatus": 1}
|
||||
)
|
||||
self.assertEqual(len(created), 3)
|
||||
|
||||
def test_pdf_reextract_table_from_bbox(self):
|
||||
"""Re-extracting a table from an adjusted bbox updates its rows and stores the bbox."""
|
||||
html = """
|
||||
<html><body>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Amount</th></tr>
|
||||
<tr><td>01/04/2024</td><td>UPI PAYMENT</td><td>500.00</td></tr>
|
||||
<tr><td>03/04/2024</td><td>SALARY</td><td>20000.00</td></tr></table>
|
||||
</body></html>
|
||||
"""
|
||||
doc = self._create_pdf_import_log(html)
|
||||
table = doc.get_pdf_tables()[0]
|
||||
bbox = table["bbox"]
|
||||
|
||||
details = reextract_pdf_table(doc.name, table["page"], table["table_index"], bbox)
|
||||
updated = details["pdf_tables"][0]
|
||||
|
||||
# Same region -> same rows; bbox is persisted
|
||||
self.assertTrue(updated["rows"])
|
||||
self.assertEqual(updated["bbox"], [round(float(v), 2) for v in bbox])
|
||||
self.assertEqual(updated["rows"], table["rows"])
|
||||
|
||||
def test_pdf_reextract_changed_bbox_updates_rows_and_transactions(self):
|
||||
"""Shrinking a table's bbox must drop rows and update the transaction count end-to-end."""
|
||||
html = """
|
||||
<html><body>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Amount</th></tr>
|
||||
<tr><td>01/04/2024</td><td>UPI PAYMENT</td><td>500.00</td></tr>
|
||||
<tr><td>03/04/2024</td><td>SALARY</td><td>20000.00</td></tr>
|
||||
<tr><td>05/04/2024</td><td>ATM WDL</td><td>2000.00</td></tr>
|
||||
<tr><td>07/04/2024</td><td>INTEREST</td><td>12.50</td></tr></table>
|
||||
</body></html>
|
||||
"""
|
||||
doc = self._create_pdf_import_log(html)
|
||||
original = doc.get_pdf_tables()[0]
|
||||
original_rows = len(original["rows"])
|
||||
original_txns = doc.number_of_transactions
|
||||
|
||||
# Shrink the box to roughly the top half (simulating a user drag).
|
||||
x0, top, x1, bottom = original["bbox"]
|
||||
shrunk = [x0, top, x1, top + (bottom - top) * 0.5]
|
||||
|
||||
details = reextract_pdf_table(doc.name, original["page"], original["table_index"], shrunk)
|
||||
updated = details["pdf_tables"][0]
|
||||
doc.reload()
|
||||
|
||||
self.assertLess(len(updated["rows"]), original_rows)
|
||||
self.assertLess(doc.number_of_transactions, original_txns)
|
||||
self.assertEqual(len(details["final_transactions"]), doc.number_of_transactions)
|
||||
|
||||
def test_pdf_set_table_header(self):
|
||||
"""User can clear a table's header (no header row) or set a specific header row."""
|
||||
html = """
|
||||
<html><body>
|
||||
<table border="1"><tr><th>Date</th><th>Narration</th><th>Amount</th></tr>
|
||||
<tr><td>01/04/2024</td><td>UPI PAYMENT</td><td>500.00</td></tr>
|
||||
<tr><td>03/04/2024</td><td>SALARY</td><td>20000.00</td></tr></table>
|
||||
</body></html>
|
||||
"""
|
||||
doc = self._create_pdf_import_log(html)
|
||||
table = doc.get_pdf_tables()[0]
|
||||
self.assertEqual(table["header_index"], 0)
|
||||
original = {
|
||||
c["maps_to"]: c["index"] for c in table["column_mapping"] if c["maps_to"] != "Do not import"
|
||||
}
|
||||
|
||||
# Clear the header (-1): header is removed but the mapping is preserved (not re-guessed).
|
||||
details = set_pdf_table_header(doc.name, table["page"], table["table_index"], -1)
|
||||
updated = details["pdf_tables"][0]
|
||||
self.assertIsNone(updated["header_index"])
|
||||
preserved = {
|
||||
c["maps_to"]: c["index"] for c in updated["column_mapping"] if c["maps_to"] != "Do not import"
|
||||
}
|
||||
self.assertEqual(preserved, original)
|
||||
|
||||
# Set row 0 back as the header: it resolves meaningfully, so mapping is re-derived.
|
||||
details = set_pdf_table_header(doc.name, table["page"], table["table_index"], 0)
|
||||
updated = details["pdf_tables"][0]
|
||||
self.assertEqual(updated["header_index"], 0)
|
||||
mapped = {
|
||||
c["maps_to"]: c["index"] for c in updated["column_mapping"] if c["maps_to"] != "Do not import"
|
||||
}
|
||||
self.assertEqual(mapped.get("Date"), 0)
|
||||
self.assertEqual(mapped.get("Description"), 1)
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# CSV/XLSX column mapping + header overrides
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
def _create_csv_import_log(self, csv_text: str) -> BankStatementImportLog:
|
||||
file_doc = frappe.get_doc(
|
||||
{
|
||||
"doctype": "File",
|
||||
"file_name": f"test-statement-{frappe.generate_hash(length=8)}.csv",
|
||||
"is_private": 1,
|
||||
"content": csv_text,
|
||||
}
|
||||
).insert(ignore_permissions=True)
|
||||
|
||||
doc = frappe.get_doc(
|
||||
{
|
||||
"doctype": "Bank Statement Import Log",
|
||||
"bank_account": self.bank_account,
|
||||
"file": file_doc.file_url,
|
||||
}
|
||||
)
|
||||
return doc.insert()
|
||||
|
||||
def test_csv_update_column_mapping(self):
|
||||
"""Overriding the column mapping recomputes the transaction count."""
|
||||
csv_text = "Date,Narration,Amount\n01/04/2024,UPI PAYMENT,500.00\n03/04/2024,SALARY,20000.00\n"
|
||||
doc = self._create_csv_import_log(csv_text)
|
||||
self.assertEqual(doc.number_of_transactions, 2)
|
||||
|
||||
# Drop the amount column -> no amount -> no transactions detected.
|
||||
mapping = [
|
||||
{"index": c.index, "maps_to": "Do not import" if c.maps_to == "Amount" else c.maps_to}
|
||||
for c in doc.column_mapping
|
||||
]
|
||||
details = update_column_mapping(doc.name, mapping)
|
||||
doc.reload()
|
||||
self.assertEqual(doc.number_of_transactions, 0)
|
||||
self.assertEqual(len(details["final_transactions"]), 0)
|
||||
|
||||
def test_csv_set_header_index_preserves_mapping(self):
|
||||
"""Clearing the header keeps the user's mapping; it is not re-guessed."""
|
||||
csv_text = "Date,Narration,Amount\n01/04/2024,UPI PAYMENT,500.00\n03/04/2024,SALARY,20000.00\n"
|
||||
doc = self._create_csv_import_log(csv_text)
|
||||
self.assertEqual(doc.detected_header_index, 0)
|
||||
|
||||
# Manually map the Narration column (1) as Reference.
|
||||
mapping = [
|
||||
{
|
||||
"index": c.index,
|
||||
"maps_to": "Reference" if c.index == 1 else c.maps_to,
|
||||
"header_text": c.header_text,
|
||||
}
|
||||
for c in doc.column_mapping
|
||||
]
|
||||
update_column_mapping(doc.name, mapping)
|
||||
doc.reload()
|
||||
|
||||
# Clear the header row: the manual mapping must be preserved (column 1 stays Reference,
|
||||
# not re-guessed to Description). The label row fails date parsing, so 2 transactions remain.
|
||||
set_header_index(doc.name, -1)
|
||||
doc.reload()
|
||||
self.assertEqual(doc.detected_header_index, -1)
|
||||
self.assertEqual(doc.number_of_transactions, 2)
|
||||
current = {c.index: c.maps_to for c in doc.column_mapping}
|
||||
self.assertEqual(current.get(1), "Reference")
|
||||
|
||||
# Restore row 0 as the header (resolves meaningfully -> re-derived from labels).
|
||||
set_header_index(doc.name, 0)
|
||||
doc.reload()
|
||||
self.assertEqual(doc.detected_header_index, 0)
|
||||
restored = {c.maps_to: c.index for c in doc.column_mapping if c.maps_to != "Do not import"}
|
||||
self.assertEqual(restored.get("Description"), 1)
|
||||
|
||||
|
||||
test_hdfc_sample_statement_data = [
|
||||
["HDFC BANK Ltd. Page No .: 1 Statement of accounts", "", "", "", "", "", ""],
|
||||
|
||||
Reference in New Issue
Block a user