Skip to content

TASK-2025-02771: Convert Task override to extended class inheritance #595

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anjusha12 wants to merge 1 commit into
base: version-15
Choose a base branch
from
Open

TASK-2025-02771: Convert Task override to extended class inheritance #595

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
275 changes: 7 additions & 268 deletions one_compliance/one_compliance/doc_events/task.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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.")