feat(Transaction Deletion Record): Editable "DocTypes To Delete" List with CSV import/export (#50592)

* feat: add editable DocTypes To Delete list with import/export

Add user control over transaction deletion with reviewable and reusable deletion templates.

- New "DocTypes To Delete" table allows users to review and customize what will be deleted before submission
- Import/Export CSV templates for reusability across environments
- Company field rule: only filter by company if field is specifically named "company", otherwise delete all records
- Child tables (istable=1) automatically excluded from selection
- "Remove Zero Counts" helper button to clean up list
- Backward compatible with existing deletion records

* refactor: improve Transaction Deletion Record code quality

- Remove unnecessary chatty comments from AI-generated code
- Add concise docstrings to all new methods
- Remove redundant @frappe.whitelist() decorators from internal methods
- Improve CSV import validation (header check, child table filtering)
- Add better error feedback with consolidated skip messages
- Reorder form fields: To Delete list now appears before Excluded list
- Add conditional visibility for Summary table (legacy records only)
- Improve architectural clarity: single API entry point per feature

Technical improvements:
- export_to_delete_template_method and import_to_delete_template_method
  are now internal helpers without whitelist decorators
- CSV import now validates format and provides detailed skip reasons
- Summary table only shows for submitted records without To Delete list
- Maintains backward compatibility for existing deletion records

* fix: field order

* test: fix broken tests and add new ones

* fix: adapt create_transaction_deletion_request

* test: fix assertRaises trigger

* fix: conditionally execute Transaction Deletion pre-tasks based on selected DocTypes

* refactor: replace boolean task flags with status fields

* fix: remove UI comment

* fix: don't allow virtual doctype selection and improve protected Doctype List

* fix: replace outdated frappe.db.sql by frappe.qb

* feat: add support for multiple company fields

* fix: autofill comapny field, add docstrings, filter for company_field

* fix: add edge case handling for update_naming_series and add tests for prefix extraction

* fix: use redis for running deletion validation, check per doctype instead of company
This commit is contained in:
Henning Wendtland
2026-01-25 09:50:28 +01:00
committed by GitHub
parent 88069779b2
commit 0fb37ad792
14 changed files with 1784 additions and 150 deletions

View File

@@ -43,16 +43,18 @@
"read_only": 1 "read_only": 1
} }
], ],
"grid_page_length": 50,
"index_web_pages_for_search": 1, "index_web_pages_for_search": 1,
"istable": 1, "istable": 1,
"links": [], "links": [],
"modified": "2024-03-27 13:10:55.008837", "modified": "2025-11-14 16:17:25.584675",
"modified_by": "Administrator", "modified_by": "Administrator",
"module": "Accounts", "module": "Accounts",
"name": "Transaction Deletion Record Details", "name": "Transaction Deletion Record Details",
"owner": "Administrator", "owner": "Administrator",
"permissions": [], "permissions": [],
"row_format": "Dynamic",
"sort_field": "creation", "sort_field": "creation",
"sort_order": "DESC", "sort_order": "DESC",
"states": [] "states": []
} }

View File

@@ -459,3 +459,4 @@ erpnext.patches.v16_0.fix_barcode_typo
erpnext.patches.v16_0.set_post_change_gl_entries_on_pos_settings erpnext.patches.v16_0.set_post_change_gl_entries_on_pos_settings
erpnext.patches.v15_0.create_accounting_dimensions_in_advance_taxes_and_charges erpnext.patches.v15_0.create_accounting_dimensions_in_advance_taxes_and_charges
execute:frappe.delete_doc_if_exists("Workspace Sidebar", "Opening & Closing") execute:frappe.delete_doc_if_exists("Workspace Sidebar", "Opening & Closing")
erpnext.patches.v16_0.migrate_transaction_deletion_task_flags_to_status # 2

View File

@@ -0,0 +1,42 @@
import frappe
def execute():
"""
Migrate Transaction Deletion Record boolean task flags to status Select fields.
Renames fields from old names to new names with _status suffix.
Maps: 0 -> "Pending", 1 -> "Completed"
"""
if not frappe.db.table_exists("tabTransaction Deletion Record"):
return
# Field mapping: old boolean field name -> new status field name
field_mapping = {
"delete_bin_data": "delete_bin_data_status",
"delete_leads_and_addresses": "delete_leads_and_addresses_status",
"reset_company_default_values": "reset_company_default_values_status",
"clear_notifications": "clear_notifications_status",
"initialize_doctypes_table": "initialize_doctypes_table_status",
"delete_transactions": "delete_transactions_status",
}
# Get all Transaction Deletion Records
records = frappe.db.get_all("Transaction Deletion Record", pluck="name")
for name in records or []:
updates = {}
for old_field, new_field in field_mapping.items():
# Read from old boolean field
current_value = frappe.db.get_value("Transaction Deletion Record", name, old_field)
# Map to new status and write to new field name
if current_value in (1, "1", True):
updates[new_field] = "Completed"
else:
# Handle 0, "0", False, None, empty string
updates[new_field] = "Pending"
# Update all fields at once
if updates:
frappe.db.set_value("Transaction Deletion Record", name, updates, update_modified=False)

View File

@@ -182,6 +182,10 @@ def create_transaction_deletion_record(company):
transaction_deletion_record.company = company transaction_deletion_record.company = company
transaction_deletion_record.process_in_single_transaction = True transaction_deletion_record.process_in_single_transaction = True
transaction_deletion_record.save(ignore_permissions=True) transaction_deletion_record.save(ignore_permissions=True)
transaction_deletion_record.generate_to_delete_list()
transaction_deletion_record.reload()
transaction_deletion_record.submit() transaction_deletion_record.submit()
transaction_deletion_record.start_deletion_tasks() transaction_deletion_record.start_deletion_tasks()

View File

