refactor: Convert Task override to extended class inheritance

Author: anjusha12

one_compliance/one_compliance/doc_events/task.py

@@ -4,129 +4,22 @@
 )
 from erpnext.accounts.party import get_party_account
 from frappe import _, throw
-from frappe.desk.form.assign_to import clear, close_all_assignments
+from frappe.desk.form.assign_to import close_all_assignments
 from frappe.email.doctype.notification.notification import get_context
-from frappe.utils import add_days, cstr, date_diff, flt, getdate, today
-from frappe.utils.data import format_date
-from frappe.utils.nestedset import NestedSet
-from erpnext.projects.doctype.task.task import check_if_child_exists, CircularReferenceError
+from frappe.utils import add_days, getdate, today
+from erpnext.projects.doctype.task.task import Task
 
 from one_compliance.one_compliance.utils import (
 	create_project_completion_todos,
 	send_notification,
 	send_notification_to_roles,
 )
 
-
-class CustomTask(NestedSet):
-	# 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 erpnext.projects.doctype.task_depends_on.task_depends_on import (
-			TaskDependsOn,
-		)
-		from frappe.types import DF
-
-		act_end_date: DF.Date | None
-		act_start_date: DF.Date | None
-		actual_time: DF.Float
-		closing_date: DF.Date | None
-		color: DF.Color | None
-		company: DF.Link | None
-		completed_by: DF.Link | None
-		completed_on: DF.Date | None
-		department: DF.Link | None
-		depends_on: DF.Table[TaskDependsOn]
-		depends_on_tasks: DF.Code | None
-		description: DF.TextEditor | None
-		duration: DF.Int
-		exp_end_date: DF.Date | None
-		exp_start_date: DF.Date | None
-		expected_time: DF.Float
-		is_group: DF.Check
-		is_milestone: DF.Check
-		is_template: DF.Check
-		issue: DF.Link | None
-		lft: DF.Int
-		old_parent: DF.Data | None
-		parent_task: DF.Link | None
-		priority: DF.Literal["Low", "Medium", "High", "Urgent"]
-		progress: DF.Percent
-		project: DF.Link | None
-		review_date: DF.Date | None
-		rgt: DF.Int
-		start: DF.Int
-		status: DF.Literal[
-			"Open", "Working", "Pending Review", "Overdue", "Template", "Completed", "Cancelled"
-		]
-		subject: DF.Data
-		task_weight: DF.Float
-		template_task: DF.Data | None
-		total_billing_amount: DF.Currency
-		total_costing_amount: DF.Currency
-		type: DF.Link | None
-	# end: auto-generated types
-
-	nsm_parent_field = "parent_task"
-
-	def get_customer_details(self):
-		cust = frappe.db.sql("select customer_name from `tabCustomer` where name=%s", self.customer)
-		if cust:
-			ret = {"customer_name": cust and cust[0][0] or ""}
-			return ret
-
+class CustomTask(Task):
 	def validate(self):
-		self.validate_dates()
-		self.validate_progress()
-		self.validate_status()
-		self.update_depends_on()
-		self.validate_dependencies_for_template_task()
-		self.validate_completed_on()
+		super(CustomTask, self).validate()
 		self.validate_reimbursement_check()
