feat: Scheduling Multiple shifts and Auto Attendance

.git-blame-ignore-revs

@@ -26,3 +26,6 @@ b147b85e6ac19a9220cd1e2958a6ebd99373283a
 
 # bulk format python code with black
 494bd9ef78313436f0424b918f200dab8fc7c20b
+
+# bulk format python code with black
+baec607ff5905b1c67531096a9cf50ec7ff00a5d

erpnext/hr/doctype/attendance/attendance.py

@@ -5,11 +5,21 @@
 import frappe
 from frappe import _
 from frappe.model.document import Document
-from frappe.utils import cint, cstr, formatdate, get_datetime, getdate, nowdate
+from frappe.query_builder import Criterion
+from frappe.utils import cint, cstr, formatdate, get_datetime, get_link_to_form, getdate, nowdate
 
+from erpnext.hr.doctype.shift_assignment.shift_assignment import has_overlapping_timings
 from erpnext.hr.utils import get_holiday_dates_for_employee, validate_active_employee
 
 
+class DuplicateAttendanceError(frappe.ValidationError):
+	pass
+
+
+class OverlappingShiftAttendanceError(frappe.ValidationError):
+	pass
+
+
 class Attendance(Document):
 	def validate(self):
 		from erpnext.controllers.status_updater import validate_status
@@ -18,6 +28,7 @@ def validate(self):
 		validate_active_employee(self.employee)
 		self.validate_attendance_date()
 		self.validate_duplicate_record()
+		self.validate_overlapping_shift_attendance()
 		self.validate_employee_status()
 		self.check_leave_record()
 
@@ -35,21 +46,35 @@ def validate_attendance_date(self):
 			frappe.throw(_("Attendance date can not be less than employee's joining date"))
 
 	def validate_duplicate_record(self):
