feat: Transaction Deletion Record (#25354)

Co-authored-by: Saqib <nextchamp.saqib@gmail.com>
This commit is contained in:
Ganga Manoj 2021-05-10 14:02:58 +05:30 committed by GitHub
parent 6e179c3092
commit f2eb8dd1d5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 411 additions and 129 deletions

View File

@ -100,6 +100,10 @@ status_map = {
["Queued", "eval:self.status == 'Queued'"],
["Failed", "eval:self.status == 'Failed'"],
["Cancelled", "eval:self.docstatus == 2"],
],
"Transaction Deletion Record": [
["Draft", None],
["Completed", "eval:self.docstatus == 1"],
]
}

View File

@ -169,9 +169,9 @@ frappe.ui.form.on("Company", {
return;
}
frappe.call({
method: "erpnext.setup.doctype.company.delete_company_transactions.delete_company_transactions",
method: "erpnext.setup.doctype.company.company.create_transaction_deletion_request",
args: {
company_name: data.company_name
company: data.company_name
},
freeze: true,
callback: function(r, rt) {

View File

@ -613,4 +613,13 @@ def get_default_company_address(name, sort_key='is_primary_address', existing_ad
if out:
return sorted(out, key = functools.cmp_to_key(lambda x,y: cmp(y[1], x[1])))[0][0]
else:
return None
return None
@frappe.whitelist()
def create_transaction_deletion_request(company):
tdr = frappe.get_doc({
'doctype': 'Transaction Deletion Record',
'company': company
})
tdr.insert()
tdr.submit()

View File

@ -1,117 +0,0 @@
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
from __future__ import unicode_literals
import frappe
from frappe.utils import cint
from frappe import _
from frappe.desk.notifications import clear_notifications
import functools
@frappe.whitelist()
def delete_company_transactions(company_name):
frappe.only_for("System Manager")
doc = frappe.get_doc("Company", company_name)
if frappe.session.user != doc.owner and frappe.session.user != 'Administrator':
frappe.throw(_("Transactions can only be deleted by the creator of the Company"),
frappe.PermissionError)
delete_bins(company_name)
delete_lead_addresses(company_name)
for doctype in frappe.db.sql_list("""select parent from
tabDocField where fieldtype='Link' and options='Company'"""):
if doctype not in ("Account", "Cost Center", "Warehouse", "Budget",
"Party Account", "Employee", "Sales Taxes and Charges Template",
"Purchase Taxes and Charges Template", "POS Profile", "BOM",
"Company", "Bank Account", "Item Tax Template", "Mode Of Payment", "Mode of Payment Account",
"Item Default", "Customer", "Supplier", "GST Account"):
delete_for_doctype(doctype, company_name)
# reset company values
doc.total_monthly_sales = 0
doc.sales_monthly_history = None
doc.save()
# Clear notification counts
clear_notifications()
def delete_for_doctype(doctype, company_name):
meta = frappe.get_meta(doctype)
company_fieldname = meta.get("fields", {"fieldtype": "Link",
"options": "Company"})[0].fieldname
if not meta.issingle:
if not meta.istable:
# delete communication
delete_communications(doctype, company_name, company_fieldname)
# delete children
for df in meta.get_table_fields():
frappe.db.sql("""delete from `tab{0}` where parent in
(select name from `tab{1}` where `{2}`=%s)""".format(df.options,
doctype, company_fieldname), company_name)
#delete version log
frappe.db.sql("""delete from `tabVersion` where ref_doctype=%s and docname in
(select name from `tab{0}` where `{1}`=%s)""".format(doctype,
company_fieldname), (doctype, company_name))
# delete parent
frappe.db.sql("""delete from `tab{0}`
where {1}= %s """.format(doctype, company_fieldname), company_name)
# reset series
naming_series = meta.get_field("naming_series")
if naming_series and naming_series.options:
prefixes = sorted(naming_series.options.split("\n"),
key=functools.cmp_to_key(lambda a, b: len(b) - len(a)))
for prefix in prefixes:
if prefix:
last = frappe.db.sql("""select max(name) from `tab{0}`
where name like %s""".format(doctype), prefix + "%")
if last and last[0][0]:
last = cint(last[0][0].replace(prefix, ""))
else:
last = 0
frappe.db.sql("""update tabSeries set current = %s
where name=%s""", (last, prefix))
def delete_bins(company_name):
frappe.db.sql("""delete from tabBin where warehouse in
(select name from tabWarehouse where company=%s)""", company_name)
def delete_lead_addresses(company_name):
"""Delete addresses to which leads are linked"""
leads = frappe.get_all("Lead", filters={"company": company_name})
leads = [ "'%s'"%row.get("name") for row in leads ]
addresses = []
if leads:
addresses = frappe.db.sql_list("""select parent from `tabDynamic Link` where link_name
in ({leads})""".format(leads=",".join(leads)))
if addresses:
addresses = ["%s" % frappe.db.escape(addr) for addr in addresses]
frappe.db.sql("""delete from tabAddress where name in ({addresses}) and
name not in (select distinct dl1.parent from `tabDynamic Link` dl1
inner join `tabDynamic Link` dl2 on dl1.parent=dl2.parent
and dl1.link_doctype<>dl2.link_doctype)""".format(addresses=",".join(addresses)))
frappe.db.sql("""delete from `tabDynamic Link` where link_doctype='Lead'
and parenttype='Address' and link_name in ({leads})""".format(leads=",".join(leads)))
frappe.db.sql("""update tabCustomer set lead_name=NULL where lead_name in ({leads})""".format(leads=",".join(leads)))
def delete_communications(doctype, company_name, company_fieldname):
reference_docs = frappe.get_all(doctype, filters={company_fieldname:company_name})
reference_doc_names = [r.name for r in reference_docs]
communications = frappe.get_all("Communication", filters={"reference_doctype":doctype,"reference_name":["in", reference_doc_names]})
communication_names = [c.name for c in communications]
frappe.delete_doc("Communication", communication_names, ignore_permissions=True)

View File

@ -86,15 +86,6 @@ class TestCompany(unittest.TestCase):
self.delete_mode_of_payment(template)
frappe.delete_doc("Company", template)
def test_delete_communication(self):
from erpnext.setup.doctype.company.delete_company_transactions import delete_communications
company = create_child_company()
lead = create_test_lead_in_company(company)
communication = create_company_communication("Lead", lead)
delete_communications("Lead", "Test Company", "company")
self.assertFalse(frappe.db.exists("Communcation", communication))
self.assertFalse(frappe.db.exists({"doctype":"Comunication Link", "link_name": communication}))
def delete_mode_of_payment(self, company):
frappe.db.sql(""" delete from `tabMode of Payment Account`
where company =%s """, (company))

View File

@ -0,0 +1,68 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
from __future__ import unicode_literals
import frappe
import unittest
class TestTransactionDeletionRecord(unittest.TestCase):
def setUp(self):
create_company('Dunder Mifflin Paper Co')
def tearDown(self):
frappe.db.rollback()
def test_doctypes_contain_company_field(self):
tdr = create_transaction_deletion_request('Dunder Mifflin Paper Co')
for doctype in tdr.doctypes:
contains_company = False
doctype_fields = frappe.get_meta(doctype.doctype_name).as_dict()['fields']
for doctype_field in doctype_fields:
if doctype_field['fieldtype'] == 'Link' and doctype_field['options'] == 'Company':
contains_company = True
break
self.assertTrue(contains_company)
def test_no_of_docs_is_correct(self):
for i in range(5):
create_task('Dunder Mifflin Paper Co')
tdr = create_transaction_deletion_request('Dunder Mifflin Paper Co')
for doctype in tdr.doctypes:
if doctype.doctype_name == 'Task':
self.assertEqual(doctype.no_of_docs, 5)
def test_deletion_is_successful(self):
create_task('Dunder Mifflin Paper Co')
create_transaction_deletion_request('Dunder Mifflin Paper Co')
tasks_containing_company = frappe.get_all('Task',
filters = {
'company' : 'Dunder Mifflin Paper Co'
})
self.assertEqual(tasks_containing_company, [])
def create_company(company_name):
company = frappe.get_doc({
'doctype': 'Company',
'company_name': company_name,
'default_currency': 'INR'
})
company.insert(ignore_if_duplicate = True)
def create_transaction_deletion_request(company):
tdr = frappe.get_doc({
'doctype': 'Transaction Deletion Record',
'company': company
})
tdr.insert()
tdr.submit()
return tdr
def create_task(company):
task = frappe.get_doc({
'doctype': 'Task',
'company': company,
'subject': 'Delete'
})
task.insert()

View File

@ -0,0 +1,40 @@
// Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and contributors
// For license information, please see license.txt
frappe.ui.form.on('Transaction Deletion Record', {
onload: function(frm) {
if (frm.doc.docstatus == 0) {
let doctypes_to_be_ignored_array;
frappe.call({
method: 'erpnext.setup.doctype.transaction_deletion_record.transaction_deletion_record.get_doctypes_to_be_ignored',
callback: function(r) {
doctypes_to_be_ignored_array = r.message;
populate_doctypes_to_be_ignored(doctypes_to_be_ignored_array, frm);
frm.fields_dict['doctypes_to_be_ignored'].grid.set_column_disp('no_of_docs', false);
frm.refresh_field('doctypes_to_be_ignored');
}
});
}
frm.get_field('doctypes_to_be_ignored').grid.cannot_add_rows = true;
frm.fields_dict['doctypes_to_be_ignored'].grid.set_column_disp('no_of_docs', false);
frm.refresh_field('doctypes_to_be_ignored');
},
refresh: function(frm) {
frm.fields_dict['doctypes_to_be_ignored'].grid.set_column_disp('no_of_docs', false);
frm.refresh_field('doctypes_to_be_ignored');
}
});
function populate_doctypes_to_be_ignored(doctypes_to_be_ignored_array, frm) {
if (!(frm.doc.doctypes_to_be_ignored)) {
var i;
for (i = 0; i < doctypes_to_be_ignored_array.length; i++) {
frm.add_child('doctypes_to_be_ignored', {
doctype_name: doctypes_to_be_ignored_array[i]
});
}
}
}

View File

@ -0,0 +1,79 @@
{
"actions": [],
"autoname": "TDL.####",
"creation": "2021-04-06 20:17:18.404716",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"company",
"doctypes",
"doctypes_to_be_ignored",
"amended_from",
"status"
],
"fields": [
{
"fieldname": "company",
"fieldtype": "Link",
"in_list_view": 1,
"label": "Company",
"options": "Company",
"reqd": 1
},
{
"fieldname": "doctypes",
"fieldtype": "Table",
"label": "Summary",
"options": "Transaction Deletion Record Item",
"read_only": 1
},
{
"fieldname": "doctypes_to_be_ignored",
"fieldtype": "Table",
"label": "Excluded DocTypes",
"options": "Transaction Deletion Record Item"
},
{
"fieldname": "amended_from",
"fieldtype": "Link",
"label": "Amended From",
"no_copy": 1,
"options": "Transaction Deletion Record",
"print_hide": 1,
"read_only": 1
},
{
"fieldname": "status",
"fieldtype": "Select",
"hidden": 1,
"label": "Status",
"options": "Draft\nCompleted"
}
],
"index_web_pages_for_search": 1,
"is_submittable": 1,
"links": [],
"modified": "2021-05-08 23:13:48.049879",
"modified_by": "Administrator",
"module": "Setup",
"name": "Transaction Deletion Record",
"owner": "Administrator",
"permissions": [
{
"create": 1,
"delete": 1,
"email": 1,
"export": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "System Manager",
"share": 1,
"write": 1
}
],
"sort_field": "modified",
"sort_order": "DESC",
"track_changes": 1
}

View File

@ -0,0 +1,147 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
from __future__ import unicode_literals
from frappe.utils import cint
import frappe
from frappe.model.document import Document
from frappe import _
from frappe.desk.notifications import clear_notifications
class TransactionDeletionRecord(Document):
def validate(self):
frappe.only_for('System Manager')
company_obj = frappe.get_doc('Company', self.company)
if frappe.session.user != company_obj.owner and frappe.session.user != 'Administrator':
frappe.throw(_('Transactions can only be deleted by the creator of the Company or the Administrator.'),
frappe.PermissionError)
doctypes_to_be_ignored_list = get_doctypes_to_be_ignored()
for doctype in self.doctypes_to_be_ignored:
if doctype.doctype_name not in doctypes_to_be_ignored_list:
frappe.throw(_("DocTypes should not be added manually to the 'Excluded DocTypes' table. You are only allowed to remove entries from it. "), title=_("Not Allowed"))
def before_submit(self):
if not self.doctypes_to_be_ignored:
self.populate_doctypes_to_be_ignored_table()
self.delete_bins()
self.delete_lead_addresses()
company_obj = frappe.get_doc('Company', self.company)
# reset company values
company_obj.total_monthly_sales = 0
company_obj.sales_monthly_history = None
company_obj.save()
# Clear notification counts
clear_notifications()
singles = frappe.get_all('DocType', filters = {'issingle': 1}, pluck = 'name')
tables = frappe.get_all('DocType', filters = {'istable': 1}, pluck = 'name')
doctypes_to_be_ignored_list = singles
for doctype in self.doctypes_to_be_ignored:
doctypes_to_be_ignored_list.append(doctype.doctype_name)
docfields = frappe.get_all('DocField',
filters = {
'fieldtype': 'Link',
'options': 'Company',
'parent': ['not in', doctypes_to_be_ignored_list]},
fields=['parent', 'fieldname'])
for docfield in docfields:
if docfield['parent'] != self.doctype:
no_of_docs = frappe.db.count(docfield['parent'], {
docfield['fieldname'] : self.company
})
if no_of_docs > 0:
self.delete_version_log(docfield['parent'], docfield['fieldname'])
self.delete_communications(docfield['parent'], docfield['fieldname'])
# populate DocTypes table
if docfield['parent'] not in tables:
self.append('doctypes', {
'doctype_name' : docfield['parent'],
'no_of_docs' : no_of_docs
})
# delete the docs linked with the specified company
frappe.db.delete(docfield['parent'], {
docfield['fieldname'] : self.company
})
naming_series = frappe.db.get_value('DocType', docfield['parent'], 'autoname')
if naming_series:
if '#' in naming_series:
self.update_naming_series(naming_series, docfield['parent'])
def populate_doctypes_to_be_ignored_table(self):
doctypes_to_be_ignored_list = get_doctypes_to_be_ignored()
for doctype in doctypes_to_be_ignored_list:
self.append('doctypes_to_be_ignored', {
'doctype_name' : doctype
})
def update_naming_series(self, naming_series, doctype_name):
if '.' in naming_series:
prefix, hashes = naming_series.rsplit('.', 1)
else:
prefix, hashes = naming_series.rsplit('{', 1)
last = frappe.db.sql("""select max(name) from `tab{0}`
where name like %s""".format(doctype_name), prefix + '%')
if last and last[0][0]:
last = cint(last[0][0].replace(prefix, ''))
else:
last = 0
frappe.db.sql("""update tabSeries set current = %s where name=%s""", (last, prefix))
def delete_version_log(self, doctype, company_fieldname):
frappe.db.sql("""delete from `tabVersion` where ref_doctype=%s and docname in
(select name from `tab{0}` where `{1}`=%s)""".format(doctype,
company_fieldname), (doctype, self.company))
def delete_communications(self, doctype, company_fieldname):
reference_docs = frappe.get_all(doctype, filters={company_fieldname:self.company})
reference_doc_names = [r.name for r in reference_docs]
communications = frappe.get_all('Communication', filters={'reference_doctype':doctype,'reference_name':['in', reference_doc_names]})
communication_names = [c.name for c in communications]
frappe.delete_doc('Communication', communication_names, ignore_permissions=True)
def delete_bins(self):
frappe.db.sql("""delete from tabBin where warehouse in
(select name from tabWarehouse where company=%s)""", self.company)
def delete_lead_addresses(self):
"""Delete addresses to which leads are linked"""
leads = frappe.get_all('Lead', filters={'company': self.company})
leads = ["'%s'" % row.get("name") for row in leads]
addresses = []
if leads:
addresses = frappe.db.sql_list("""select parent from `tabDynamic Link` where link_name
in ({leads})""".format(leads=",".join(leads)))
if addresses:
addresses = ["%s" % frappe.db.escape(addr) for addr in addresses]
frappe.db.sql("""delete from tabAddress where name in ({addresses}) and
name not in (select distinct dl1.parent from `tabDynamic Link` dl1
inner join `tabDynamic Link` dl2 on dl1.parent=dl2.parent
and dl1.link_doctype<>dl2.link_doctype)""".format(addresses=",".join(addresses)))
frappe.db.sql("""delete from `tabDynamic Link` where link_doctype='Lead'
and parenttype='Address' and link_name in ({leads})""".format(leads=",".join(leads)))
frappe.db.sql("""update tabCustomer set lead_name=NULL where lead_name in ({leads})""".format(leads=",".join(leads)))
@frappe.whitelist()
def get_doctypes_to_be_ignored():
doctypes_to_be_ignored_list = ['Account', 'Cost Center', 'Warehouse', 'Budget',
'Party Account', 'Employee', 'Sales Taxes and Charges Template',
'Purchase Taxes and Charges Template', 'POS Profile', 'BOM',
'Company', 'Bank Account', 'Item Tax Template', 'Mode of Payment',
'Item Default', 'Customer', 'Supplier', 'GST Account']
return doctypes_to_be_ignored_list

View File

@ -0,0 +1,12 @@
// Copyright (c) 2018, Frappe Technologies Pvt. Ltd. and Contributors
// License: GNU General Public License v3. See license.txt
frappe.listview_settings['Transaction Deletion Record'] = {
get_indicator: function(doc) {
if (doc.docstatus == 0) {
return [__("Draft"), "red"];
} else {
return [__("Completed"), "green"];
}
}
};

View File

@ -0,0 +1,39 @@
{
"actions": [],
"creation": "2021-04-07 07:34:00.124124",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"doctype_name",
"no_of_docs"
],
"fields": [
{
"fieldname": "doctype_name",
"fieldtype": "Link",
"in_list_view": 1,
"label": "DocType",
"options": "DocType",
"reqd": 1
},
{
"fieldname": "no_of_docs",
"fieldtype": "Data",
"in_list_view": 1,
"label": "Number of Docs"
}
],
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2021-05-08 23:10:46.166744",
"modified_by": "Administrator",
"module": "Setup",
"name": "Transaction Deletion Record Item",
"owner": "Administrator",
"permissions": [],
"sort_field": "modified",
"sort_order": "DESC",
"track_changes": 1
}

View File

@ -0,0 +1,10 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
from __future__ import unicode_literals
# import frappe
from frappe.model.document import Document
class TransactionDeletionRecordItem(Document):
pass