-
-	def validate_dates(self):
-		self.validate_from_to_dates("exp_start_date", "exp_end_date")
-		self.validate_from_to_dates("act_start_date", "act_end_date")
-		self.validate_parent_expected_end_date()
-		self.validate_parent_project_dates()
-
-	def validate_parent_expected_end_date(self):
-		if not self.parent_task or not self.exp_end_date:
-			return
-
-		parent_exp_end_date = frappe.db.get_value("Task", self.parent_task, "exp_end_date")
-		if not parent_exp_end_date:
-			return
-
-		if getdate(self.exp_end_date) > getdate(parent_exp_end_date):
-			frappe.throw(
-				_(
-					"Expected End Date should be less than or equal to parent task's Expected End Date {0}."
-				).format(format_date(parent_exp_end_date)),
-				frappe.exceptions.InvalidDates,
-			)
-
-	def validate_parent_project_dates(self):
-		if not self.project or frappe.flags.in_test:
-			return
-
-		if project_end_date := frappe.db.get_value("Project", self.project, "expected_end_date"):
-			project_end_date = getdate(project_end_date)
-			for fieldname in ("exp_start_date", "exp_end_date", "act_start_date", "act_end_date"):
-				task_date = self.get(fieldname)
-				if task_date and date_diff(project_end_date, getdate(task_date)) < 0:
-					frappe.throw(
-						_("{0}'s {1} cannot be after {2}'s Expected End Date.").format(
-							frappe.bold(frappe.get_desk_link("Task", self.name)),
-							_(self.meta.get_label(fieldname)),
-							frappe.bold(frappe.get_desk_link("Project", self.project)),
-						),
-						frappe.exceptions.InvalidDates,
-					)
-
+	
 	def validate_status(self):
 		if self.is_template and self.status != "Template":
 			self.status = "Template"
@@ -145,35 +38,6 @@ def validate_status(self):
 
 			close_all_assignments(self.doctype, self.name)
 