@@ -1081,6 +1081,8 @@ def get_billing_shipping_address(name, billing_address=None, shipping_address=No
@frappe.whitelist() @frappe.whitelist()
def create_transaction_deletion_request(company): def create_transaction_deletion_request(company):
frappe.only_for("System Manager")
from erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record import ( from erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record import (
is_deletion_doc_running, is_deletion_doc_running,
) )
@@ -1088,12 +1090,16 @@ def create_transaction_deletion_request(company):
is_deletion_doc_running(company) is_deletion_doc_running(company)
tdr = frappe.get_doc({"doctype": "Transaction Deletion Record", "company": company}) tdr = frappe.get_doc({"doctype": "Transaction Deletion Record", "company": company})
tdr.insert()
tdr.generate_to_delete_list()
tdr.reload()
tdr.submit() tdr.submit()
tdr.start_deletion_tasks() tdr.start_deletion_tasks()
frappe.msgprint( frappe.msgprint(
_("A Transaction Deletion Document: {0} is triggered for {0}").format( _("Transaction Deletion Document {0} has been triggered for company {1}").format(
get_link_to_form("Transaction Deletion Record", tdr.name) get_link_to_form("Transaction Deletion Record", tdr.name), frappe.bold(company)
), )
frappe.bold(company),
) )

View File

@@ -8,38 +8,77 @@ from frappe.tests import IntegrationTestCase
class TestTransactionDeletionRecord(IntegrationTestCase): class TestTransactionDeletionRecord(IntegrationTestCase):
def setUp(self): def setUp(self):
# Clear all deletion cache flags from previous tests
self._clear_all_deletion_cache_flags()
create_company("Dunder Mifflin Paper Co") create_company("Dunder Mifflin Paper Co")
def tearDown(self): def tearDown(self):
# Clean up all deletion cache flags after each test
self._clear_all_deletion_cache_flags()
frappe.db.rollback() frappe.db.rollback()
def _clear_all_deletion_cache_flags(self):
"""Clear all deletion_running_doctype:* cache keys"""
# Get all keys matching the pattern
cache_keys = frappe.cache.get_keys("deletion_running_doctype:*")
if cache_keys:
for key in cache_keys:
# Decode bytes to string if needed
key_str = key.decode() if isinstance(key, bytes) else key
# Extract just the key name (remove site prefix if present)
# Keys are in format: site_prefix|deletion_running_doctype:DocType
if "|" in key_str:
key_name = key_str.split("|")[1]
else:
key_name = key_str
frappe.cache.delete_value(key_name)
def test_doctypes_contain_company_field(self): def test_doctypes_contain_company_field(self):
tdr = create_transaction_deletion_doc("Dunder Mifflin Paper Co") """Test that all DocTypes in To Delete list have a valid company link field"""
for doctype in tdr.doctypes: tdr = create_and_submit_transaction_deletion_doc("Dunder Mifflin Paper Co")
contains_company = False for doctype_row in tdr.doctypes_to_delete:
doctype_fields = frappe.get_meta(doctype.doctype_name).as_dict()["fields"] # If company_field is specified, verify it's a valid Company link field
for doctype_field in doctype_fields: if doctype_row.company_field:
if doctype_field["fieldtype"] == "Link" and doctype_field["options"] == "Company": field_found = False
contains_company = True doctype_fields = frappe.get_meta(doctype_row.doctype_name).as_dict()["fields"]
break for doctype_field in doctype_fields:
self.assertTrue(contains_company) if (
doctype_field["fieldname"] == doctype_row.company_field
and doctype_field["fieldtype"] == "Link"
and doctype_field["options"] == "Company"
):
field_found = True
break
self.assertTrue(
field_found,
f"DocType {doctype_row.doctype_name} should have company field '{doctype_row.company_field}'",
)
def test_no_of_docs_is_correct(self): def test_no_of_docs_is_correct(self):
for _i in range(5): """Test that document counts are calculated correctly in To Delete list"""
for _ in range(5):
create_task("Dunder Mifflin Paper Co") create_task("Dunder Mifflin Paper Co")
tdr = create_transaction_deletion_doc("Dunder Mifflin Paper Co") tdr = create_and_submit_transaction_deletion_doc("Dunder Mifflin Paper Co")
tdr.reload() tdr.reload()
for doctype in tdr.doctypes:
# Check To Delete list has correct count
task_found = False
for doctype in tdr.doctypes_to_delete:
if doctype.doctype_name == "Task": if doctype.doctype_name == "Task":
self.assertEqual(doctype.no_of_docs, 5) self.assertEqual(doctype.document_count, 5)
task_found = True
break
self.assertTrue(task_found, "Task should be in To Delete list")
def test_deletion_is_successful(self): def test_deletion_is_successful(self):
"""Test that deletion actually removes documents"""
create_task("Dunder Mifflin Paper Co") create_task("Dunder Mifflin Paper Co")
create_transaction_deletion_doc("Dunder Mifflin Paper Co") create_and_submit_transaction_deletion_doc("Dunder Mifflin Paper Co")
tasks_containing_company = frappe.get_all("Task", filters={"company": "Dunder Mifflin Paper Co"}) tasks_containing_company = frappe.get_all("Task", filters={"company": "Dunder Mifflin Paper Co"})
self.assertEqual(tasks_containing_company, []) self.assertEqual(tasks_containing_company, [])
def test_company_transaction_deletion_request(self): def test_company_transaction_deletion_request(self):
"""Test creation via company deletion request method"""
from erpnext.setup.doctype.company.company import create_transaction_deletion_request from erpnext.setup.doctype.company.company import create_transaction_deletion_request
# don't reuse below company for other test cases # don't reuse below company for other test cases
@@ -49,15 +88,314 @@ class TestTransactionDeletionRecord(IntegrationTestCase):
# below call should not raise any exceptions or throw errors # below call should not raise any exceptions or throw errors
create_transaction_deletion_request(company) create_transaction_deletion_request(company)
def test_generate_to_delete_list(self):
"""Test automatic generation of To Delete list"""
company = "Dunder Mifflin Paper Co"
create_task(company)
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
tdr.insert()
# Generate To Delete list
tdr.generate_to_delete_list()
tdr.reload()
# Should have at least Task in the list
self.assertGreater(len(tdr.doctypes_to_delete), 0)
task_in_list = any(d.doctype_name == "Task" for d in tdr.doctypes_to_delete)
self.assertTrue(task_in_list, "Task should be in To Delete list")
def test_validation_prevents_child_tables(self):
"""Test that child tables cannot be added to To Delete list"""
company = "Dunder Mifflin Paper Co"
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
tdr.append("doctypes_to_delete", {"doctype_name": "Sales Invoice Item"}) # Child table
# Should throw validation error
with self.assertRaises(frappe.ValidationError):
tdr.insert()
def test_validation_prevents_protected_doctypes(self):
"""Test that protected DocTypes cannot be added to To Delete list"""
company = "Dunder Mifflin Paper Co"
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
tdr.append("doctypes_to_delete", {"doctype_name": "DocType"}) # Protected
# Should throw validation error
with self.assertRaises(frappe.ValidationError):
tdr.insert()
def test_csv_export_import(self):
"""Test CSV export and import functionality with company_field column"""
company = "Dunder Mifflin Paper Co"
create_task(company)
# Create and generate To Delete list
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
tdr.insert()
tdr.generate_to_delete_list()
tdr.reload()
original_count = len(tdr.doctypes_to_delete)
self.assertGreater(original_count, 0)
# Export as CSV
tdr.export_to_delete_template_method()
csv_content = frappe.response.get("result")
self.assertIsNotNone(csv_content)
self.assertIn("doctype_name", csv_content)
self.assertIn("company_field", csv_content) # New: verify company_field column exists
# Create new record and import
tdr2 = frappe.new_doc("Transaction Deletion Record")
tdr2.company = company
tdr2.insert()
result = tdr2.import_to_delete_template_method(csv_content)
tdr2.reload()
# Should have same entries (counts may differ due to new task)
self.assertEqual(len(tdr2.doctypes_to_delete), original_count)
self.assertGreaterEqual(result["imported"], 1)
# Verify company_field values are preserved
for row in tdr2.doctypes_to_delete:
if row.doctype_name == "Task":
# Task should have company field set
self.assertIsNotNone(row.company_field, "Task should have company_field set after import")
def test_progress_tracking(self):
"""Test that deleted checkbox is marked when DocType deletion completes"""
company = "Dunder Mifflin Paper Co"
create_task(company)
tdr = create_and_submit_transaction_deletion_doc(company)
tdr.reload()
# After deletion, Task should be marked as deleted in To Delete list
# Note: Must match using composite key (doctype_name + company_field)
task_row = None
for doctype in tdr.doctypes_to_delete:
if doctype.doctype_name == "Task":
task_row = doctype
break
if task_row:
self.assertEqual(task_row.deleted, 1, "Task should be marked as deleted")
def test_composite_key_validation(self):
"""Test that duplicate (doctype_name + company_field) combinations are prevented"""
company = "Dunder Mifflin Paper Co"
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
tdr.append("doctypes_to_delete", {"doctype_name": "Task", "company_field": "company"})
tdr.append("doctypes_to_delete", {"doctype_name": "Task", "company_field": "company"}) # Duplicate!
# Should throw validation error for duplicate composite key
with self.assertRaises(frappe.ValidationError):
tdr.insert()
def test_same_doctype_different_company_field_allowed(self):
"""Test that same DocType can be added with different company_field values"""
company = "Dunder Mifflin Paper Co"
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
# Same DocType but one with company field, one without (None)
tdr.append("doctypes_to_delete", {"doctype_name": "Task", "company_field": "company"})
tdr.append("doctypes_to_delete", {"doctype_name": "Task", "company_field": None})
# Should NOT throw error - different company_field values are allowed
try:
tdr.insert()
self.assertEqual(
len(tdr.doctypes_to_delete),
2,
"Should allow 2 Task entries with different company_field values",
)
except frappe.ValidationError as e:
self.fail(f"Should allow same DocType with different company_field values, but got error: {e}")
def test_company_field_validation(self):
"""Test that invalid company_field values are rejected"""
company = "Dunder Mifflin Paper Co"
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
# Add Task with invalid company field
tdr.append("doctypes_to_delete", {"doctype_name": "Task", "company_field": "nonexistent_field"})
# Should throw validation error for invalid company field
with self.assertRaises(frappe.ValidationError):
tdr.insert()
def test_get_naming_series_prefix_with_dot(self):
"""Test prefix extraction for standard dot-separated naming series"""
from erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record import (
TransactionDeletionRecord,
)
# Standard patterns with dot separator
self.assertEqual(TransactionDeletionRecord.get_naming_series_prefix("TDL.####", "Task"), "TDL")
self.assertEqual(TransactionDeletionRecord.get_naming_series_prefix("PREFIX.#####", "Task"), "PREFIX")
self.assertEqual(
TransactionDeletionRecord.get_naming_series_prefix("TASK-.YYYY.-.#####", "Task"), "TASK-.YYYY.-"
)
def test_get_naming_series_prefix_with_brace(self):
"""Test prefix extraction for format patterns with brace separators"""
from erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record import (
TransactionDeletionRecord,
)
# Format patterns with brace separator
self.assertEqual(
TransactionDeletionRecord.get_naming_series_prefix("QA-ACT-{#####}", "Quality Action"), "QA-ACT-"
)
self.assertEqual(
TransactionDeletionRecord.get_naming_series_prefix("PREFIX-{####}", "Task"), "PREFIX-"
)
self.assertEqual(TransactionDeletionRecord.get_naming_series_prefix("{####}", "Task"), "")
def test_get_naming_series_prefix_fallback(self):
"""Test prefix extraction fallback for patterns without standard separators"""
from erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record import (
TransactionDeletionRecord,
)
# Edge case: pattern with # but no dot or brace (shouldn't happen in practice)
self.assertEqual(TransactionDeletionRecord.get_naming_series_prefix("PREFIX####", "Task"), "PREFIX")
# Edge case: pattern with no # at all
self.assertEqual(
TransactionDeletionRecord.get_naming_series_prefix("JUSTPREFIX", "Task"), "JUSTPREFIX"
)
def test_cache_flag_management(self):
"""Test that cache flags can be set and cleared correctly"""
company = "Dunder Mifflin Paper Co"
create_task(company)
tdr = frappe.new_doc("Transaction Deletion Record")
tdr.company = company
tdr.insert()
tdr.generate_to_delete_list()
tdr.reload()
# Test _set_deletion_cache
tdr._set_deletion_cache()
# Verify flag is set for Task specifically
cached_value = frappe.cache.get_value("deletion_running_doctype:Task")
self.assertEqual(cached_value, tdr.name, "Cache flag should be set for Task")
# Test _clear_deletion_cache
tdr._clear_deletion_cache()
# Verify flag is cleared
cached_value = frappe.cache.get_value("deletion_running_doctype:Task")
self.assertIsNone(cached_value, "Cache flag should be cleared for Task")
def test_check_for_running_deletion_blocks_save(self):
"""Test that check_for_running_deletion_job blocks saves when cache flag exists"""
from erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record import (
check_for_running_deletion_job,
)
company = "Dunder Mifflin Paper Co"
# Manually set cache flag to simulate running deletion
frappe.cache.set_value("deletion_running_doctype:Task", "TDR-00001", expires_in_sec=60)
try:
# Try to validate a new Task
new_task = frappe.new_doc("Task")
new_task.company = company
new_task.subject = "Should be blocked"
# Should throw error when cache flag exists
with self.assertRaises(frappe.ValidationError) as context:
check_for_running_deletion_job(new_task)
error_message = str(context.exception)
self.assertIn("currently deleting", error_message)
self.assertIn("TDR-00001", error_message)
finally:
# Cleanup: clear the manually set flag
frappe.cache.delete_value("deletion_running_doctype:Task")
def test_check_for_running_deletion_allows_save_when_no_flag(self):
"""Test that documents can be saved when no deletion is running"""
company = "Dunder Mifflin Paper Co"
# Ensure no cache flag exists
frappe.cache.delete_value("deletion_running_doctype:Task")
# Try to create and save a new Task
new_task = frappe.new_doc("Task")
new_task.company = company
new_task.subject = "Should be allowed"
# Should NOT throw error when no cache flag - actually save it
try:
new_task.insert()
# Cleanup
frappe.delete_doc("Task", new_task.name)
except frappe.ValidationError as e:
self.fail(f"Should allow save when no deletion is running, but got: {e}")
def test_only_one_deletion_allowed_globally(self):
"""Test that only one deletion can be submitted at a time (global enforcement)"""
company1 = "Dunder Mifflin Paper Co"
company2 = "Sabre Corporation"
create_company(company2)
# Create and submit first deletion (but don't start it)
tdr1 = frappe.new_doc("Transaction Deletion Record")
tdr1.company = company1
tdr1.insert()
tdr1.append("doctypes_to_delete", {"doctype_name": "Task", "company_field": "company"})
tdr1.save()
tdr1.submit() # Status becomes "Queued"
try:
# Try to submit second deletion for different company
tdr2 = frappe.new_doc("Transaction Deletion Record")
tdr2.company = company2 # Different company!
tdr2.insert()
tdr2.append("doctypes_to_delete", {"doctype_name": "Lead", "company_field": "company"})
tdr2.save()
# Should throw error - only one deletion allowed globally
with self.assertRaises(frappe.ValidationError) as context:
tdr2.submit()
self.assertIn("already", str(context.exception).lower())
self.assertIn(tdr1.name, str(context.exception))
finally:
# Cleanup
tdr1.cancel()
def create_company(company_name): def create_company(company_name):
company = frappe.get_doc({"doctype": "Company", "company_name": company_name, "default_currency": "INR"}) company = frappe.get_doc({"doctype": "Company", "company_name": company_name, "default_currency": "INR"})
company.insert(ignore_if_duplicate=True) company.insert(ignore_if_duplicate=True)
def create_transaction_deletion_doc(company): def create_and_submit_transaction_deletion_doc(company):
"""Create and execute a transaction deletion record"""
tdr = frappe.get_doc({"doctype": "Transaction Deletion Record", "company": company}) tdr = frappe.get_doc({"doctype": "Transaction Deletion Record", "company": company})
tdr.insert() tdr.insert()
tdr.generate_to_delete_list()
tdr.reload()
tdr.process_in_single_transaction = True tdr.process_in_single_transaction = True
tdr.submit() tdr.submit()
tdr.start_deletion_tasks() tdr.start_deletion_tasks()

View File

@@ -2,13 +2,58 @@
// For license information, please see license.txt // For license information, please see license.txt
frappe.ui.form.on("Transaction Deletion Record", { frappe.ui.form.on("Transaction Deletion Record", {
setup: function (frm) {
// Set up query for DocTypes to exclude child tables and virtual doctypes
// Note: Same DocType can be added multiple times with different company_field values
frm.set_query("doctype_name", "doctypes_to_delete", function () {
// Build exclusion list from protected and ignored doctypes
let excluded_doctypes = ["Transaction Deletion Record"]; // Always exclude self
// Add protected doctypes (fetched in onload)
if (frm.protected_doctypes_list && frm.protected_doctypes_list.length > 0) {
excluded_doctypes = excluded_doctypes.concat(frm.protected_doctypes_list);
}
// Add doctypes from the ignore list
if (frm.doc.doctypes_to_be_ignored && frm.doc.doctypes_to_be_ignored.length > 0) {
frm.doc.doctypes_to_be_ignored.forEach((row) => {
if (row.doctype_name) {
excluded_doctypes.push(row.doctype_name);
}
});
}
let filters = [
["DocType", "istable", "=", 0], // Exclude child tables
["DocType", "is_virtual", "=", 0], // Exclude virtual doctypes
];
// Only add "not in" filter if we have items to exclude
if (excluded_doctypes.length > 0) {
filters.push(["DocType", "name", "not in", excluded_doctypes]);
}
return { filters: filters };
});
},
onload: function (frm) { onload: function (frm) {
if (frm.doc.docstatus == 0) { if (frm.doc.docstatus == 0) {
let doctypes_to_be_ignored_array; // Fetch protected doctypes list for filtering
frappe.call({
method: "erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.get_protected_doctypes",
callback: function (r) {
if (r.message) {
frm.protected_doctypes_list = r.message;
}
},
});
// Fetch ignored doctypes and populate table
frappe.call({ frappe.call({
method: "erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.get_doctypes_to_be_ignored", method: "erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.get_doctypes_to_be_ignored",
callback: function (r) { callback: function (r) {
doctypes_to_be_ignored_array = r.message; let doctypes_to_be_ignored_array = r.message;
populate_doctypes_to_be_ignored(doctypes_to_be_ignored_array, frm); populate_doctypes_to_be_ignored(doctypes_to_be_ignored_array, frm);
frm.refresh_field("doctypes_to_be_ignored"); frm.refresh_field("doctypes_to_be_ignored");
}, },
@@ -17,20 +62,264 @@ frappe.ui.form.on("Transaction Deletion Record", {
}, },
refresh: function (frm) { refresh: function (frm) {
if (frm.doc.docstatus == 1 && ["Queued", "Failed"].find((x) => x == frm.doc.status)) { // Override submit button to show custom confirmation
let execute_btn = frm.doc.status == "Queued" ? __("Start Deletion") : __("Retry"); if (frm.doc.docstatus === 0 && !frm.is_new()) {
frm.page.clear_primary_action();
frm.page.set_primary_action(__("Submit"), () => {
if (!frm.doc.doctypes_to_delete || frm.doc.doctypes_to_delete.length === 0) {
frappe.msgprint(__("Please generate the To Delete list before submitting"));
return;
}
frm.add_custom_button(execute_btn, () => { let message =
// Entry point for chain of events `<div style='margin-bottom: 15px;'><b style='color: #d73939;'>⚠ ${__(
"Warning: This action cannot be undone!"
)}</b></div>` +
`<div style='margin-bottom: 10px;'>${__(
"You are about to permanently delete data for {0} entries for company {1}.",
[`<b>${frm.doc.doctypes_to_delete.length}</b>`, `<b>${frm.doc.company}</b>`]
)}</div>` +
`<div style='margin-bottom: 10px;'><b>${__("What will be deleted:")}</b></div>` +
`<ul style='margin-left: 20px; margin-bottom: 10px;'>` +
`<li><b>${__("DocTypes with a company field:")}</b> ${__(
"Only records belonging to {0} will be deleted",
[`<b>${frm.doc.company}</b>`]
)}</li>` +
`<li><b>${__("DocTypes without a company field:")}</b> ${__(
"ALL records will be deleted (entire DocType cleared)"
)}</li>` +
`</ul>` +
`<div style='margin-bottom: 10px; padding: 10px; background-color: #fff3cd; border: 1px solid #ffc107; border-radius: 4px;'>` +
`<b style='color: #856404;'>📦 ${__(
"IMPORTANT: Create a backup before proceeding!"
)}</b>` +
`</div>` +
`<div style='margin-top: 10px;'>${__(
"Deletion will start automatically after submission."
)}</div>`;
frappe.confirm(
message,
() => {
frm.save("Submit");
},
() => {}
);
});
}
if (frm.doc.docstatus == 0) {
frm.add_custom_button(__("Generate To Delete List"), () => {
frm.call({
method: "generate_to_delete_list",
doc: frm.doc,
callback: (r) => {
frappe.show_alert({
message: __("To Delete list generated with {0} DocTypes", [r.message.count]),
indicator: "green",
});
frm.refresh();
},
});
});
if (frm.doc.doctypes_to_delete && frm.doc.doctypes_to_delete.length > 0) {
frm.add_custom_button(
__("Export"),
() => {
open_url_post(
"/api/method/erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.export_to_delete_template",
{
name: frm.doc.name,
}
);
},
__("Template")
);
frm.add_custom_button(__("Remove Zero Counts"), () => {
let removed_count = 0;
let rows_to_keep = [];
frm.doc.doctypes_to_delete.forEach((row) => {
if (row.document_count && row.document_count > 0) {
rows_to_keep.push(row);
} else {
removed_count++;
}
});
if (removed_count === 0) {
frappe.msgprint(__("No rows with zero document count found"));
return;
}
frm.doc.doctypes_to_delete = rows_to_keep;
frm.refresh_field("doctypes_to_delete");
frm.dirty();
frappe.show_alert({
message: __(
"Removed {0} rows with zero document count. Please save to persist changes.",
[removed_count]
),
indicator: "orange",
});
});
}
frm.add_custom_button(
__("Import"),
() => {
new frappe.ui.FileUploader({
doctype: "Transaction Deletion Record",
docname: frm.doc.name,
folder: "Home/Attachments",
restrictions: {
allowed_file_types: [".csv"],
},
on_success: (file_doc) => {
frappe.call({
method: "erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.process_import_template",
args: {
transaction_deletion_record_name: frm.doc.name,
file_url: file_doc.file_url,
},
freeze: true,
freeze_message: __("Processing import..."),
callback: (r) => {
if (r.message) {
frappe.show_alert({
message: __("Imported {0} DocTypes", [r.message.imported]),
indicator: "green",
});
frappe.model.clear_doc(frm.doctype, frm.docname);
frm.reload_doc();
}
},
});
},
});
},
__("Template")
);
}
// Only show Retry button for Failed status (deletion starts automatically on submit)
if (frm.doc.docstatus == 1 && frm.doc.status == "Failed") {
frm.add_custom_button(__("Retry"), () => {
frm.call({ frm.call({
method: "start_deletion_tasks", method: "start_deletion_tasks",
doc: frm.doc, doc: frm.doc,
callback: () => {
frappe.show_alert({
message: __("Deletion process restarted"),
indicator: "blue",
});
frm.reload_doc();
},
}); });
}); });
} }
}, },
}); });
frappe.ui.form.on("Transaction Deletion Record To Delete", {
doctype_name: function (frm, cdt, cdn) {
let row = locals[cdt][cdn];
if (row.doctype_name) {
// Fetch company fields for auto-selection (only if exactly 1 field exists)
frappe.call({
method: "erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.get_company_link_fields",
args: {
doctype_name: row.doctype_name,
},
callback: function (r) {
if (r.message && r.message.length === 1 && !row.company_field) {
frappe.model.set_value(cdt, cdn, "company_field", r.message[0]);
} else if (r.message && r.message.length > 1) {
// Show message with available options when multiple company fields exist
frappe.show_alert({
message: __("Multiple company fields available: {0}. Please select manually.", [
r.message.join(", "),
]),
indicator: "blue",
});
}
},
});
// Auto-populate child DocTypes and document count
frm.call({
method: "populate_doctype_details",
doc: frm.doc,
args: {
doctype_name: row.doctype_name,
company: frm.doc.company,
company_field: row.company_field,
},
callback: function (r) {
if (r.message) {
if (r.message.error) {
frappe.msgprint({
title: __("Error"),
indicator: "red",
message: __("Error getting details for {0}: {1}", [
row.doctype_name,
r.message.error,
]),
});
}
frappe.model.set_value(cdt, cdn, "child_doctypes", r.message.child_doctypes || "");
frappe.model.set_value(cdt, cdn, "document_count", r.message.document_count || 0);
}
},
});
}
},
company_field: function (frm, cdt, cdn) {
let row = locals[cdt][cdn];
if (row.doctype_name && row.company_field !== undefined) {
// Check for duplicates using composite key (doctype_name + company_field)
let duplicates = frm.doc.doctypes_to_delete.filter(
(r) =>
r.doctype_name === row.doctype_name &&
r.company_field === row.company_field &&
r.name !== row.name
);
if (duplicates.length > 0) {
frappe.msgprint(
__("DocType {0} with company field '{1}' is already in the list", [
row.doctype_name,
row.company_field || __("(none)"),
])
);
frappe.model.set_value(cdt, cdn, "company_field", "");
return;
}
// Recalculate document count if company_field changes
if (row.doctype_name) {
frm.call({
method: "populate_doctype_details",
doc: frm.doc,
args: {
doctype_name: row.doctype_name,
company: frm.doc.company,
company_field: row.company_field,
},
callback: function (r) {
if (r.message && r.message.document_count !== undefined) {
frappe.model.set_value(cdt, cdn, "document_count", r.message.document_count || 0);
}
},
});
}
}
},
});
function populate_doctypes_to_be_ignored(doctypes_to_be_ignored_array, frm) { function populate_doctypes_to_be_ignored(doctypes_to_be_ignored_array, frm) {
if (frm.doc.doctypes_to_be_ignored.length === 0) { if (frm.doc.doctypes_to_be_ignored.length === 0) {
var i; var i;

View File

@@ -11,14 +11,17 @@
"status", "status",
"error_log", "error_log",
"tasks_section", "tasks_section",
"delete_bin_data", "delete_bin_data_status",
"delete_leads_and_addresses", "delete_leads_and_addresses_status",
"reset_company_default_values", "column_break_tasks_1",
"clear_notifications", "reset_company_default_values_status",
"initialize_doctypes_table", "clear_notifications_status",
"delete_transactions", "column_break_tasks_2",
"initialize_doctypes_table_status",
"delete_transactions_status",
"section_break_tbej", "section_break_tbej",
"doctypes", "doctypes",
"doctypes_to_delete",
"doctypes_to_be_ignored", "doctypes_to_be_ignored",
"amended_from", "amended_from",
"process_in_single_transaction" "process_in_single_transaction"
@@ -33,6 +36,7 @@
"reqd": 1 "reqd": 1
}, },
{ {
"depends_on": "eval:doc.docstatus > 0 && (!doc.doctypes_to_delete || doc.doctypes_to_delete.length == 0)",
"fieldname": "doctypes", "fieldname": "doctypes",
"fieldtype": "Table", "fieldtype": "Table",
"label": "Summary", "label": "Summary",
@@ -41,11 +45,17 @@
"read_only": 1 "read_only": 1
}, },
{ {
"fieldname": "doctypes_to_delete",
"fieldtype": "Table",
"label": "DocTypes To Delete",
"options": "Transaction Deletion Record To Delete"
},
{
"description": "DocTypes that will NOT be deleted.",
"fieldname": "doctypes_to_be_ignored", "fieldname": "doctypes_to_be_ignored",
"fieldtype": "Table", "fieldtype": "Table",
"label": "Excluded DocTypes", "label": "Excluded DocTypes",
"options": "Transaction Deletion Record Item", "options": "Transaction Deletion Record Item"
"read_only": 1
}, },
{ {
"fieldname": "amended_from", "fieldname": "amended_from",
@@ -69,56 +79,71 @@
"fieldtype": "Section Break" "fieldtype": "Section Break"
}, },
{ {
"depends_on": "eval:doc.docstatus==1",
"fieldname": "tasks_section", "fieldname": "tasks_section",
"fieldtype": "Section Break", "fieldtype": "Section Break",
"label": "Tasks" "label": "Tasks"
}, },
{ {
"default": "0", "default": "Pending",
"fieldname": "delete_bin_data", "fieldname": "delete_bin_data_status",
"fieldtype": "Check", "fieldtype": "Select",
"label": "Delete Bins", "label": "Delete Bins",
"no_copy": 1, "no_copy": 1,
"options": "Pending\nCompleted\nSkipped",
"read_only": 1 "read_only": 1
}, },
{ {
"default": "0", "default": "Pending",
"fieldname": "delete_leads_and_addresses", "fieldname": "delete_leads_and_addresses_status",
"fieldtype": "Check", "fieldtype": "Select",
"label": "Delete Leads and Addresses", "label": "Delete Leads and Addresses",
"no_copy": 1, "no_copy": 1,
"options": "Pending\nCompleted\nSkipped",
"read_only": 1 "read_only": 1
}, },
{ {
"default": "0", "fieldname": "column_break_tasks_1",
"fieldname": "clear_notifications", "fieldtype": "Column Break"
"fieldtype": "Check",
"label": "Clear Notifications",
"no_copy": 1,
"read_only": 1
}, },
{ {
"default": "0", "default": "Pending",
"fieldname": "reset_company_default_values", "fieldname": "reset_company_default_values_status",
"fieldtype": "Check", "fieldtype": "Select",
"label": "Reset Company Default Values", "label": "Reset Company Default Values",
"no_copy": 1, "no_copy": 1,
"options": "Pending\nCompleted\nSkipped",
"read_only": 1 "read_only": 1
}, },
{ {
"default": "0", "default": "Pending",
"fieldname": "delete_transactions", "fieldname": "clear_notifications_status",
"fieldtype": "Check", "fieldtype": "Select",
"label": "Delete Transactions", "label": "Clear Notifications",
"no_copy": 1, "no_copy": 1,
"options": "Pending\nCompleted\nSkipped",
"read_only": 1 "read_only": 1
}, },
{ {
"default": "0", "fieldname": "column_break_tasks_2",
"fieldname": "initialize_doctypes_table", "fieldtype": "Column Break"
"fieldtype": "Check", },
{
"default": "Pending",
"fieldname": "initialize_doctypes_table_status",
"fieldtype": "Select",
"label": "Initialize Summary Table", "label": "Initialize Summary Table",
"no_copy": 1, "no_copy": 1,
"options": "Pending\nCompleted\nSkipped",
"read_only": 1
},
{
"default": "Pending",
"fieldname": "delete_transactions_status",
"fieldtype": "Select",
"label": "Delete Transactions",
"no_copy": 1,
"options": "Pending\nCompleted\nSkipped",
"read_only": 1 "read_only": 1
}, },
{ {
@@ -144,7 +169,7 @@
"index_web_pages_for_search": 1, "index_web_pages_for_search": 1,
"is_submittable": 1, "is_submittable": 1,
"links": [], "links": [],
"modified": "2024-03-27 13:10:54.828051", "modified": "2025-11-18 15:02:46.427695",
"modified_by": "Administrator", "modified_by": "Administrator",
"module": "Setup", "module": "Setup",
"name": "Transaction Deletion Record", "name": "Transaction Deletion Record",
@@ -165,8 +190,9 @@
"write": 1 "write": 1
} }
], ],
"row_format": "Dynamic",
"sort_field": "creation", "sort_field": "creation",
"sort_order": "DESC", "sort_order": "DESC",
"states": [], "states": [],
"track_changes": 1 "track_changes": 1
} }

View File

@@ -7,6 +7,7 @@ import frappe
from frappe import _, qb from frappe import _, qb
from frappe.desk.notifications import clear_notifications from frappe.desk.notifications import clear_notifications
from frappe.model.document import Document from frappe.model.document import Document
from frappe.query_builder.functions import Max
from frappe.utils import cint, comma_and, create_batch, get_link_to_form from frappe.utils import cint, comma_and, create_batch, get_link_to_form
from frappe.utils.background_jobs import get_job, is_job_enqueued from frappe.utils.background_jobs import get_job, is_job_enqueued
from frappe.utils.caching import request_cache from frappe.utils.caching import request_cache
@@ -19,6 +20,95 @@ LEDGER_ENTRY_DOCTYPES = frozenset(
) )
) )
DELETION_CACHE_TTL = 4 * 60 * 60 # 4 hours in seconds
PROTECTED_CORE_DOCTYPES = frozenset(
(
# Core Meta
"DocType",
"DocField",
"Custom Field",
"Property Setter",
"DocPerm",
"Custom DocPerm",
# User & Permissions
"User",
"Role",
"Has Role",
"User Permission",
"User Type",
# System Configuration
"Module Def",
"Workflow",
"Workflow State",
"System Settings",
# Critical System DocTypes
"File",
"Version",
"Activity Log",
"Error Log",
"Scheduled Job Type",
"Scheduled Job Log",
"Server Script",
"Client Script",
"Data Import",
"Data Export",
"Report",
"Print Format",
"Email Template",
"Assignment Rule",
"Workspace",
"Dashboard",
"Access Log",
# Transaction Deletion
"Transaction Deletion Record",
"Company",
)
)
@frappe.whitelist()
def get_protected_doctypes():
"""Get list of protected DocTypes that cannot be deleted (whitelisted for frontend)"""
frappe.only_for("System Manager")
return _get_protected_doctypes_internal()
@frappe.whitelist()
def get_company_link_fields(doctype_name):
"""Get all Company Link field names for a DocType (whitelisted for frontend autocomplete)
Args:
doctype_name: The DocType to check
Returns:
list: List of field names that link to Company DocType, ordered by field index
"""
frappe.only_for("System Manager")
if not doctype_name or not frappe.db.exists("DocType", doctype_name):
return []
return frappe.get_all(
"DocField",
filters={"parent": doctype_name, "fieldtype": "Link", "options": "Company"},
pluck="fieldname",
order_by="idx",
)
def _get_protected_doctypes_internal():
"""Internal method to get protected doctypes"""
protected = []
for doctype in PROTECTED_CORE_DOCTYPES:
if frappe.db.exists("DocType", doctype):
protected.append(doctype)
singles = frappe.get_all("DocType", filters={"issingle": 1}, pluck="name")
protected.extend(singles)
return protected
class TransactionDeletionRecord(Document): class TransactionDeletionRecord(Document):
# begin: auto-generated types # begin: auto-generated types
@@ -35,19 +125,23 @@ class TransactionDeletionRecord(Document):
from erpnext.setup.doctype.transaction_deletion_record_item.transaction_deletion_record_item import ( from erpnext.setup.doctype.transaction_deletion_record_item.transaction_deletion_record_item import (
TransactionDeletionRecordItem, TransactionDeletionRecordItem,
) )
from erpnext.setup.doctype.transaction_deletion_record_to_delete.transaction_deletion_record_to_delete import (
TransactionDeletionRecordToDelete,
)
amended_from: DF.Link | None amended_from: DF.Link | None
clear_notifications: DF.Check clear_notifications_status: DF.Literal["Pending", "Completed", "Skipped"]
company: DF.Link company: DF.Link
delete_bin_data: DF.Check delete_bin_data_status: DF.Literal["Pending", "Completed", "Skipped"]
delete_leads_and_addresses: DF.Check delete_leads_and_addresses_status: DF.Literal["Pending", "Completed", "Skipped"]
delete_transactions: DF.Check delete_transactions_status: DF.Literal["Pending", "Completed", "Skipped"]
doctypes: DF.Table[TransactionDeletionRecordDetails] doctypes: DF.Table[TransactionDeletionRecordDetails]
doctypes_to_be_ignored: DF.Table[TransactionDeletionRecordItem] doctypes_to_be_ignored: DF.Table[TransactionDeletionRecordItem]
doctypes_to_delete: DF.Table[TransactionDeletionRecordToDelete]
error_log: DF.LongText | None error_log: DF.LongText | None
initialize_doctypes_table: DF.Check initialize_doctypes_table_status: DF.Literal["Pending", "Completed", "Skipped"]
process_in_single_transaction: DF.Check process_in_single_transaction: DF.Check
reset_company_default_values: DF.Check reset_company_default_values_status: DF.Literal["Pending", "Completed", "Skipped"]
status: DF.Literal["Queued", "Running", "Failed", "Completed", "Cancelled"] status: DF.Literal["Queued", "Running", "Failed", "Completed", "Cancelled"]
# end: auto-generated types # end: auto-generated types
@@ -71,33 +165,90 @@ class TransactionDeletionRecord(Document):
def validate(self): def validate(self):
frappe.only_for("System Manager") frappe.only_for("System Manager")
self.validate_doctypes_to_be_ignored() self.validate_to_delete_list()
def validate_doctypes_to_be_ignored(self): def validate_to_delete_list(self):
doctypes_to_be_ignored_list = get_doctypes_to_be_ignored() """Validate To Delete list: existence, protection status, child table exclusion, duplicates"""
for doctype in self.doctypes_to_be_ignored: if not self.doctypes_to_delete:
if doctype.doctype_name not in doctypes_to_be_ignored_list: return
protected = _get_protected_doctypes_internal()
seen_combinations = set()
for item in self.doctypes_to_delete:
if not frappe.db.exists("DocType", item.doctype_name):
frappe.throw(_("DocType {0} does not exist").format(item.doctype_name))
# Check for duplicates using composite key
composite_key = (item.doctype_name, item.company_field or None)
if composite_key in seen_combinations:
field_desc = f" with company field '{item.company_field}'" if item.company_field else ""
frappe.throw( frappe.throw(
_( _("Duplicate entry: {0}{1}").format(item.doctype_name, field_desc),
"DocTypes should not be added manually to the 'Excluded DocTypes' table. You are only allowed to remove entries from it." title=_("Duplicate DocType"),
), )
title=_("Not Allowed"), seen_combinations.add(composite_key)
# Validate protected DocTypes
if item.doctype_name in protected:
frappe.throw(
_("Cannot delete protected core DocType: {0}").format(item.doctype_name),
title=_("Protected DocType"),
) )
is_child_table = frappe.db.get_value("DocType", item.doctype_name, "istable")
if is_child_table:
frappe.throw(
_(
"Cannot add child table {0} to deletion list. Child tables are automatically deleted with their parent DocTypes."
).format(item.doctype_name),
title=_("Child Table Not Allowed"),
)
is_virtual = frappe.db.get_value("DocType", item.doctype_name, "is_virtual")
if is_virtual:
frappe.throw(
_(
"Cannot delete virtual DocType: {0}. Virtual DocTypes do not have database tables."
).format(item.doctype_name),
title=_("Virtual DocType"),
)
# Validate company_field if specified
if item.company_field:
valid_company_fields = self._get_company_link_fields(item.doctype_name)
if item.company_field not in valid_company_fields:
frappe.throw(
_("Field '{0}' is not a valid Company link field for DocType {1}").format(
item.company_field, item.doctype_name
),
title=_("Invalid Company Field"),
)
def _is_any_doctype_in_deletion_list(self, doctypes_list):
"""Check if any DocType from the list is in the To Delete list"""
if not self.doctypes_to_delete:
return False
deletion_doctypes = {d.doctype_name for d in self.doctypes_to_delete}
return any(doctype in deletion_doctypes for doctype in doctypes_list)
def generate_job_name_for_task(self, task=None): def generate_job_name_for_task(self, task=None):
"""Generate unique job name for a specific task"""
method = self.task_to_internal_method_map[task] method = self.task_to_internal_method_map[task]
return f"{self.name}_{method}" return f"{self.name}_{method}"
def generate_job_name_for_next_tasks(self, task=None): def generate_job_name_for_next_tasks(self, task=None):
"""Generate job names for all tasks following the specified task"""
job_names = [] job_names = []
current_task_idx = list(self.task_to_internal_method_map).index(task) current_task_idx = list(self.task_to_internal_method_map).index(task)
for idx, task in enumerate(self.task_to_internal_method_map.keys(), 0): for idx, task in enumerate(self.task_to_internal_method_map.keys(), 0):
# generate job_name for next tasks
if idx > current_task_idx: if idx > current_task_idx:
job_names.append(self.generate_job_name_for_task(task)) job_names.append(self.generate_job_name_for_task(task))
return job_names return job_names
def generate_job_name_for_all_tasks(self): def generate_job_name_for_all_tasks(self):
"""Generate job names for all tasks in the deletion workflow"""
job_names = [] job_names = []
for task in self.task_to_internal_method_map.keys(): for task in self.task_to_internal_method_map.keys():
job_names.append(self.generate_job_name_for_task(task)) job_names.append(self.generate_job_name_for_task(task))
@@ -106,28 +257,28 @@ class TransactionDeletionRecord(Document):
def before_submit(self): def before_submit(self):
if queued_docs := frappe.db.get_all( if queued_docs := frappe.db.get_all(
"Transaction Deletion Record", "Transaction Deletion Record",
filters={"company": self.company, "status": ("in", ["Running", "Queued"]), "docstatus": 1}, filters={"status": ("in", ["Running", "Queued"]), "docstatus": 1},
pluck="name", pluck="name",
): ):
frappe.throw( frappe.throw(
_( _(
"Cannot enqueue multi docs for one company. {0} is already queued/running for company: {1}" "Cannot start deletion. Another deletion {0} is already queued/running. Please wait for it to complete."
).format( ).format(comma_and([get_link_to_form("Transaction Deletion Record", x) for x in queued_docs]))
comma_and([get_link_to_form("Transaction Deletion Record", x) for x in queued_docs]),
frappe.bold(self.company),
)
) )
if not self.doctypes_to_delete and not self.doctypes_to_be_ignored:
frappe.throw(_("Please generate To Delete list before submitting"))
if not self.doctypes_to_be_ignored: if not self.doctypes_to_be_ignored:
self.populate_doctypes_to_be_ignored_table() self.populate_doctypes_to_be_ignored_table()
def reset_task_flags(self): def reset_task_flags(self):
self.clear_notifications = 0 self.clear_notifications_status = "Pending"
self.delete_bin_data = 0 self.delete_bin_data_status = "Pending"
self.delete_leads_and_addresses = 0 self.delete_leads_and_addresses_status = "Pending"
self.delete_transactions = 0 self.delete_transactions_status = "Pending"
self.initialize_doctypes_table = 0 self.initialize_doctypes_table_status = "Pending"
self.reset_company_default_values = 0 self.reset_company_default_values_status = "Pending"
def before_save(self): def before_save(self):
self.status = "" self.status = ""
@@ -136,17 +287,288 @@ class TransactionDeletionRecord(Document):
def on_submit(self): def on_submit(self):
self.db_set("status", "Queued") self.db_set("status", "Queued")
self.start_deletion_tasks()
def on_cancel(self): def on_cancel(self):
self.db_set("status", "Cancelled") self.db_set("status", "Cancelled")
self._clear_deletion_cache()
def _set_deletion_cache(self):
"""Set Redis cache flags for per-doctype validation"""
for item in self.doctypes_to_delete:
frappe.cache.set_value(
f"deletion_running_doctype:{item.doctype_name}",
self.name,
expires_in_sec=DELETION_CACHE_TTL,
)
def _clear_deletion_cache(self):
"""Clear Redis cache flags"""
for item in self.doctypes_to_delete:
frappe.cache.delete_value(f"deletion_running_doctype:{item.doctype_name}")
def _get_child_tables(self, doctype_name):
"""Get list of child table DocType names for a given DocType
Args:
doctype_name: The parent DocType to check
Returns:
list: List of child table DocType names (Table field options)
"""
return frappe.get_all(
"DocField", filters={"parent": doctype_name, "fieldtype": "Table"}, pluck="options"
)
def _get_to_delete_row_infos(self, doctype_name, company_field=None, company=None):
"""Get child tables and document count for a To Delete list row
Args:
doctype_name: The DocType to get information for
company_field: Optional company field name to filter by
company: Optional company value (defaults to self.company)
Returns:
dict: {"child_doctypes": str, "document_count": int}
"""
company = company or self.company
child_tables = self._get_child_tables(doctype_name)
child_doctypes_str = ", ".join(child_tables) if child_tables else ""
if company_field and company:
doc_count = frappe.db.count(doctype_name, filters={company_field: company})
else:
doc_count = frappe.db.count(doctype_name)
return {
"child_doctypes": child_doctypes_str,
"document_count": doc_count,
}
def _has_company_field(self, doctype_name):
"""Check if DocType has a field specifically named 'company' linking to Company"""
return frappe.db.exists(
"DocField",
{"parent": doctype_name, "fieldname": "company", "fieldtype": "Link", "options": "Company"},
)
def _get_company_link_fields(self, doctype_name):
"""Get all Company Link field names for a DocType
Args:
doctype_name: The DocType to check
Returns:
list: List of field names that link to Company DocType, ordered by field index
"""
company_fields = frappe.get_all(
"DocField",
filters={"parent": doctype_name, "fieldtype": "Link", "options": "Company"},
pluck="fieldname",
order_by="idx",
)
return company_fields or []
@frappe.whitelist()
def generate_to_delete_list(self):
"""Generate To Delete list with one row per company field"""
self.doctypes_to_delete = []
excluded = [d.doctype_name for d in self.doctypes_to_be_ignored]
excluded.extend(_get_protected_doctypes_internal())
excluded.append(self.doctype) # Exclude self
# Get all DocTypes that have Company link fields
doctypes_with_company_field = frappe.get_all(
"DocField",
filters={"fieldtype": "Link", "options": "Company"},
pluck="parent",
distinct=True,
)
# Filter to get only valid DocTypes (not child tables, not virtual, not excluded)
doctypes_with_company = []
for doctype_name in doctypes_with_company_field:
if doctype_name in excluded:
continue
# Check if doctype exists and is not a child table or virtual
if frappe.db.exists("DocType", doctype_name):
meta = frappe.get_meta(doctype_name)
if not meta.istable and not meta.is_virtual:
doctypes_with_company.append(doctype_name)
for doctype_name in doctypes_with_company:
# Get ALL company fields for this DocType
company_fields = self._get_company_link_fields(doctype_name)
# Get child tables once (same for all company fields of this DocType)
child_tables = self._get_child_tables(doctype_name)
child_doctypes_str = ", ".join(child_tables) if child_tables else ""
for company_field in company_fields:
doc_count = frappe.db.count(doctype_name, {company_field: self.company})
self.append(
"doctypes_to_delete",
{
"doctype_name": doctype_name,
"company_field": company_field,
"document_count": doc_count,
"child_doctypes": child_doctypes_str,
},
)
self.save()
return {"count": len(self.doctypes_to_delete)}
@frappe.whitelist()
def populate_doctype_details(self, doctype_name, company=None, company_field=None):
"""Get child DocTypes and document count for specified DocType
Args:
doctype_name: The DocType to get details for
company: Optional company value for filtering (defaults to self.company)
company_field: Optional company field name to use for filtering
"""
frappe.only_for("System Manager")
if not doctype_name:
return {}
if not frappe.db.exists("DocType", doctype_name):
frappe.throw(_("DocType {0} does not exist").format(doctype_name))
is_child_table = frappe.db.get_value("DocType", doctype_name, "istable")
if is_child_table:
return {
"child_doctypes": "",
"document_count": 0,
"error": _("{0} is a child table and will be deleted automatically with its parent").format(
doctype_name
),
}
try:
return self._get_to_delete_row_infos(doctype_name, company_field=company_field, company=company)
except Exception as e:
frappe.log_error(
f"Error in populate_doctype_details for {doctype_name}: {e!s}", "Transaction Deletion Record"
)
return {
"child_doctypes": "",
"document_count": 0,
"error": _("Unable to fetch DocType details. Please contact system administrator."),
}
def export_to_delete_template_method(self):
"""Export To Delete list as CSV template"""
if not self.doctypes_to_delete:
frappe.throw(_("Generate To Delete list first"))
import csv
from io import StringIO
output = StringIO()
writer = csv.writer(output)
writer.writerow(["doctype_name", "company_field", "child_doctypes"])
for item in self.doctypes_to_delete:
writer.writerow([item.doctype_name, item.company_field or "", item.child_doctypes or ""])
frappe.response["result"] = output.getvalue()
frappe.response["type"] = "csv"
frappe.response[
"doctype"
] = f"deletion_template_{self.company}_{frappe.utils.now_datetime().strftime('%Y%m%d')}"
def import_to_delete_template_method(self, csv_content):
"""Import CSV template and regenerate counts"""
import csv
from io import StringIO
reader = csv.DictReader(StringIO(csv_content))
if "doctype_name" not in (reader.fieldnames or []):
frappe.throw(_("Invalid CSV format. Expected column: doctype_name"))
self.doctypes_to_delete = []
protected = _get_protected_doctypes_internal()
imported_count = 0
skipped = []
for row in reader:
doctype_name = row.get("doctype_name", "").strip()
company_field = row.get("company_field", "").strip() or None
if not doctype_name:
continue
if doctype_name in protected:
skipped.append(_("{0}: Protected DocType").format(doctype_name))
continue
if not frappe.db.exists("DocType", doctype_name):
skipped.append(_("{0}: Not found").format(doctype_name))
continue
is_child = frappe.db.get_value("DocType", doctype_name, "istable")
if is_child:
skipped.append(_("{0}: Child table (auto-deleted with parent)").format(doctype_name))
continue
is_virtual = frappe.db.get_value("DocType", doctype_name, "is_virtual")
if is_virtual:
skipped.append(_("{0}: Virtual DocType (no database table)").format(doctype_name))
continue
db_company_fields = self._get_company_link_fields(doctype_name)
import_company_field = ""
if not db_company_fields: # Case no company field exists
details = self._get_to_delete_row_infos(doctype_name)
elif (
company_field and company_field in db_company_fields
): # Case it is provided by export and valid
details = self._get_to_delete_row_infos(doctype_name, company_field)
import_company_field = company_field
else: # Company field exists but not provided by export or invalid
if "company" in db_company_fields: # Check if 'company' is a valid field
details = self._get_to_delete_row_infos(doctype_name, "company")
import_company_field = "company"
else: # Fallback to first valid company field
details = self._get_to_delete_row_infos(doctype_name, db_company_fields[0])
import_company_field = db_company_fields[0]
self.append(
"doctypes_to_delete",
{
"doctype_name": doctype_name,
"company_field": import_company_field,
"document_count": details["document_count"],
"child_doctypes": details["child_doctypes"],
},
)
imported_count += 1
self.save()
if skipped:
frappe.msgprint(
_("Skipped {0} DocType(s):<br>{1}").format(len(skipped), "<br>".join(skipped)),
title=_("Import Summary"),
indicator="orange",
)
return {"imported": imported_count, "skipped": len(skipped)}
def enqueue_task(self, task: str | None = None): def enqueue_task(self, task: str | None = None):
"""Enqueue a deletion task for background execution"""
if task and task in self.task_to_internal_method_map: if task and task in self.task_to_internal_method_map:
# make sure that none of next tasks are already running
job_names = self.generate_job_name_for_next_tasks(task=task) job_names = self.generate_job_name_for_next_tasks(task=task)
self.validate_running_task_for_doc(job_names=job_names) self.validate_running_task_for_doc(job_names=job_names)
# Generate Job Id to uniquely identify each task for this document
job_id = self.generate_job_name_for_task(task) job_id = self.generate_job_name_for_task(task)
if self.process_in_single_transaction: if self.process_in_single_transaction:
@@ -176,12 +598,13 @@ class TransactionDeletionRecord(Document):
message = "Traceback: <br>" + traceback message = "Traceback: <br>" + traceback
frappe.db.set_value(self.doctype, self.name, "error_log", message) frappe.db.set_value(self.doctype, self.name, "error_log", message)
frappe.db.set_value(self.doctype, self.name, "status", "Failed") frappe.db.set_value(self.doctype, self.name, "status", "Failed")
self._clear_deletion_cache()
def delete_notifications(self): def delete_notifications(self):
self.validate_doc_status() self.validate_doc_status()
if not self.clear_notifications: if self.clear_notifications_status == "Pending":
clear_notifications() clear_notifications()
self.db_set("clear_notifications", 1) self.db_set("clear_notifications_status", "Completed")
self.enqueue_task(task="Initialize Summary Table") self.enqueue_task(task="Initialize Summary Table")
def populate_doctypes_to_be_ignored_table(self): def populate_doctypes_to_be_ignored_table(self):
@@ -215,23 +638,46 @@ class TransactionDeletionRecord(Document):
def start_deletion_tasks(self): def start_deletion_tasks(self):
# This method is the entry point for the chain of events that follow # This method is the entry point for the chain of events that follow
self.db_set("status", "Running") self.db_set("status", "Running")
self._set_deletion_cache()
self.enqueue_task(task="Delete Bins") self.enqueue_task(task="Delete Bins")
def delete_bins(self): def delete_bins(self):
self.validate_doc_status() self.validate_doc_status()
if not self.delete_bin_data: if self.delete_bin_data_status == "Pending":
stock_related_doctypes = [
"Item",
"Warehouse",
"Stock Entry",
"Delivery Note",
"Purchase Receipt",
"Stock Reconciliation",
"Material Request",
"Purchase Invoice",
"Sales Invoice",
]
if not self._is_any_doctype_in_deletion_list(stock_related_doctypes):
self.db_set("delete_bin_data_status", "Skipped")
self.enqueue_task(task="Delete Leads and Addresses")
return
frappe.db.sql( frappe.db.sql(
"""delete from `tabBin` where warehouse in """delete from `tabBin` where warehouse in
(select name from tabWarehouse where company=%s)""", (select name from tabWarehouse where company=%s)""",
self.company, self.company,
) )
self.db_set("delete_bin_data", 1) self.db_set("delete_bin_data_status", "Completed")
self.enqueue_task(task="Delete Leads and Addresses") self.enqueue_task(task="Delete Leads and Addresses")
def delete_lead_addresses(self): def delete_lead_addresses(self):
"""Delete addresses to which leads are linked""" """Delete addresses to which leads are linked"""
self.validate_doc_status() self.validate_doc_status()
if not self.delete_leads_and_addresses: if self.delete_leads_and_addresses_status == "Pending":
if not self._is_any_doctype_in_deletion_list(["Lead"]):
self.db_set("delete_leads_and_addresses_status", "Skipped")
self.enqueue_task(task="Reset Company Values")
return
leads = frappe.db.get_all("Lead", filters={"company": self.company}, pluck="name") leads = frappe.db.get_all("Lead", filters={"company": self.company}, pluck="name")
addresses = [] addresses = []
if leads: if leads:
@@ -268,54 +714,94 @@ class TransactionDeletionRecord(Document):
customer = qb.DocType("Customer") customer = qb.DocType("Customer")
qb.update(customer).set(customer.lead_name, None).where(customer.lead_name.isin(leads)).run() qb.update(customer).set(customer.lead_name, None).where(customer.lead_name.isin(leads)).run()
self.db_set("delete_leads_and_addresses", 1) self.db_set("delete_leads_and_addresses_status", "Completed")
self.enqueue_task(task="Reset Company Values") self.enqueue_task(task="Reset Company Values")
def reset_company_values(self): def reset_company_values(self):
self.validate_doc_status() self.validate_doc_status()
if not self.reset_company_default_values: if self.reset_company_default_values_status == "Pending":
sales_related_doctypes = [
"Sales Order",
"Sales Invoice",
"Quotation",
"Delivery Note",
]
if not self._is_any_doctype_in_deletion_list(sales_related_doctypes):
self.db_set("reset_company_default_values_status", "Skipped")
self.enqueue_task(task="Clear Notifications")
return
company_obj = frappe.get_doc("Company", self.company) company_obj = frappe.get_doc("Company", self.company)
company_obj.total_monthly_sales = 0 company_obj.total_monthly_sales = 0
company_obj.sales_monthly_history = None company_obj.sales_monthly_history = None
company_obj.save() company_obj.save()
self.db_set("reset_company_default_values", 1) self.db_set("reset_company_default_values_status", "Completed")
self.enqueue_task(task="Clear Notifications") self.enqueue_task(task="Clear Notifications")
def initialize_doctypes_to_be_deleted_table(self): def initialize_doctypes_to_be_deleted_table(self):
"""Initialize deletion table from To Delete list or fall back to original logic"""
self.validate_doc_status() self.validate_doc_status()
if not self.initialize_doctypes_table: if self.initialize_doctypes_table_status == "Pending":
doctypes_to_be_ignored_list = self.get_doctypes_to_be_ignored_list() # Use To Delete list if available (new behavior)
docfields = self.get_doctypes_with_company_field(doctypes_to_be_ignored_list) if not self.doctypes_to_delete:
frappe.throw(
_("No DocTypes in To Delete list. Please generate or import the list before submitting."),
title=_("Empty To Delete List"),
)
tables = self.get_all_child_doctypes() tables = self.get_all_child_doctypes()
for docfield in docfields:
if docfield["parent"] != self.doctype: for to_delete_item in self.doctypes_to_delete:
no_of_docs = self.get_number_of_docs_linked_with_specified_company( if to_delete_item.document_count > 0:
docfield["parent"], docfield["fieldname"] # Add parent DocType only - child tables are handled automatically
# by delete_child_tables() when the parent is deleted
# Use company_field directly from To Delete item
self.populate_doctypes_table(
tables, to_delete_item.doctype_name, to_delete_item.company_field, 0
) )
if no_of_docs > 0: self.db_set("initialize_doctypes_table_status", "Completed")
# Initialize
self.populate_doctypes_table(tables, docfield["parent"], docfield["fieldname"], 0)
self.db_set("initialize_doctypes_table", 1)
self.enqueue_task(task="Delete Transactions") self.enqueue_task(task="Delete Transactions")
def delete_company_transactions(self): def delete_company_transactions(self):
self.validate_doc_status() self.validate_doc_status()
if not self.delete_transactions: if self.delete_transactions_status == "Pending":
doctypes_to_be_ignored_list = self.get_doctypes_to_be_ignored_list() protected_doctypes = _get_protected_doctypes_internal()
self.get_doctypes_with_company_field(doctypes_to_be_ignored_list)
self.get_all_child_doctypes()
for docfield in self.doctypes: for docfield in self.doctypes:
if docfield.doctype_name != self.doctype and not docfield.done: if docfield.doctype_name != self.doctype and not docfield.done:
no_of_docs = self.get_number_of_docs_linked_with_specified_company( if docfield.doctype_name in protected_doctypes:
docfield.doctype_name, docfield.docfield_name error_msg = (
) f"CRITICAL: Attempted to delete protected DocType: {docfield.doctype_name}"
if no_of_docs > 0:
reference_docs = frappe.get_all(
docfield.doctype_name,
filters={docfield.docfield_name: self.company},
limit=self.batch_size,
) )
frappe.log_error(error_msg, "Transaction Deletion Security")
frappe.throw(
_("Cannot delete protected core DocType: {0}").format(docfield.doctype_name),
title=_("Protected DocType"),
)
# Get company_field from stored value (could be any Company link field)
company_field = docfield.docfield_name
if company_field:
no_of_docs = self.get_number_of_docs_linked_with_specified_company(
docfield.doctype_name, company_field
)
else:
no_of_docs = frappe.db.count(docfield.doctype_name)
if no_of_docs > 0:
if company_field:
reference_docs = frappe.get_all(
docfield.doctype_name,
filters={company_field: self.company},
fields=["name"],
limit=self.batch_size,
)
else:
reference_docs = frappe.get_all(
docfield.doctype_name, fields=["name"], limit=self.batch_size
)
reference_doc_names = [r.name for r in reference_docs] reference_doc_names = [r.name for r in reference_docs]
self.delete_version_log(docfield.doctype_name, reference_doc_names) self.delete_version_log(docfield.doctype_name, reference_doc_names)
@@ -329,26 +815,38 @@ class TransactionDeletionRecord(Document):
processed = int(docfield.no_of_docs) + len(reference_doc_names) processed = int(docfield.no_of_docs) + len(reference_doc_names)
frappe.db.set_value(docfield.doctype, docfield.name, "no_of_docs", processed) frappe.db.set_value(docfield.doctype, docfield.name, "no_of_docs", processed)
else: else:
# reset naming series
naming_series = frappe.db.get_value("DocType", docfield.doctype_name, "autoname") naming_series = frappe.db.get_value("DocType", docfield.doctype_name, "autoname")
if naming_series: if naming_series:
if "#" in naming_series: if "#" in naming_series:
self.update_naming_series(naming_series, docfield.doctype_name) self.update_naming_series(naming_series, docfield.doctype_name)
frappe.db.set_value(docfield.doctype, docfield.name, "done", 1) frappe.db.set_value(docfield.doctype, docfield.name, "done", 1)
to_delete_row = frappe.db.get_value(
"Transaction Deletion Record To Delete",
{
"parent": self.name,
"doctype_name": docfield.doctype_name,
"company_field": company_field,
},
"name",
)
if to_delete_row:
frappe.db.set_value(
"Transaction Deletion Record To Delete", to_delete_row, "deleted", 1
)
pending_doctypes = frappe.db.get_all( pending_doctypes = frappe.db.get_all(
"Transaction Deletion Record Details", "Transaction Deletion Record Details",
filters={"parent": self.name, "done": 0}, filters={"parent": self.name, "done": 0},
pluck="doctype_name", pluck="doctype_name",
) )
if pending_doctypes: if pending_doctypes:
# as method is enqueued after commit, calling itself will not make validate_doc_status to throw
# recursively call this task to delete all transactions
self.enqueue_task(task="Delete Transactions") self.enqueue_task(task="Delete Transactions")
else: else:
self.db_set("status", "Completed") self.db_set("status", "Completed")
self.db_set("delete_transactions", 1) self.db_set("delete_transactions_status", "Completed")
self.db_set("error_log", None) self.db_set("error_log", None)
self._clear_deletion_cache()
def get_doctypes_to_be_ignored_list(self): def get_doctypes_to_be_ignored_list(self):
doctypes_to_be_ignored_list = frappe.get_all( doctypes_to_be_ignored_list = frappe.get_all(
@@ -378,18 +876,33 @@ class TransactionDeletionRecord(Document):
def get_number_of_docs_linked_with_specified_company(self, doctype, company_fieldname): def get_number_of_docs_linked_with_specified_company(self, doctype, company_fieldname):
return frappe.db.count(doctype, {company_fieldname: self.company}) return frappe.db.count(doctype, {company_fieldname: self.company})
def populate_doctypes_table(self, tables, doctype, fieldname, no_of_docs): def get_company_field(self, doctype_name):
"""Get company field name for a DocType"""
return frappe.db.get_value(
"DocField",
{"parent": doctype_name, "fieldtype": "Link", "options": "Company"},
"fieldname",
)
def populate_doctypes_table(self, tables, doctype, company_field, no_of_docs):
"""Add doctype to processing tracker
Args:
tables: List of child table DocType names (to exclude)
doctype: DocType name to track
company_field: Company link field name (or None)
no_of_docs: Initial count
"""
self.flags.ignore_validate_update_after_submit = True self.flags.ignore_validate_update_after_submit = True
if doctype not in tables: if doctype not in tables:
self.append( self.append(
"doctypes", {"doctype_name": doctype, "docfield_name": fieldname, "no_of_docs": no_of_docs} "doctypes",
{"doctype_name": doctype, "docfield_name": company_field, "no_of_docs": no_of_docs},
) )
self.save(ignore_permissions=True) self.save(ignore_permissions=True)
def delete_child_tables(self, doctype, reference_doc_names): def delete_child_tables(self, doctype, reference_doc_names):
child_tables = frappe.get_all( child_tables = self._get_child_tables(doctype)
"DocField", filters={"fieldtype": "Table", "parent": doctype}, pluck="options"
)
for table in child_tables: for table in child_tables:
frappe.db.delete(table, {"parent": ["in", reference_doc_names]}) frappe.db.delete(table, {"parent": ["in", reference_doc_names]})
@@ -397,22 +910,52 @@ class TransactionDeletionRecord(Document):
def delete_docs_linked_with_specified_company(self, doctype, reference_doc_names): def delete_docs_linked_with_specified_company(self, doctype, reference_doc_names):
frappe.db.delete(doctype, {"name": ("in", reference_doc_names)}) frappe.db.delete(doctype, {"name": ("in", reference_doc_names)})
def update_naming_series(self, naming_series, doctype_name): @staticmethod
def get_naming_series_prefix(naming_series: str, doctype_name: str) -> str:
"""Extract the static prefix from an autoname pattern.
Args:
naming_series: The autoname pattern (e.g., "PREFIX.####", "format:PRE-{####}")
doctype_name: DocType name for error logging
Returns:
The static prefix before the counter placeholders
"""
if "." in naming_series: if "." in naming_series:
prefix, hashes = naming_series.rsplit(".", 1) prefix = naming_series.rsplit(".", 1)[0]
elif "{" in naming_series:
prefix = naming_series.rsplit("{", 1)[0]
else: else:
prefix, hashes = naming_series.rsplit("{", 1) # Fallback for unexpected patterns (shouldn't happen with valid Frappe naming series)
last = frappe.db.sql( frappe.log_error(
f"""select max(name) from `tab{doctype_name}` title=_("Unexpected Naming Series Pattern"),
where name like %s""", message=_(
prefix + "%", "Naming series '{0}' for DocType '{1}' does not contain standard '.' or '{{' separator. Using fallback extraction."
).format(naming_series, doctype_name),
)
prefix = naming_series.split("#", 1)[0] if "#" in naming_series else naming_series
return prefix
def update_naming_series(self, naming_series, doctype_name):
# Derive a static prefix from the autoname pattern
prefix = self.get_naming_series_prefix(naming_series, doctype_name)
# Find the highest number used in the naming series to reset the counter
doctype_table = qb.DocType(doctype_name)
result = (
qb.from_(doctype_table)
.select(Max(doctype_table.name))
.where(doctype_table.name.like(prefix + "%"))
.run()
) )
if last and last[0][0]:
last = cint(last[0][0].replace(prefix, "")) if result and result[0][0]:
last = cint(result[0][0].replace(prefix, ""))
else: else:
last = 0 last = 0
frappe.db.sql("""update `tabSeries` set current = %s where name=%s""", (last, prefix)) frappe.db.set_value("Series", prefix, "current", last, update_modified=False)
def delete_version_log(self, doctype, docnames): def delete_version_log(self, doctype, docnames):
versions = qb.DocType("Version") versions = qb.DocType("Version")
@@ -487,15 +1030,61 @@ def get_doctypes_to_be_ignored():
return doctypes_to_be_ignored return doctypes_to_be_ignored
@frappe.whitelist()
def export_to_delete_template(name):
"""Export To Delete list as CSV via URL access"""
frappe.only_for("System Manager")
doc = frappe.get_doc("Transaction Deletion Record", name)
doc.check_permission("read")
return doc.export_to_delete_template_method()
@frappe.whitelist()
def process_import_template(transaction_deletion_record_name, file_url):
"""Import CSV template and populate To Delete list"""
import os
doc = frappe.get_doc("Transaction Deletion Record", transaction_deletion_record_name)
doc.check_permission("write")
if not file_url or ".." in file_url:
frappe.throw(_("Invalid file URL"))
try:
file_doc = frappe.get_doc("File", {"file_url": file_url})
except frappe.DoesNotExistError:
frappe.throw(_("File not found"))
if (
file_doc.attached_to_doctype != "Transaction Deletion Record"
or file_doc.attached_to_name != transaction_deletion_record_name
):
frappe.throw(_("File does not belong to this Transaction Deletion Record"))
if not file_doc.file_name or not file_doc.file_name.lower().endswith(".csv"):
frappe.throw(_("Only CSV files are allowed"))
file_path = file_doc.get_full_path()
if not os.path.isfile(file_path):
frappe.throw(_("File not found on server"))
with open(file_path, encoding="utf-8") as f:
csv_content = f.read()
return doc.import_to_delete_template_method(csv_content)
@frappe.whitelist() @frappe.whitelist()
@request_cache @request_cache
def is_deletion_doc_running(company: str | None = None, err_msg: str | None = None): def is_deletion_doc_running(company: str | None = None, err_msg: str | None = None):
if not company: """Check if any deletion is running globally
return
The company parameter is kept for backwards compatibility but is now ignored.
"""
running_deletion_job = frappe.db.get_value( running_deletion_job = frappe.db.get_value(
"Transaction Deletion Record", "Transaction Deletion Record",
{"docstatus": 1, "company": company, "status": "Running"}, {"docstatus": 1, "status": ("in", ["Running", "Queued"])},
"name", "name",
) )
@@ -504,17 +1093,28 @@ def is_deletion_doc_running(company: str | None = None, err_msg: str | None = No
frappe.throw( frappe.throw(
title=_("Deletion in Progress!"), title=_("Deletion in Progress!"),
msg=_("Transaction Deletion Document: {0} is running for this Company. {1}").format( msg=_("Transaction Deletion Record {0} is already running. {1}").format(
get_link_to_form("Transaction Deletion Record", running_deletion_job), err_msg or "" get_link_to_form("Transaction Deletion Record", running_deletion_job), err_msg or ""
), ),
) )
def check_for_running_deletion_job(doc, method=None): def check_for_running_deletion_job(doc, method=None):
# Check if DocType has 'company' field """Hook function called on document validate - checks Redis cache for running deletions"""
if doc.doctype in LEDGER_ENTRY_DOCTYPES or not doc.meta.has_field("company"): if doc.doctype in LEDGER_ENTRY_DOCTYPES:
return return
is_deletion_doc_running( if doc.doctype in PROTECTED_CORE_DOCTYPES:
doc.company, _("Cannot make any transactions until the deletion job is completed") return
)
deletion_name = frappe.cache.get_value(f"deletion_running_doctype:{doc.doctype}")
if deletion_name:
frappe.throw(
title=_("Deletion in Progress!"),
msg=_(
"Transaction Deletion Record {0} is currently deleting {1}. Cannot save documents until deletion completes."
).format(
get_link_to_form("Transaction Deletion Record", deletion_name), frappe.bold(doc.doctype)
),
)

View File

@@ -17,17 +17,19 @@
"reqd": 1 "reqd": 1
} }
], ],
"grid_page_length": 50,
"index_web_pages_for_search": 1, "index_web_pages_for_search": 1,
"istable": 1, "istable": 1,
"links": [], "links": [],
"modified": "2024-03-27 13:10:55.128861", "modified": "2025-11-14 16:17:47.755531",
"modified_by": "Administrator", "modified_by": "Administrator",
"module": "Setup", "module": "Setup",
"name": "Transaction Deletion Record Item", "name": "Transaction Deletion Record Item",
"owner": "Administrator", "owner": "Administrator",
"permissions": [], "permissions": [],
"row_format": "Dynamic",
"sort_field": "creation", "sort_field": "creation",
"sort_order": "DESC", "sort_order": "DESC",
"states": [], "states": [],
"track_changes": 1 "track_changes": 1
} }

View File

@@ -0,0 +1,67 @@
{
"actions": [],
"creation": "2025-11-14 00:00:00",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"doctype_name",
"company_field",
"document_count",
"child_doctypes",
"deleted"
],
"fields": [
{
"fieldname": "doctype_name",
"fieldtype": "Link",
"in_list_view": 1,
"label": "DocType",
"options": "DocType"
},
{
"description": "Company link field name used for filtering (optional - leave empty to delete all records)",
"fieldname": "company_field",
"fieldtype": "Data",
"in_list_view": 1,
"label": "Company Field"
},
{
"fieldname": "document_count",
"fieldtype": "Int",
"in_list_view": 1,
"label": "Document Count",
"read_only": 1
},
{
"description": "Child tables that will also be deleted",
"fieldname": "child_doctypes",
"fieldtype": "Small Text",
"in_list_view": 1,
"label": "Child DocTypes",
"read_only": 1
},
{
"default": "0",
"fieldname": "deleted",
"fieldtype": "Check",
"in_list_view": 1,
"label": "Deleted",
"read_only": 1
}
],
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2025-11-14 16:17:04.494126",
"modified_by": "Administrator",
"module": "Setup",
"name": "Transaction Deletion Record To Delete",
"owner": "Administrator",
"permissions": [],
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": [],
"track_changes": 1
}

View File

@@ -0,0 +1,27 @@
# Copyright (c) 2025, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
# import frappe
from frappe.model.document import Document
class TransactionDeletionRecordToDelete(Document):
# begin: auto-generated types
# This code is auto-generated. Do not modify anything in this block.
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from frappe.types import DF
child_doctypes: DF.SmallText | None
company_field: DF.Data | None
deleted: DF.Check
doctype_name: DF.Link | None
document_count: DF.Int
parent: DF.Data
parentfield: DF.Data
parenttype: DF.Data
# end: auto-generated types
pass

View File

@@ -0,0 +1,230 @@
# Transaction Deletion CSV Import Logic - Updated Behavior
## Auto-Detection of Company Field
When importing a CSV without a `company_field` column or with empty values, the system uses smart auto-detection:
### Priority Order:
1. **"company" field** (most common convention)
- Check if a field named `company` exists that links to Company DocType
- ✅ Use "company" if found
2. **First Company link field** (custom fields)
- If no "company" field, get all fields linking to Company DocType
- ✅ Use the first one (sorted by field index)
3. **No company field** (DocTypes without company filtering)
- If no Company link fields exist at all
- ✅ Leave `company_field` as None/empty
- ✅ Delete ALL records (no company filtering)
## Import CSV Format
### Minimal Format (Auto-Detection)
```csv
doctype_name,child_doctypes
Sales Order,Sales Order Item
Note,
Task,
```
**Result:**
- `Sales Order`: Auto-detects "company" field → Filters by company
- `Note`: No company field → Deletes all Note records
- `Task`: Has "company" field → Filters by company
### Explicit Format (Recommended)
```csv
doctype_name,company_field,child_doctypes
Sales Order,company,Sales Order Item
Sales Contract,primary_company,Sales Contract Item
Sales Contract,billing_company,Sales Contract Item
Note,,
```
**Result:**
- `Sales Order`: Uses "company" field explicitly
- `Sales Contract` (row 1): Uses "primary_company" field
- `Sales Contract` (row 2): Uses "billing_company" field (separate row!)
- `Note`: No company field, deletes all records
### Multiple Company Fields Example
```csv
doctype_name,company_field,child_doctypes
Customer Invoice,head_office,Customer Invoice Item
Customer Invoice,billing_company,Customer Invoice Item
```
**Deletion Process:**
1. Row 1 deletes: `WHERE head_office = 'ABC Company'`
2. Row 2 deletes: `WHERE billing_company = 'ABC Company'`
3. Documents with both fields = ABC get deleted in first pass
4. Documents with only billing_company = ABC get deleted in second pass
## Validation Rules
### ✅ Accepted Cases
1. **DocType with "company" field** - Auto-detected
2. **DocType with custom Company link field** - Auto-detected (first field used)
3. **DocType with multiple Company fields** - Auto-detected (first field used), but user can add multiple rows
4. **DocType with NO Company fields** - Accepted! Deletes ALL records
5. **Explicit company_field provided** - Validated and used
### ❌ Rejected Cases
1. **Protected DocTypes** - User, Role, DocType, etc.
2. **Child tables** - Auto-deleted with parent
3. **Virtual DocTypes** - No database table
4. **Invalid company_field** - Field doesn't exist or isn't a Company link
5. **DocType doesn't exist** - Not found in system
## Code Flow
```python
# 1. Read company_field from CSV (may be empty)
company_field = row.get("company_field", "").strip()
# 2. Auto-detect if not provided
if not company_field:
# Try "company" first
if exists("company" field linking to Company):
company_field = "company"
else:
# Check for other Company link fields
company_fields = get_all_company_link_fields()
if company_fields:
company_field = company_fields[0] # Use first
# else: company_field stays empty
# 3. Validate if company_field was provided/detected
if company_field:
if not is_valid_company_link_field(company_field):
skip_with_error()
# 4. Count documents
if company_field:
count = count(WHERE company_field = self.company)
else:
count = count(all records)
# 5. Store in To Delete list
append({
"doctype_name": doctype_name,
"company_field": company_field or None, # Store None if empty
"document_count": count
})
```
## Examples
### Example 1: Standard DocType with "company" Field
**CSV:**
```csv
doctype_name,company_field,child_doctypes
Sales Order,,
```
**Auto-Detection:**
- Finds "company" field linking to Company
- Sets `company_field = "company"`
- Counts: `WHERE company = 'Test Company'`
- Result: Deletes only Test Company's Sales Orders
### Example 2: Custom Company Field
**CSV:**
```csv
doctype_name,company_field,child_doctypes
Project Contract,,
```
**Auto-Detection:**
- No "company" field found
- Finds "contracting_company" field linking to Company
- Sets `company_field = "contracting_company"`
- Counts: `WHERE contracting_company = 'Test Company'`
- Result: Deletes only Test Company's Project Contracts
### Example 3: No Company Field (Global DocType)
**CSV:**
```csv
doctype_name,company_field,child_doctypes
Note,,
Global Settings,,
```
**Auto-Detection:**
- No Company link fields found
- Sets `company_field = None`
- Counts: All records
- Result: Deletes ALL Note and Global Settings records
### Example 4: Multiple Company Fields (Explicit)
**CSV:**
```csv
doctype_name,company_field,child_doctypes
Sales Contract,primary_company,Sales Contract Item
Sales Contract,billing_company,Sales Contract Item
```
**No Auto-Detection:**
- Row 1: Uses "primary_company" explicitly
- Row 2: Uses "billing_company" explicitly
- Both rows validated as valid Company link fields
- Result: Two separate deletion passes
### Example 5: Mixed Approaches
**CSV:**
```csv
doctype_name,company_field,child_doctypes
Sales Order,,Sales Order Item
Sales Contract,billing_company,Sales Contract Item
Note,,
```
**Result:**
- Row 1: Auto-detects "company" field
- Row 2: Uses "billing_company" explicitly
- Row 3: No company field (deletes all)
## User Benefits
**Flexible**: Supports auto-detection and explicit specification
**Safe**: Validates all fields before processing
**Clear**: Empty company_field means "delete all"
**Powerful**: Can target specific company fields in multi-company setups
**Backward Compatible**: Old CSVs (without company_field column) still work
## Migration from Old Format
**Old CSV (without company_field):**
```csv
doctype_name,child_doctypes
Sales Order,Sales Order Item
```
**New System Behavior:**
- Auto-detects "company" field
- Works identically to before
- ✅ Backward compatible
**New CSV (with company_field):**
```csv
doctype_name,company_field,child_doctypes
Sales Order,company,Sales Order Item
```
**Benefits:**
- Explicit and clear
- Supports multiple rows per DocType
- Can specify custom company fields
---
*Generated for Transaction Deletion Record enhancement*