-		res = frappe.db.sql(
-			"""
-			select name from `tabAttendance`
-			where employee = %s
-				and attendance_date = %s
-				and name != %s
-				and docstatus != 2
-		""",
-			(self.employee, getdate(self.attendance_date), self.name),
+		duplicate = get_duplicate_attendance_record(
+			self.employee, self.attendance_date, self.shift, self.name
 		)
-		if res:
+
+		if duplicate:
 			frappe.throw(
-				_("Attendance for employee {0} is already marked for the date {1}").format(
-					frappe.bold(self.employee), frappe.bold(self.attendance_date)
-				)
+				_("Attendance for employee {0} is already marked for the date {1}: {2}").format(
+					frappe.bold(self.employee),
+					frappe.bold(self.attendance_date),
+					get_link_to_form("Attendance", duplicate[0].name),
+				),
+				title=_("Duplicate Attendance"),
+				exc=DuplicateAttendanceError,
+			)
+
+	def validate_overlapping_shift_attendance(self):
+		attendance = get_overlapping_shift_attendance(
+			self.employee, self.attendance_date, self.shift, self.name
+		)
+
+		if attendance:
+			frappe.throw(
+				_("Attendance for employee {0} is already marked for an overlapping shift {1}: {2}").format(
+					frappe.bold(self.employee),
+					frappe.bold(attendance.shift),
+					get_link_to_form("Attendance", attendance.name),
+				),
+				title=_("Overlapping Shift Attendance"),
+				exc=OverlappingShiftAttendanceError,
 			)
 
 	def validate_employee_status(self):
@@ -103,6 +128,69 @@ def validate_employee(self):
 			frappe.throw(_("Employee {0} is not active or does not exist").format(self.employee))
 
 
+def get_duplicate_attendance_record(employee, attendance_date, shift, name=None):
+	attendance = frappe.qb.DocType("Attendance")
+	query = (
+		frappe.qb.from_(attendance)
+		.select(attendance.name)
+		.where((attendance.employee == employee) & (attendance.docstatus < 2))
+	)
+
+	if shift:
+		query = query.where(
+			Criterion.any(
+				[
+					Criterion.all(
+						[
+							((attendance.shift.isnull()) | (attendance.shift == "")),
+							(attendance.attendance_date == attendance_date),
+						]
+					),
+					Criterion.all(
+						[
+							((attendance.shift.isnotnull()) | (attendance.shift != "")),
+							(attendance.attendance_date == attendance_date),
+							(attendance.shift == shift),
+						]
+					),
+				]
+			)
+		)
+	else:
+		query = query.where((attendance.attendance_date == attendance_date))
+
+	if name:
+		query = query.where(attendance.name != name)
+
+	return query.run(as_dict=True)
+
+
+def get_overlapping_shift_attendance(employee, attendance_date, shift, name=None):
+	if not shift:
+		return {}
+
+	attendance = frappe.qb.DocType("Attendance")
+	query = (
+		frappe.qb.from_(attendance)
+		.select(attendance.name, attendance.shift)
+		.where(
+			(attendance.employee == employee)
+			& (attendance.docstatus < 2)
+			& (attendance.attendance_date == attendance_date)
+			& (attendance.shift != shift)
+		)
+	)
+
+	if name:
+		query = query.where(attendance.name != name)
+
+	overlapping_attendance = query.run(as_dict=True)
+
+	if overlapping_attendance and has_overlapping_timings(shift, overlapping_attendance[0].shift):
+		return overlapping_attendance[0]
+	return {}
+
+
 @frappe.whitelist()
 def get_events(start, end, filters=None):
 	events = []
@@ -141,28 +229,39 @@ def add_attendance(events, start, end, conditions=None):
 
 
 def mark_attendance(
-	employee, attendance_date, status, shift=None, leave_type=None, ignore_validate=False
+	employee,
+	attendance_date,
+	status,
+	shift=None,
+	leave_type=None,
+	ignore_validate=False,
+	late_entry=False,
+	early_exit=False,
 ):
-	if not frappe.db.exists(
-		"Attendance",
-		{"employee": employee, "attendance_date": attendance_date, "docstatus": ("!=", "2")},
-	):
-		company = frappe.db.get_value("Employee", employee, "company")
-		attendance = frappe.get_doc(
-			{
-				"doctype": "Attendance",
-				"employee": employee,
-				"attendance_date": attendance_date,
-				"status": status,
-				"company": company,
-				"shift": shift,
-				"leave_type": leave_type,
-			}
-		)
-		attendance.flags.ignore_validate = ignore_validate
-		attendance.insert()
-		attendance.submit()
-		return attendance.name
+	if get_duplicate_attendance_record(employee, attendance_date, shift):
+		return
+
+	if get_overlapping_shift_attendance(employee, attendance_date, shift):
+		return
+
+	company = frappe.db.get_value("Employee", employee, "company")
+	attendance = frappe.get_doc(
+		{
+			"doctype": "Attendance",
+			"employee": employee,
+			"attendance_date": attendance_date,
+			"status": status,
+			"company": company,
+			"shift": shift,
+			"leave_type": leave_type,
+			"late_entry": late_entry,
+			"early_exit": early_exit,
+		}
+	)
+	attendance.flags.ignore_validate = ignore_validate
+	attendance.insert()
+	attendance.submit()
+	return attendance.name
 
 
 @frappe.whitelist()

erpnext/hr/doctype/attendance/test_attendance.py

@@ -6,6 +6,8 @@
 from frappe.utils import add_days, get_year_ending, get_year_start, getdate, now_datetime, nowdate
 
 from erpnext.hr.doctype.attendance.attendance import (
+	DuplicateAttendanceError,
+	OverlappingShiftAttendanceError,
 	get_month_map,
 	get_unmarked_days,
 	mark_attendance,
@@ -23,11 +25,112 @@ def setUp(self):
 		from_date = get_year_start(getdate())
 		to_date = get_year_ending(getdate())
 		self.holiday_list = make_holiday_list(from_date=from_date, to_date=to_date)
+		frappe.db.delete("Attendance")
+
+	def test_duplicate_attendance(self):
+		employee = make_employee("test_duplicate_attendance@example.com", company="_Test Company")
+		date = nowdate()
+
+		mark_attendance(employee, date, "Present")
+		attendance = frappe.get_doc(
+			{
+				"doctype": "Attendance",
+				"employee": employee,
+				"attendance_date": date,
+				"status": "Absent",
+				"company": "_Test Company",
+			}
+		)
+
+		self.assertRaises(DuplicateAttendanceError, attendance.insert)
+
+	def test_duplicate_attendance_with_shift(self):
+		from erpnext.hr.doctype.shift_type.test_shift_type import setup_shift_type
+
+		employee = make_employee("test_duplicate_attendance@example.com", company="_Test Company")
+		date = nowdate()
+
+		shift_1 = setup_shift_type(shift_type="Shift 1", start_time="08:00:00", end_time="10:00:00")
+		mark_attendance(employee, date, "Present", shift=shift_1.name)
+
+		# attendance record with shift
+		attendance = frappe.get_doc(
+			{
+				"doctype": "Attendance",
+				"employee": employee,
+				"attendance_date": date,
+				"status": "Absent",
+				"company": "_Test Company",
+				"shift": shift_1.name,
+			}
+		)
+
+		self.assertRaises(DuplicateAttendanceError, attendance.insert)
+
+		# attendance record without any shift
+		attendance = frappe.get_doc(
+			{
+				"doctype": "Attendance",
+				"employee": employee,
+				"attendance_date": date,
+				"status": "Absent",
+				"company": "_Test Company",
+			}
+		)
+
+		self.assertRaises(DuplicateAttendanceError, attendance.insert)
+
+	def test_overlapping_shift_attendance_validation(self):
+		from erpnext.hr.doctype.shift_type.test_shift_type import setup_shift_type
+
+		employee = make_employee("test_overlap_attendance@example.com", company="_Test Company")
+		date = nowdate()
+
+		shift_1 = setup_shift_type(shift_type="Shift 1", start_time="08:00:00", end_time="10:00:00")
+		shift_2 = setup_shift_type(shift_type="Shift 2", start_time="09:30:00", end_time="11:00:00")
+
+		mark_attendance(employee, date, "Present", shift=shift_1.name)
+
+		# attendance record with overlapping shift
+		attendance = frappe.get_doc(
+			{
+				"doctype": "Attendance",
+				"employee": employee,
+				"attendance_date": date,
+				"status": "Absent",
+				"company": "_Test Company",
+				"shift": shift_2.name,
+			}
+		)
+
+		self.assertRaises(OverlappingShiftAttendanceError, attendance.insert)
+
+	def test_allow_attendance_with_different_shifts(self):
+		# allows attendance with 2 different non-overlapping shifts
+		from erpnext.hr.doctype.shift_type.test_shift_type import setup_shift_type
+
+		employee = make_employee("test_duplicate_attendance@example.com", company="_Test Company")
+		date = nowdate()
+
+		shift_1 = setup_shift_type(shift_type="Shift 1", start_time="08:00:00", end_time="10:00:00")
+		shift_2 = setup_shift_type(shift_type="Shift 2", start_time="11:00:00", end_time="12:00:00")
+
+		mark_attendance(employee, date, "Present", shift_1.name)
+		frappe.get_doc(
+			{
+				"doctype": "Attendance",
+				"employee": employee,
+				"attendance_date": date,
+				"status": "Absent",
+				"company": "_Test Company",
+				"shift": shift_2.name,
+			}
+		).insert()
 
 	def test_mark_absent(self):
 		employee = make_employee("test_mark_absent@example.com")
 		date = nowdate()
-		frappe.db.delete("Attendance", {"employee": employee, "attendance_date": date})
+
 		attendance = mark_attendance(employee, date, "Absent")
 		fetch_attendance = frappe.get_value(
 			"Attendance", {"employee": employee, "attendance_date": date, "status": "Absent"}
@@ -42,7 +145,6 @@ def test_unmarked_days(self):
 		employee = make_employee(
 			"test_unmarked_days@example.com", date_of_joining=add_days(first_day, -1)
 		)
-		frappe.db.delete("Attendance", {"employee": employee})
 		frappe.db.set_value("Employee", employee, "holiday_list", self.holiday_list)
 
 		first_sunday = get_first_sunday(self.holiday_list, for_date=first_day)
@@ -67,8 +169,6 @@ def test_unmarked_days_excluding_holidays(self):
 		employee = make_employee(
 			"test_unmarked_days@example.com", date_of_joining=add_days(first_day, -1)
 		)
-		frappe.db.delete("Attendance", {"employee": employee})
-
 		frappe.db.set_value("Employee", employee, "holiday_list", self.holiday_list)
 
 		first_sunday = get_first_sunday(self.holiday_list, for_date=first_day)
@@ -95,7 +195,6 @@ def test_unmarked_days_as_per_joining_and_relieving_dates(self):
 		employee = make_employee(
 			"test_unmarked_days_as_per_doj@example.com", date_of_joining=doj, relieving_date=relieving_date
 		)
-		frappe.db.delete("Attendance", {"employee": employee})
 
 		frappe.db.set_value("Employee", employee, "holiday_list", self.holiday_list)