-	def validate_progress(self):
-		if flt(self.progress or 0) > 100:
-			frappe.throw(_("Progress % for a task cannot be more than 100."))
-
-		if self.status == "Completed":
-			self.progress = 100
-
-	def validate_dependencies_for_template_task(self):
-		if self.is_template:
-			self.validate_parent_template_task()
-			self.validate_depends_on_tasks()
-
-	def validate_parent_template_task(self):
-		if self.parent_task:
-			if not frappe.db.get_value("Task", self.parent_task, "is_template"):
-				parent_task_format = f"""<a href="#Form/Task/{self.parent_task}">{self.parent_task}</a>"""
-				frappe.throw(_("Parent Task {0} is not a Template Task").format(parent_task_format))
-
-	def validate_depends_on_tasks(self):
-		if self.depends_on:
-			for task in self.depends_on:
-				if not frappe.db.get_value("Task", task.task, "is_template"):
-					dependent_task_format = f"""<a href="#Form/Task/{task.task}">{task.task}</a>"""
-					frappe.throw(_("Dependent Task {0} is not a Template Task").format(dependent_task_format))
-
-	def validate_completed_on(self):
-		if self.completed_on and getdate(self.completed_on) > getdate():
-			frappe.throw(_("Completed On cannot be greater than Today"))
-
 	def validate_reimbursement_check(self):
 		'''
 			Validate Rembursement JV on Task Completion
@@ -183,132 +47,6 @@ def validate_reimbursement_check(self):
 				title=_("Reimbursement Journal Entry Missing"),
 				msg=_("Please create Reimbursement Journal Entry before marking the task <b>`{0}`</b> as Completed".format(self.name)))
 
-	def update_depends_on(self):
-		depends_on_tasks = ""
-		for d in self.depends_on:
-			if d.task and d.task not in depends_on_tasks:
-				depends_on_tasks += d.task + ","
-		self.depends_on_tasks = depends_on_tasks
-
-	def update_nsm_model(self):
-		frappe.utils.nestedset.update_nsm(self)
-
-	def on_update(self):
-		self.update_nsm_model()
-		self.check_recursion()
-		self.reschedule_dependent_tasks()
-		self.update_project()
-		self.unassign_todo()
-		self.populate_depends_on()
-
-	def unassign_todo(self):
-		if self.status == "Completed":
-			close_all_assignments(self.doctype, self.name)
-		if self.status == "Cancelled":
-			clear(self.doctype, self.name)
-
-	def update_time_and_costing(self):
-		tl = frappe.db.sql(
-			"""select min(from_time) as start_date, max(to_time) as end_date,
-			sum(billing_amount) as total_billing_amount, sum(costing_amount) as total_costing_amount,
-			sum(hours) as time from `tabTimesheet Detail` where task = %s and docstatus=1""",
-			self.name,
-			as_dict=1,
-		)[0]
-		if self.status == "Open":
-			self.status = "Working"
-		self.total_costing_amount = tl.total_costing_amount
-		self.total_billing_amount = tl.total_billing_amount
-		self.actual_time = tl.time
-		self.act_start_date = tl.start_date
-		self.act_end_date = tl.end_date
-
-	def update_project(self):
-		if self.project and not self.flags.from_project:
-			frappe.get_cached_doc("Project", self.project).update_project()
-
-	def check_recursion(self):
-		if self.flags.ignore_recursion_check:
-			return
-		check_list = [["task", "parent"], ["parent", "task"]]
-		for d in check_list:
-			task_list, count = [self.name], 0
-			while len(task_list) > count:
-				tasks = frappe.db.sql(
-					" select {} from `tabTask Depends On` where {} = {} ".format(d[0], d[1], "%s"),
-					cstr(task_list[count]),
-				)
-				count = count + 1
-				for b in tasks:
-					if b[0] == self.name:
-						frappe.throw(_("Circular Reference Error"), CircularReferenceError)
-					if b[0]:
-						task_list.append(b[0])
-
-				if count == 15:
-					break
-
-	def reschedule_dependent_tasks(self):
-		end_date = self.exp_end_date or self.act_end_date
-		if end_date:
-			for task_name in frappe.db.sql(
-				"""
-				select name from `tabTask` as parent
-				where parent.project = %(project)s
-					and parent.name in (
-						select parent from `tabTask Depends On` as child
-						where child.task = %(task)s and child.project = %(project)s)
-			""",
-				{"project": self.project, "task": self.name},
-				as_dict=1,
-			):
-				task = frappe.get_doc("Task", task_name.name)
-				if (
-					task.exp_start_date
-					and task.exp_end_date
-					and task.exp_start_date < getdate(end_date)
-					and task.status == "Open"
-				):
-					task_duration = date_diff(task.exp_end_date, task.exp_start_date)
-					task.exp_start_date = add_days(end_date, 1)
-					task.exp_end_date = add_days(task.exp_start_date, task_duration)
-					task.flags.ignore_recursion_check = True
-					task.save()
-
-	def has_webform_permission(self):
-		project_user = frappe.db.get_value(
-			"Project User", {"parent": self.project, "user": frappe.session.user}, "user"
-		)
-		if project_user:
-			return True
-
-	def populate_depends_on(self):
-		if self.parent_task:
-			parent = frappe.get_doc("Task", self.parent_task)
-			if self.name not in [row.task for row in parent.depends_on]:
-				parent.append(
-					"depends_on", {"doctype": "Task Depends On", "task": self.name, "subject": self.subject}
-				)
-				parent.save()
-
-	def on_trash(self):
-		if check_if_child_exists(self.name):
-			throw(_("Child Task exists for this Task. You can not delete this Task."))
-
-		self.update_nsm_model()
-
-	def after_delete(self):
-		self.update_project()
-
-	def update_status(self):
-		if self.status not in ("Cancelled", "Completed") and self.exp_end_date:
-			from datetime import datetime
-
-			if self.exp_end_date < datetime.now().date():
-				self.db_set("status", "Overdue", update_modified=False)
-				self.update_project()
-
-
 @frappe.whitelist()
 def append_users_to_project(doc, method):
 	if doc.assigned_to and doc.project:
@@ -704,3 +442,4 @@ def enable_customer_on_task_completion(doc, method):
 		customer.aml_compliance_checked = 1
 		customer.save(ignore_permissions=True)
 		frappe.msgprint(f"Customer {customer.name} has been enabled after AML compliance task completion.")
+