Merge branch 'develop' into multiple-shifts
diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs
index e9cb6cf..3bc22af 100644
--- a/.git-blame-ignore-revs
+++ b/.git-blame-ignore-revs
@@ -26,3 +26,6 @@
# bulk format python code with black
494bd9ef78313436f0424b918f200dab8fc7c20b
+
+# bulk format python code with black
+baec607ff5905b1c67531096a9cf50ec7ff00a5d
\ No newline at end of file
diff --git a/erpnext/hr/doctype/attendance/attendance.py b/erpnext/hr/doctype/attendance/attendance.py
index 7f4bd83..a2487b3 100644
--- a/erpnext/hr/doctype/attendance/attendance.py
+++ b/erpnext/hr/doctype/attendance/attendance.py
@@ -5,11 +5,16 @@
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.utils import get_holiday_dates_for_employee, validate_active_employee
+class DuplicateAttendanceError(frappe.ValidationError):
+ pass
+
+
class Attendance(Document):
def validate(self):
from erpnext.controllers.status_updater import validate_status
@@ -35,21 +40,19 @@
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_employee_status(self):
@@ -103,6 +106,43 @@
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)
+
+
@frappe.whitelist()
def get_events(start, end, filters=None):
events = []
@@ -141,12 +181,16 @@
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")},
- ):
+ if not get_duplicate_attendance_record(employee, attendance_date, shift):
company = frappe.db.get_value("Employee", employee, "company")
attendance = frappe.get_doc(
{
@@ -157,6 +201,8 @@
"company": company,
"shift": shift,
"leave_type": leave_type,
+ "late_entry": late_entry,
+ "early_exit": early_exit,
}
)
attendance.flags.ignore_validate = ignore_validate
diff --git a/erpnext/hr/doctype/employee_checkin/employee_checkin.py b/erpnext/hr/doctype/employee_checkin/employee_checkin.py
index 87f48b7..81c9a46 100644
--- a/erpnext/hr/doctype/employee_checkin/employee_checkin.py
+++ b/erpnext/hr/doctype/employee_checkin/employee_checkin.py
@@ -7,6 +7,7 @@
from frappe.model.document import Document
from frappe.utils import cint, get_datetime
+from erpnext.hr.doctype.attendance.attendance import get_duplicate_attendance_record
from erpnext.hr.doctype.shift_assignment.shift_assignment import (
get_actual_start_end_datetime_of_shift,
)
@@ -33,24 +34,24 @@
shift_actual_timings = get_actual_start_end_datetime_of_shift(
self.employee, get_datetime(self.time), True
)
- if shift_actual_timings[0] and shift_actual_timings[1]:
+ if shift_actual_timings:
if (
- shift_actual_timings[2].shift_type.determine_check_in_and_check_out
+ shift_actual_timings.shift_type.determine_check_in_and_check_out
== "Strictly based on Log Type in Employee Checkin"
and not self.log_type
and not self.skip_auto_attendance
):
frappe.throw(
_("Log Type is required for check-ins falling in the shift: {0}.").format(
- shift_actual_timings[2].shift_type.name
+ shift_actual_timings.shift_type.name
)
)
if not self.attendance:
- self.shift = shift_actual_timings[2].shift_type.name
- self.shift_actual_start = shift_actual_timings[0]
- self.shift_actual_end = shift_actual_timings[1]
- self.shift_start = shift_actual_timings[2].start_datetime
- self.shift_end = shift_actual_timings[2].end_datetime
+ self.shift = shift_actual_timings.shift_type.name
+ self.shift_actual_start = shift_actual_timings.actual_start
+ self.shift_actual_end = shift_actual_timings.actual_end
+ self.shift_start = shift_actual_timings.start_datetime
+ self.shift_end = shift_actual_timings.end_datetime
else:
self.shift = None
@@ -136,10 +137,7 @@
return None
elif attendance_status in ("Present", "Absent", "Half Day"):
employee_doc = frappe.get_doc("Employee", employee)
- if not frappe.db.exists(
- "Attendance",
- {"employee": employee, "attendance_date": attendance_date, "docstatus": ("!=", "2")},
- ):
+ if not get_duplicate_attendance_record(employee, attendance_date, shift):
doc_dict = {
"doctype": "Attendance",
"employee": employee,
diff --git a/erpnext/hr/doctype/shift_assignment/shift_assignment.py b/erpnext/hr/doctype/shift_assignment/shift_assignment.py
index 5a12486..fd0b4d5 100644
--- a/erpnext/hr/doctype/shift_assignment/shift_assignment.py
+++ b/erpnext/hr/doctype/shift_assignment/shift_assignment.py
@@ -3,83 +3,131 @@
from datetime import datetime, timedelta
+from typing import Dict, List
import frappe
from frappe import _
from frappe.model.document import Document
-from frappe.utils import cstr, getdate, now_datetime, nowdate
+from frappe.query_builder import Criterion
+from frappe.utils import cstr, get_datetime, get_link_to_form, get_time, getdate, now_datetime
from erpnext.hr.doctype.employee.employee import get_holiday_list_for_employee
from erpnext.hr.doctype.holiday_list.holiday_list import is_holiday
from erpnext.hr.utils import validate_active_employee
+class OverlappingShiftError(frappe.ValidationError):
+ pass
+
+
class ShiftAssignment(Document):
def validate(self):
validate_active_employee(self.employee)
- self.validate_overlapping_dates()
+ self.validate_overlapping_shifts()
if self.end_date:
self.validate_from_to_dates("start_date", "end_date")
- def validate_overlapping_dates(self):
+ def validate_overlapping_shifts(self):
+ overlapping_dates = self.get_overlapping_dates()
+ if len(overlapping_dates):
+ # if dates are overlapping, check if timings are overlapping, else allow
+ overlapping_timings = self.has_overlapping_timings(overlapping_dates[0].shift_type)
+ if overlapping_timings:
+ self.throw_overlap_error(overlapping_dates[0])
+
+ def get_overlapping_dates(self):
if not self.name:
self.name = "New Shift Assignment"
- condition = """and (
- end_date is null
- or
- %(start_date)s between start_date and end_date
- """
-
- if self.end_date:
- condition += """ or
- %(end_date)s between start_date and end_date
- or
- start_date between %(start_date)s and %(end_date)s
- ) """
- else:
- condition += """ ) """
-
- assigned_shifts = frappe.db.sql(
- """
- select name, shift_type, start_date ,end_date, docstatus, status
- from `tabShift Assignment`
- where
- employee=%(employee)s and docstatus = 1
- and name != %(name)s
- and status = "Active"
- {0}
- """.format(
- condition
- ),
- {
- "employee": self.employee,
- "shift_type": self.shift_type,
- "start_date": self.start_date,
- "end_date": self.end_date,
- "name": self.name,
- },
- as_dict=1,
+ shift = frappe.qb.DocType("Shift Assignment")
+ query = (
+ frappe.qb.from_(shift)
+ .select(
+ shift.name, shift.shift_type, shift.start_date, shift.end_date, shift.docstatus, shift.status
+ )
+ .where(
+ (shift.employee == self.employee)
+ & (shift.docstatus == 1)
+ & (shift.name != self.name)
+ & (shift.status == "Active")
+ )
)
- if len(assigned_shifts):
- self.throw_overlap_error(assigned_shifts[0])
+ if self.end_date:
+ query = query.where(
+ Criterion.any(
+ [
+ Criterion.any(
+ [
+ shift.end_date.isnull(),
+ ((self.start_date >= shift.start_date) & (self.start_date <= shift.end_date)),
+ ]
+ ),
+ Criterion.any(
+ [
+ ((self.end_date >= shift.start_date) & (self.end_date <= shift.end_date)),
+ shift.start_date.between(self.start_date, self.end_date),
+ ]
+ ),
+ ]
+ )
+ )
+ else:
+ query = query.where(
+ shift.end_date.isnull()
+ | ((self.start_date >= shift.start_date) & (self.start_date <= shift.end_date))
+ )
+
+ return query.run(as_dict=True)
+
+ def has_overlapping_timings(self, overlapping_shift):
+ curr_shift = frappe.db.get_value(
+ "Shift Type", self.shift_type, ["start_time", "end_time"], as_dict=True
+ )
+ overlapping_shift = frappe.db.get_value(
+ "Shift Type", overlapping_shift, ["start_time", "end_time"], as_dict=True
+ )
+
+ if (
+ (
+ curr_shift.start_time > overlapping_shift.start_time
+ and curr_shift.start_time < overlapping_shift.end_time
+ )
+ or (
+ curr_shift.end_time > overlapping_shift.start_time
+ and curr_shift.end_time < overlapping_shift.end_time
+ )
+ or (
+ curr_shift.start_time <= overlapping_shift.start_time
+ and curr_shift.end_time >= overlapping_shift.end_time
+ )
+ ):
+ return True
+ return False
def throw_overlap_error(self, shift_details):
shift_details = frappe._dict(shift_details)
+ msg = None
if shift_details.docstatus == 1 and shift_details.status == "Active":
- msg = _("Employee {0} already has Active Shift {1}: {2}").format(
- frappe.bold(self.employee), frappe.bold(self.shift_type), frappe.bold(shift_details.name)
- )
- if shift_details.start_date:
- msg += _(" from {0}").format(getdate(self.start_date).strftime("%d-%m-%Y"))
- title = "Ongoing Shift"
- if shift_details.end_date:
- msg += _(" to {0}").format(getdate(self.end_date).strftime("%d-%m-%Y"))
- title = "Active Shift"
+ if shift_details.start_date and shift_details.end_date:
+ msg = _("Employee {0} already has an active Shift {1}: {2} from {3} to {4}").format(
+ frappe.bold(self.employee),
+ frappe.bold(self.shift_type),
+ get_link_to_form("Shift Assignment", shift_details.name),
+ getdate(self.start_date).strftime("%d-%m-%Y"),
+ getdate(self.end_date).strftime("%d-%m-%Y"),
+ )
+ else:
+ msg = _("Employee {0} already has an active Shift {1}: {2} from {3}").format(
+ frappe.bold(self.employee),
+ frappe.bold(self.shift_type),
+ get_link_to_form("Shift Assignment", shift_details.name),
+ getdate(self.start_date).strftime("%d-%m-%Y"),
+ )
+
if msg:
- frappe.throw(msg, title=title)
+ frappe.throw(msg, title=_("Overlapping Shifts"), exc=OverlappingShiftError)
@frappe.whitelist()
@@ -155,102 +203,195 @@
return shift_timing_map
+def get_shift_for_time(shifts: List[Dict], for_timestamp: datetime) -> Dict:
+ """Returns shift with details for given timestamp"""
+ valid_shifts = []
+
+ for entry in shifts:
+ shift_details = get_shift_details(entry.shift_type, for_timestamp=for_timestamp)
+
+ if (
+ get_datetime(shift_details.actual_start)
+ <= get_datetime(for_timestamp)
+ <= get_datetime(shift_details.actual_end)
+ ):
+ valid_shifts.append(shift_details)
+
+ valid_shifts.sort(key=lambda x: x["actual_start"])
+
+ if len(valid_shifts) > 1:
+ for i in range(len(valid_shifts) - 1):
+ # comparing 2 consecutive shifts and adjusting start and end times
+ # if they are overlapping within grace period
+ curr_shift = valid_shifts[i]
+ next_shift = valid_shifts[i + 1]
+
+ if curr_shift and next_shift:
+ next_shift.actual_start = (
+ curr_shift.end_datetime
+ if next_shift.actual_start < curr_shift.end_datetime
+ else next_shift.actual_start
+ )
+ curr_shift.actual_end = (
+ next_shift.actual_start
+ if curr_shift.actual_end > next_shift.actual_start
+ else curr_shift.actual_end
+ )
+
+ valid_shifts[i] = curr_shift
+ valid_shifts[i + 1] = next_shift
+
+ return get_exact_shift(valid_shifts, for_timestamp) or {}
+
+ return (valid_shifts and valid_shifts[0]) or {}
+
+
+def get_shifts_for_date(employee: str, for_timestamp: datetime) -> List[Dict[str, str]]:
+ """Returns list of shifts with details for given date"""
+ assignment = frappe.qb.DocType("Shift Assignment")
+
+ return (
+ frappe.qb.from_(assignment)
+ .select(assignment.name, assignment.shift_type)
+ .where(
+ (assignment.employee == employee)
+ & (assignment.docstatus == 1)
+ & (assignment.status == "Active")
+ & (assignment.start_date <= getdate(for_timestamp.date()))
+ & (
+ Criterion.any(
+ [
+ assignment.end_date.isnull(),
+ (assignment.end_date.isnotnull() & (getdate(for_timestamp.date()) >= assignment.end_date)),
+ ]
+ )
+ )
+ )
+ ).run(as_dict=True)
+
+
+def get_shift_for_timestamp(employee: str, for_timestamp: datetime) -> Dict:
+ shifts = get_shifts_for_date(employee, for_timestamp)
+ if shifts:
+ return get_shift_for_time(shifts, for_timestamp)
+ return {}
+
+
def get_employee_shift(
- employee, for_date=None, consider_default_shift=False, next_shift_direction=None
-):
+ employee: str,
+ for_timestamp: datetime = None,
+ consider_default_shift: bool = False,
+ next_shift_direction: str = None,
+) -> Dict:
"""Returns a Shift Type for the given employee on the given date. (excluding the holidays)
:param employee: Employee for which shift is required.
- :param for_date: Date on which shift are required
+ :param for_timestamp: DateTime on which shift is required
:param consider_default_shift: If set to true, default shift is taken when no shift assignment is found.
:param next_shift_direction: One of: None, 'forward', 'reverse'. Direction to look for next shift if shift not found on given date.
"""
- if for_date is None:
- for_date = nowdate()
+ if for_timestamp is None:
+ for_timestamp = now_datetime()
+
+ shift_details = get_shift_for_timestamp(employee, for_timestamp)
+
+ # if shift assignment is not found, consider default shift
default_shift = frappe.db.get_value("Employee", employee, "default_shift")
- shift_type_name = None
- shift_assignment_details = frappe.db.get_value(
- "Shift Assignment",
- {"employee": employee, "start_date": ("<=", for_date), "docstatus": "1", "status": "Active"},
- ["shift_type", "end_date"],
+ if not shift_details and consider_default_shift:
+ shift_details = get_shift_details(default_shift, for_timestamp)
+
+ # if its a holiday, reset
+ if shift_details and is_holiday_date(employee, shift_details):
+ shift_details = None
+
+ # if no shift is found, find next or prev shift assignment based on direction
+ if not shift_details and next_shift_direction:
+ shift_details = get_prev_or_next_shift(
+ employee, for_timestamp, consider_default_shift, default_shift, next_shift_direction
+ )
+
+ return shift_details or {}
+
+
+def get_prev_or_next_shift(
+ employee: str,
+ for_timestamp: datetime,
+ consider_default_shift: bool,
+ default_shift: str,
+ next_shift_direction: str,
+) -> Dict:
+ """Returns a dict of shift details for the next or prev shift based on the next_shift_direction"""
+ MAX_DAYS = 366
+ shift_details = {}
+
+ if consider_default_shift and default_shift:
+ direction = -1 if next_shift_direction == "reverse" else 1
+ for i in range(MAX_DAYS):
+ date = for_timestamp + timedelta(days=direction * (i + 1))
+ shift_details = get_employee_shift(employee, date, consider_default_shift, None)
+ if shift_details:
+ break
+ else:
+ direction = "<" if next_shift_direction == "reverse" else ">"
+ sort_order = "desc" if next_shift_direction == "reverse" else "asc"
+ dates = frappe.db.get_all(
+ "Shift Assignment",
+ ["start_date", "end_date"],
+ {
+ "employee": employee,
+ "start_date": (direction, for_timestamp.date()),
+ "docstatus": 1,
+ "status": "Active",
+ },
+ as_list=True,
+ limit=MAX_DAYS,
+ order_by="start_date " + sort_order,
+ )
+
+ if dates:
+ for date in dates:
+ if date[1] and date[1] < for_timestamp.date():
+ continue
+ shift_details = get_employee_shift(
+ employee, datetime.combine(date[0], for_timestamp.time()), consider_default_shift, None
+ )
+ if shift_details:
+ break
+
+ return shift_details or {}
+
+
+def is_holiday_date(employee: str, shift_details: Dict) -> bool:
+ holiday_list_name = frappe.db.get_value(
+ "Shift Type", shift_details.shift_type.name, "holiday_list"
)
- if shift_assignment_details:
- shift_type_name = shift_assignment_details[0]
+ if not holiday_list_name:
+ holiday_list_name = get_holiday_list_for_employee(employee, False)
- # if end_date present means that shift is over after end_date else it is a ongoing shift.
- if shift_assignment_details[1] and for_date >= shift_assignment_details[1]:
- shift_type_name = None
-
- if not shift_type_name and consider_default_shift:
- shift_type_name = default_shift
- if shift_type_name:
- holiday_list_name = frappe.db.get_value("Shift Type", shift_type_name, "holiday_list")
- if not holiday_list_name:
- holiday_list_name = get_holiday_list_for_employee(employee, False)
- if holiday_list_name and is_holiday(holiday_list_name, for_date):
- shift_type_name = None
-
- if not shift_type_name and next_shift_direction:
- MAX_DAYS = 366
- if consider_default_shift and default_shift:
- direction = -1 if next_shift_direction == "reverse" else +1
- for i in range(MAX_DAYS):
- date = for_date + timedelta(days=direction * (i + 1))
- shift_details = get_employee_shift(employee, date, consider_default_shift, None)
- if shift_details:
- shift_type_name = shift_details.shift_type.name
- for_date = date
- break
- else:
- direction = "<" if next_shift_direction == "reverse" else ">"
- sort_order = "desc" if next_shift_direction == "reverse" else "asc"
- dates = frappe.db.get_all(
- "Shift Assignment",
- ["start_date", "end_date"],
- {
- "employee": employee,
- "start_date": (direction, for_date),
- "docstatus": "1",
- "status": "Active",
- },
- as_list=True,
- limit=MAX_DAYS,
- order_by="start_date " + sort_order,
- )
-
- if dates:
- for date in dates:
- if date[1] and date[1] < for_date:
- continue
- shift_details = get_employee_shift(employee, date[0], consider_default_shift, None)
- if shift_details:
- shift_type_name = shift_details.shift_type.name
- for_date = date[0]
- break
-
- return get_shift_details(shift_type_name, for_date)
+ return holiday_list_name and is_holiday(holiday_list_name, shift_details.start_datetime.date())
-def get_employee_shift_timings(employee, for_timestamp=None, consider_default_shift=False):
+def get_employee_shift_timings(
+ employee: str, for_timestamp: datetime = None, consider_default_shift: bool = False
+) -> List[Dict]:
"""Returns previous shift, current/upcoming shift, next_shift for the given timestamp and employee"""
if for_timestamp is None:
for_timestamp = now_datetime()
+
# write and verify a test case for midnight shift.
prev_shift = curr_shift = next_shift = None
- curr_shift = get_employee_shift(employee, for_timestamp.date(), consider_default_shift, "forward")
+ curr_shift = get_employee_shift(employee, for_timestamp, consider_default_shift, "forward")
if curr_shift:
next_shift = get_employee_shift(
- employee,
- curr_shift.start_datetime.date() + timedelta(days=1),
- consider_default_shift,
- "forward",
+ employee, curr_shift.start_datetime + timedelta(days=1), consider_default_shift, "forward"
)
prev_shift = get_employee_shift(
- employee, for_timestamp.date() + timedelta(days=-1), consider_default_shift, "reverse"
+ employee, for_timestamp + timedelta(days=-1), consider_default_shift, "reverse"
)
if curr_shift:
+ # adjust actual start and end times if they are overlapping with grace period (before start and after end)
if prev_shift:
curr_shift.actual_start = (
prev_shift.end_datetime
@@ -273,31 +414,102 @@
if curr_shift.actual_end > next_shift.actual_start
else curr_shift.actual_end
)
+
return prev_shift, curr_shift, next_shift
-def get_shift_details(shift_type_name, for_date=None):
- """Returns Shift Details which contain some additional information as described below.
- 'shift_details' contains the following keys:
- 'shift_type' - Object of DocType Shift Type,
- 'start_datetime' - Date and Time of shift start on given date,
- 'end_datetime' - Date and Time of shift end on given date,
- 'actual_start' - datetime of shift start after adding 'begin_check_in_before_shift_start_time',
- 'actual_end' - datetime of shift end after adding 'allow_check_out_after_shift_end_time'(None is returned if this is zero)
+def get_actual_start_end_datetime_of_shift(
+ employee: str, for_timestamp: datetime, consider_default_shift: bool = False
+) -> Dict:
+ """Returns a Dict containing shift details with actual_start and actual_end datetime values
+ Here 'actual' means taking into account the "begin_check_in_before_shift_start_time" and "allow_check_out_after_shift_end_time".
+ Empty Dict is returned if the timestamp is outside any actual shift timings.
- :param shift_type_name: shift type name for which shift_details is required.
- :param for_date: Date on which shift_details are required
+ :param employee (str): Employee name
+ :param for_timestamp (datetime, optional): Datetime value of checkin, if not provided considers current datetime
+ :param consider_default_shift (bool, optional): Flag (defaults to False) to specify whether to consider
+ default shift in employee master if no shift assignment is found
+ """
+ shift_timings_as_per_timestamp = get_employee_shift_timings(
+ employee, for_timestamp, consider_default_shift
+ )
+ return get_exact_shift(shift_timings_as_per_timestamp, for_timestamp)
+
+
+def get_exact_shift(shifts: List, for_timestamp: datetime) -> Dict:
+ """Returns the shift details (dict) for the exact shift in which the 'for_timestamp' value falls among multiple shifts"""
+ shift_details = dict()
+ timestamp_list = []
+
+ for shift in shifts:
+ if shift:
+ timestamp_list.extend([shift.actual_start, shift.actual_end])
+ else:
+ timestamp_list.extend([None, None])
+
+ timestamp_index = None
+ for index, timestamp in enumerate(timestamp_list):
+ if not timestamp:
+ continue
+
+ if for_timestamp < timestamp:
+ timestamp_index = index
+ elif for_timestamp == timestamp:
+ # on timestamp boundary
+ if index % 2 == 1:
+ timestamp_index = index
+ else:
+ timestamp_index = index + 1
+
+ if timestamp_index:
+ break
+
+ if timestamp_index and timestamp_index % 2 == 1:
+ shift_details = shifts[int((timestamp_index - 1) / 2)]
+
+ return shift_details
+
+
+def get_shift_details(shift_type_name: str, for_timestamp: datetime = None) -> Dict:
+ """Returns a Dict containing shift details with the following data:
+ 'shift_type' - Object of DocType Shift Type,
+ 'start_datetime' - datetime of shift start on given timestamp,
+ 'end_datetime' - datetime of shift end on given timestamp,
+ 'actual_start' - datetime of shift start after adding 'begin_check_in_before_shift_start_time',
+ 'actual_end' - datetime of shift end after adding 'allow_check_out_after_shift_end_time' (None is returned if this is zero)
+
+ :param shift_type_name (str): shift type name for which shift_details are required.
+ :param for_timestamp (datetime, optional): Datetime value of checkin, if not provided considers current datetime
"""
if not shift_type_name:
- return None
- if not for_date:
- for_date = nowdate()
+ return {}
+
+ if for_timestamp is None:
+ for_timestamp = now_datetime()
+
shift_type = frappe.get_doc("Shift Type", shift_type_name)
- start_datetime = datetime.combine(for_date, datetime.min.time()) + shift_type.start_time
- for_date = (
- for_date + timedelta(days=1) if shift_type.start_time > shift_type.end_time else for_date
+ shift_actual_start = shift_type.start_time - timedelta(
+ minutes=shift_type.begin_check_in_before_shift_start_time
)
- end_datetime = datetime.combine(for_date, datetime.min.time()) + shift_type.end_time
+
+ if shift_type.start_time > shift_type.end_time:
+ # shift spans accross 2 different days
+ if get_time(for_timestamp.time()) >= get_time(shift_actual_start):
+ # if for_timestamp is greater than start time, it's within the first day
+ start_datetime = datetime.combine(for_timestamp, datetime.min.time()) + shift_type.start_time
+ for_timestamp = for_timestamp + timedelta(days=1)
+ end_datetime = datetime.combine(for_timestamp, datetime.min.time()) + shift_type.end_time
+
+ elif get_time(for_timestamp.time()) < get_time(shift_actual_start):
+ # if for_timestamp is less than start time, it's within the second day
+ end_datetime = datetime.combine(for_timestamp, datetime.min.time()) + shift_type.end_time
+ for_timestamp = for_timestamp + timedelta(days=-1)
+ start_datetime = datetime.combine(for_timestamp, datetime.min.time()) + shift_type.start_time
+ else:
+ # start and end timings fall on the same day
+ start_datetime = datetime.combine(for_timestamp, datetime.min.time()) + shift_type.start_time
+ end_datetime = datetime.combine(for_timestamp, datetime.min.time()) + shift_type.end_time
+
actual_start = start_datetime - timedelta(
minutes=shift_type.begin_check_in_before_shift_start_time
)
@@ -312,34 +524,3 @@
"actual_end": actual_end,
}
)
-
-
-def get_actual_start_end_datetime_of_shift(employee, for_datetime, consider_default_shift=False):
- """Takes a datetime and returns the 'actual' start datetime and end datetime of the shift in which the timestamp belongs.
- Here 'actual' means - taking in to account the "begin_check_in_before_shift_start_time" and "allow_check_out_after_shift_end_time".
- None is returned if the timestamp is outside any actual shift timings.
- Shift Details is also returned(current/upcoming i.e. if timestamp not in any actual shift then details of next shift returned)
- """
- actual_shift_start = actual_shift_end = shift_details = None
- shift_timings_as_per_timestamp = get_employee_shift_timings(
- employee, for_datetime, consider_default_shift
- )
- timestamp_list = []
- for shift in shift_timings_as_per_timestamp:
- if shift:
- timestamp_list.extend([shift.actual_start, shift.actual_end])
- else:
- timestamp_list.extend([None, None])
- timestamp_index = None
- for index, timestamp in enumerate(timestamp_list):
- if timestamp and for_datetime <= timestamp:
- timestamp_index = index
- break
- if timestamp_index and timestamp_index % 2 == 1:
- shift_details = shift_timings_as_per_timestamp[int((timestamp_index - 1) / 2)]
- actual_shift_start = shift_details.actual_start
- actual_shift_end = shift_details.actual_end
- elif timestamp_index:
- shift_details = shift_timings_as_per_timestamp[int(timestamp_index / 2)]
-
- return actual_shift_start, actual_shift_end, shift_details
diff --git a/erpnext/hr/doctype/shift_type/shift_type.py b/erpnext/hr/doctype/shift_type/shift_type.py
index 3f5cb22..f5689d1 100644
--- a/erpnext/hr/doctype/shift_type/shift_type.py
+++ b/erpnext/hr/doctype/shift_type/shift_type.py
@@ -3,21 +3,23 @@
import itertools
-from datetime import timedelta
+from datetime import datetime, timedelta
import frappe
from frappe.model.document import Document
-from frappe.utils import cint, get_datetime, getdate
+from frappe.utils import cint, get_datetime, get_time, getdate
+from erpnext.buying.doctype.supplier_scorecard.supplier_scorecard import daterange
from erpnext.hr.doctype.attendance.attendance import mark_attendance
from erpnext.hr.doctype.employee.employee import get_holiday_list_for_employee
from erpnext.hr.doctype.employee_checkin.employee_checkin import (
calculate_working_hours,
mark_attendance_and_link_log,
)
+from erpnext.hr.doctype.holiday_list.holiday_list import is_holiday
from erpnext.hr.doctype.shift_assignment.shift_assignment import (
- get_actual_start_end_datetime_of_shift,
get_employee_shift,
+ get_shift_details,
)
@@ -30,8 +32,9 @@
or not self.last_sync_of_checkin
):
return
+
filters = {
- "skip_auto_attendance": "0",
+ "skip_auto_attendance": 0,
"attendance": ("is", "not set"),
"time": (">=", self.process_attendance_after),
"shift_actual_end": ("<", self.last_sync_of_checkin),
@@ -40,6 +43,7 @@
logs = frappe.db.get_list(
"Employee Checkin", fields="*", filters=filters, order_by="employee,time"
)
+
for key, group in itertools.groupby(
logs, key=lambda x: (x["employee"], x["shift_actual_start"])
):
@@ -52,6 +56,7 @@
in_time,
out_time,
) = self.get_attendance(single_shift_logs)
+
mark_attendance_and_link_log(
single_shift_logs,
attendance_status,
@@ -63,15 +68,16 @@
out_time,
self.name,
)
+
for employee in self.get_assigned_employee(self.process_attendance_after, True):
self.mark_absent_for_dates_with_no_attendance(employee)
def get_attendance(self, logs):
"""Return attendance_status, working_hours, late_entry, early_exit, in_time, out_time
for a set of logs belonging to a single shift.
- Assumtion:
- 1. These logs belongs to an single shift, single employee and is not in a holiday date.
- 2. Logs are in chronological order
+ Assumptions:
+ 1. These logs belongs to a single shift, single employee and it's not in a holiday date.
+ 2. Logs are in chronological order
"""
late_entry = early_exit = False
total_working_hours, in_time, out_time = calculate_working_hours(
@@ -107,52 +113,66 @@
"""Marks Absents for the given employee on working days in this shift which have no attendance marked.
The Absent is marked starting from 'process_attendance_after' or employee creation date.
"""
+ start_date, end_date = self.get_start_and_end_dates(employee)
+
+ # no shift assignment found, no need to process absent attendance records
+ if end_date is None:
+ return
+
+ holiday_list_name = self.holiday_list
+ if not holiday_list_name:
+ holiday_list_name = get_holiday_list_for_employee(employee, False)
+
+ start_time = get_time(self.start_time)
+
+ for date in daterange(getdate(start_date), getdate(end_date)):
+ if is_holiday(holiday_list_name, date):
+ # skip marking absent on a holiday
+ continue
+
+ timestamp = datetime.combine(date, start_time)
+ shift_details = get_employee_shift(employee, timestamp, True)
+
+ if shift_details and shift_details.shift_type.name == self.name:
+ mark_attendance(employee, date, "Absent", self.name)
+
+ def get_start_and_end_dates(self, employee):
date_of_joining, relieving_date, employee_creation = frappe.db.get_value(
"Employee", employee, ["date_of_joining", "relieving_date", "creation"]
)
+
if not date_of_joining:
date_of_joining = employee_creation.date()
+
start_date = max(getdate(self.process_attendance_after), date_of_joining)
- actual_shift_datetime = get_actual_start_end_datetime_of_shift(
- employee, get_datetime(self.last_sync_of_checkin), True
- )
+ end_date = None
+
+ shift_details = get_shift_details(self.name, get_datetime(self.last_sync_of_checkin))
last_shift_time = (
- actual_shift_datetime[0]
- if actual_shift_datetime[0]
- else get_datetime(self.last_sync_of_checkin)
+ shift_details.actual_start if shift_details else get_datetime(self.last_sync_of_checkin)
)
- prev_shift = get_employee_shift(
- employee, last_shift_time.date() - timedelta(days=1), True, "reverse"
- )
+
+ prev_shift = get_employee_shift(employee, last_shift_time - timedelta(days=1), True, "reverse")
if prev_shift:
end_date = (
min(prev_shift.start_datetime.date(), relieving_date)
if relieving_date
else prev_shift.start_datetime.date()
)
- else:
- return
- holiday_list_name = self.holiday_list
- if not holiday_list_name:
- holiday_list_name = get_holiday_list_for_employee(employee, False)
- dates = get_filtered_date_list(employee, start_date, end_date, holiday_list=holiday_list_name)
- for date in dates:
- shift_details = get_employee_shift(employee, date, True)
- if shift_details and shift_details.shift_type.name == self.name:
- mark_attendance(employee, date, "Absent", self.name)
+
+ return start_date, end_date
def get_assigned_employee(self, from_date=None, consider_default_shift=False):
- filters = {"start_date": (">", from_date), "shift_type": self.name, "docstatus": "1"}
- if not from_date:
- del filters["start_date"]
+ filters = {"shift_type": self.name, "docstatus": "1"}
+ if from_date:
+ filters["start_date"] = (">", from_date)
- assigned_employees = frappe.get_all("Shift Assignment", "employee", filters, as_list=True)
- assigned_employees = [x[0] for x in assigned_employees]
+ assigned_employees = frappe.get_all("Shift Assignment", filters=filters, pluck="employee")
if consider_default_shift:
filters = {"default_shift": self.name, "status": ["!=", "Inactive"]}
- default_shift_employees = frappe.get_all("Employee", "name", filters, as_list=True)
- default_shift_employees = [x[0] for x in default_shift_employees]
+ default_shift_employees = frappe.get_all("Employee", filters=filters, pluck="name")
+
return list(set(assigned_employees + default_shift_employees))
return assigned_employees
@@ -162,42 +182,3 @@
for shift in shift_list:
doc = frappe.get_doc("Shift Type", shift[0])
doc.process_auto_attendance()
-
-
-def get_filtered_date_list(
- employee, start_date, end_date, filter_attendance=True, holiday_list=None
-):
- """Returns a list of dates after removing the dates with attendance and holidays"""
- base_dates_query = """select adddate(%(start_date)s, t2.i*100 + t1.i*10 + t0.i) selected_date from
- (select 0 i union select 1 union select 2 union select 3 union select 4 union select 5 union select 6 union select 7 union select 8 union select 9) t0,
- (select 0 i union select 1 union select 2 union select 3 union select 4 union select 5 union select 6 union select 7 union select 8 union select 9) t1,
- (select 0 i union select 1 union select 2 union select 3 union select 4 union select 5 union select 6 union select 7 union select 8 union select 9) t2"""
- condition_query = ""
- if filter_attendance:
- condition_query += """ and a.selected_date not in (
- select attendance_date from `tabAttendance`
- where docstatus = 1 and employee = %(employee)s
- and attendance_date between %(start_date)s and %(end_date)s)"""
- if holiday_list:
- condition_query += """ and a.selected_date not in (
- select holiday_date from `tabHoliday` where parenttype = 'Holiday List' and
- parentfield = 'holidays' and parent = %(holiday_list)s
- and holiday_date between %(start_date)s and %(end_date)s)"""
-
- dates = frappe.db.sql(
- """select * from
- ({base_dates_query}) as a
- where a.selected_date <= %(end_date)s {condition_query}
- """.format(
- base_dates_query=base_dates_query, condition_query=condition_query
- ),
- {
- "employee": employee,
- "start_date": start_date,
- "end_date": end_date,
- "holiday_list": holiday_list,
- },
- as_list=True,
- )
-
- return [getdate(date[0]) for date in dates]
diff --git a/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.js b/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.js
index 42f7cdb..6f4bbd5 100644
--- a/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.js
+++ b/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.js
@@ -66,8 +66,7 @@
"Default": 0,
}
],
-
- "onload": function() {
+ onload: function() {
return frappe.call({
method: "erpnext.hr.report.monthly_attendance_sheet.monthly_attendance_sheet.get_attendance_years",
callback: function(r) {
@@ -78,5 +77,25 @@
year_filter.set_input(year_filter.df.default);
}
});
+ },
+ formatter: function(value, row, column, data, default_formatter) {
+ value = default_formatter(value, row, column, data);
+ const summarized_view = frappe.query_report.get_filter_value('summarized_view');
+ const group_by = frappe.query_report.get_filter_value('group_by');
+
+ if (!summarized_view) {
+ if ((group_by && column.colIndex > 3) || (!group_by && column.colIndex > 2)) {
+ if (value == 'P' || value == 'WFH')
+ value = "<span style='color:green'>" + value + "</span>";
+ else if (value == 'A')
+ value = "<span style='color:red'>" + value + "</span>";
+ else if (value == 'HD')
+ value = "<span style='color:orange'>" + value + "</span>";
+ else if (value == 'L')
+ value = "<span style='color:#318AD8'>" + value + "</span>";
+ }
+ }
+
+ return value;
}
}
diff --git a/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.py b/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.py
index 8ea4989..efd2d38 100644
--- a/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.py
+++ b/erpnext/hr/report/monthly_attendance_sheet/monthly_attendance_sheet.py
@@ -3,365 +3,618 @@
from calendar import monthrange
+from itertools import groupby
+from typing import Dict, List, Optional, Tuple
import frappe
-from frappe import _, msgprint
+from frappe import _
+from frappe.query_builder.functions import Count, Extract, Sum
from frappe.utils import cint, cstr, getdate
+Filters = frappe._dict
+
status_map = {
+ "Present": "P",
"Absent": "A",
"Half Day": "HD",
- "Holiday": "<b>H</b>",
- "Weekly Off": "<b>WO</b>",
- "On Leave": "L",
- "Present": "P",
"Work From Home": "WFH",
+ "On Leave": "L",
+ "Holiday": "H",
+ "Weekly Off": "WO",
}
day_abbr = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
-def execute(filters=None):
- if not filters:
- filters = {}
+def execute(filters: Optional[Filters] = None) -> Tuple:
+ filters = frappe._dict(filters or {})
- if filters.hide_year_field == 1:
- filters.year = 2020
+ if not (filters.month and filters.year):
+ frappe.throw(_("Please select month and year."))
- conditions, filters = get_conditions(filters)
- columns, days = get_columns(filters)
- att_map = get_attendance_list(conditions, filters)
- if not att_map:
+ attendance_map = get_attendance_map(filters)
+ if not attendance_map:
+ frappe.msgprint(_("No attendance records found."), alert=True, indicator="orange")
+ return [], [], None, None
+
+ columns = get_columns(filters)
+ data = get_data(filters, attendance_map)
+
+ if not data:
+ frappe.msgprint(
+ _("No attendance records found for this criteria."), alert=True, indicator="orange"
+ )
return columns, [], None, None
- if filters.group_by:
- emp_map, group_by_parameters = get_employee_details(filters.group_by, filters.company)
- holiday_list = []
- for parameter in group_by_parameters:
- h_list = [
- emp_map[parameter][d]["holiday_list"]
- for d in emp_map[parameter]
- if emp_map[parameter][d]["holiday_list"]
- ]
- holiday_list += h_list
- else:
- emp_map = get_employee_details(filters.group_by, filters.company)
- holiday_list = [emp_map[d]["holiday_list"] for d in emp_map if emp_map[d]["holiday_list"]]
+ message = get_message() if not filters.summarized_view else ""
+ chart = get_chart_data(attendance_map, filters)
- default_holiday_list = frappe.get_cached_value(
- "Company", filters.get("company"), "default_holiday_list"
- )
- holiday_list.append(default_holiday_list)
- holiday_list = list(set(holiday_list))
- holiday_map = get_holiday(holiday_list, filters["month"])
-
- data = []
-
- leave_types = frappe.db.get_list("Leave Type")
- leave_list = None
- if filters.summarized_view:
- leave_list = [d.name + ":Float:120" for d in leave_types]
- columns.extend(leave_list)
- columns.extend([_("Total Late Entries") + ":Float:120", _("Total Early Exits") + ":Float:120"])
-
- if filters.group_by:
- emp_att_map = {}
- for parameter in group_by_parameters:
- emp_map_set = set([key for key in emp_map[parameter].keys()])
- att_map_set = set([key for key in att_map.keys()])
- if att_map_set & emp_map_set:
- parameter_row = ["<b>" + parameter + "</b>"] + [
- "" for day in range(filters["total_days_in_month"] + 2)
- ]
- data.append(parameter_row)
- record, emp_att_data = add_data(
- emp_map[parameter],
- att_map,
- filters,
- holiday_map,
- conditions,
- default_holiday_list,
- leave_types=leave_types,
- )
- emp_att_map.update(emp_att_data)
- data += record
- else:
- record, emp_att_map = add_data(
- emp_map,
- att_map,
- filters,
- holiday_map,
- conditions,
- default_holiday_list,
- leave_types=leave_types,
- )
- data += record
-
- chart_data = get_chart_data(emp_att_map, days)
-
- return columns, data, None, chart_data
+ return columns, data, message, chart
-def get_chart_data(emp_att_map, days):
- labels = []
- datasets = [
- {"name": "Absent", "values": []},
- {"name": "Present", "values": []},
- {"name": "Leave", "values": []},
- ]
- for idx, day in enumerate(days, start=0):
- p = day.replace("::65", "")
- labels.append(day.replace("::65", ""))
- total_absent_on_day = 0
- total_leave_on_day = 0
- total_present_on_day = 0
- total_holiday = 0
- for emp in emp_att_map.keys():
- if emp_att_map[emp][idx]:
- if emp_att_map[emp][idx] == "A":
- total_absent_on_day += 1
- if emp_att_map[emp][idx] in ["P", "WFH"]:
- total_present_on_day += 1
- if emp_att_map[emp][idx] == "HD":
- total_present_on_day += 0.5
- total_leave_on_day += 0.5
- if emp_att_map[emp][idx] == "L":
- total_leave_on_day += 1
+def get_message() -> str:
+ message = ""
+ colors = ["green", "red", "orange", "green", "#318AD8", "", ""]
- datasets[0]["values"].append(total_absent_on_day)
- datasets[1]["values"].append(total_present_on_day)
- datasets[2]["values"].append(total_leave_on_day)
+ count = 0
+ for status, abbr in status_map.items():
+ message += f"""
+ <span style='border-left: 2px solid {colors[count]}; padding-right: 12px; padding-left: 5px; margin-right: 3px;'>
+ {status} - {abbr}
+ </span>
+ """
+ count += 1
- chart = {"data": {"labels": labels, "datasets": datasets}}
-
- chart["type"] = "line"
-
- return chart
+ return message
-def add_data(
- employee_map, att_map, filters, holiday_map, conditions, default_holiday_list, leave_types=None
-):
-
- record = []
- emp_att_map = {}
- for emp in employee_map:
- emp_det = employee_map.get(emp)
- if not emp_det or emp not in att_map:
- continue
-
- row = []
- if filters.group_by:
- row += [" "]
- row += [emp, emp_det.employee_name]
-
- total_p = total_a = total_l = total_h = total_um = 0.0
- emp_status_map = []
- for day in range(filters["total_days_in_month"]):
- status = None
- status = att_map.get(emp).get(day + 1)
-
- if status is None and holiday_map:
- emp_holiday_list = emp_det.holiday_list if emp_det.holiday_list else default_holiday_list
-
- if emp_holiday_list in holiday_map:
- for idx, ele in enumerate(holiday_map[emp_holiday_list]):
- if day + 1 == holiday_map[emp_holiday_list][idx][0]:
- if holiday_map[emp_holiday_list][idx][1]:
- status = "Weekly Off"
- else:
- status = "Holiday"
- total_h += 1
-
- abbr = status_map.get(status, "")
- emp_status_map.append(abbr)
-
- if filters.summarized_view:
- if status == "Present" or status == "Work From Home":
- total_p += 1
- elif status == "Absent":
- total_a += 1
- elif status == "On Leave":
- total_l += 1
- elif status == "Half Day":
- total_p += 0.5
- total_a += 0.5
- total_l += 0.5
- elif not status:
- total_um += 1
-
- if not filters.summarized_view:
- row += emp_status_map
-
- if filters.summarized_view:
- row += [total_p, total_l, total_a, total_h, total_um]
-
- if not filters.get("employee"):
- filters.update({"employee": emp})
- conditions += " and employee = %(employee)s"
- elif not filters.get("employee") == emp:
- filters.update({"employee": emp})
-
- if filters.summarized_view:
- leave_details = frappe.db.sql(
- """select leave_type, status, count(*) as count from `tabAttendance`\
- where leave_type is not NULL %s group by leave_type, status"""
- % conditions,
- filters,
- as_dict=1,
- )
-
- time_default_counts = frappe.db.sql(
- """select (select count(*) from `tabAttendance` where \
- late_entry = 1 %s) as late_entry_count, (select count(*) from tabAttendance where \
- early_exit = 1 %s) as early_exit_count"""
- % (conditions, conditions),
- filters,
- )
-
- leaves = {}
- for d in leave_details:
- if d.status == "Half Day":
- d.count = d.count * 0.5
- if d.leave_type in leaves:
- leaves[d.leave_type] += d.count
- else:
- leaves[d.leave_type] = d.count
-
- for d in leave_types:
- if d.name in leaves:
- row.append(leaves[d.name])
- else:
- row.append("0.0")
-
- row.extend([time_default_counts[0][0], time_default_counts[0][1]])
- emp_att_map[emp] = emp_status_map
- record.append(row)
-
- return record, emp_att_map
-
-
-def get_columns(filters):
-
+def get_columns(filters: Filters) -> List[Dict]:
columns = []
if filters.group_by:
- columns = [_(filters.group_by) + ":Link/Branch:120"]
+ columns.append(
+ {
+ "label": _(filters.group_by),
+ "fieldname": frappe.scrub(filters.group_by),
+ "fieldtype": "Link",
+ "options": "Branch",
+ "width": 120,
+ }
+ )
- columns += [_("Employee") + ":Link/Employee:120", _("Employee Name") + ":Data/:120"]
- days = []
- for day in range(filters["total_days_in_month"]):
- date = str(filters.year) + "-" + str(filters.month) + "-" + str(day + 1)
- day_name = day_abbr[getdate(date).weekday()]
- days.append(cstr(day + 1) + " " + day_name + "::65")
- if not filters.summarized_view:
- columns += days
-
- if filters.summarized_view:
- columns += [
- _("Total Present") + ":Float:120",
- _("Total Leaves") + ":Float:120",
- _("Total Absent") + ":Float:120",
- _("Total Holidays") + ":Float:120",
- _("Unmarked Days") + ":Float:120",
+ columns.extend(
+ [
+ {
+ "label": _("Employee"),
+ "fieldname": "employee",
+ "fieldtype": "Link",
+ "options": "Employee",
+ "width": 135,
+ },
+ {"label": _("Employee Name"), "fieldname": "employee_name", "fieldtype": "Data", "width": 120},
]
- return columns, days
-
-
-def get_attendance_list(conditions, filters):
- attendance_list = frappe.db.sql(
- """select employee, day(attendance_date) as day_of_month,
- status from tabAttendance where docstatus = 1 %s order by employee, attendance_date"""
- % conditions,
- filters,
- as_dict=1,
)
- if not attendance_list:
- msgprint(_("No attendance record found"), alert=True, indicator="orange")
+ if filters.summarized_view:
+ columns.extend(
+ [
+ {
+ "label": _("Total Present"),
+ "fieldname": "total_present",
+ "fieldtype": "Float",
+ "width": 110,
+ },
+ {"label": _("Total Leaves"), "fieldname": "total_leaves", "fieldtype": "Float", "width": 110},
+ {"label": _("Total Absent"), "fieldname": "total_absent", "fieldtype": "Float", "width": 110},
+ {
+ "label": _("Total Holidays"),
+ "fieldname": "total_holidays",
+ "fieldtype": "Float",
+ "width": 120,
+ },
+ {
+ "label": _("Unmarked Days"),
+ "fieldname": "unmarked_days",
+ "fieldtype": "Float",
+ "width": 130,
+ },
+ ]
+ )
+ columns.extend(get_columns_for_leave_types())
+ columns.extend(
+ [
+ {
+ "label": _("Total Late Entries"),
+ "fieldname": "total_late_entries",
+ "fieldtype": "Float",
+ "width": 140,
+ },
+ {
+ "label": _("Total Early Exits"),
+ "fieldname": "total_early_exits",
+ "fieldtype": "Float",
+ "width": 140,
+ },
+ ]
+ )
+ else:
+ columns.append({"label": _("Shift"), "fieldname": "shift", "fieldtype": "Data", "width": 120})
+ columns.extend(get_columns_for_days(filters))
- att_map = {}
+ return columns
+
+
+def get_columns_for_leave_types() -> List[Dict]:
+ leave_types = frappe.db.get_all("Leave Type", pluck="name")
+ types = []
+ for entry in leave_types:
+ types.append(
+ {"label": entry, "fieldname": frappe.scrub(entry), "fieldtype": "Float", "width": 120}
+ )
+
+ return types
+
+
+def get_columns_for_days(filters: Filters) -> List[Dict]:
+ total_days = get_total_days_in_month(filters)
+ days = []
+
+ for day in range(1, total_days + 1):
+ # forms the dates from selected year and month from filters
+ date = "{}-{}-{}".format(cstr(filters.year), cstr(filters.month), cstr(day))
+ # gets abbr from weekday number
+ weekday = day_abbr[getdate(date).weekday()]
+ # sets days as 1 Mon, 2 Tue, 3 Wed
+ label = "{} {}".format(cstr(day), weekday)
+ days.append({"label": label, "fieldtype": "Data", "fieldname": day, "width": 65})
+
+ return days
+
+
+def get_total_days_in_month(filters: Filters) -> int:
+ return monthrange(cint(filters.year), cint(filters.month))[1]
+
+
+def get_data(filters: Filters, attendance_map: Dict) -> List[Dict]:
+ employee_details, group_by_param_values = get_employee_related_details(
+ filters.group_by, filters.company
+ )
+ holiday_map = get_holiday_map(filters)
+ data = []
+
+ if filters.group_by:
+ group_by_column = frappe.scrub(filters.group_by)
+
+ for value in group_by_param_values:
+ if not value:
+ continue
+
+ records = get_rows(employee_details[value], filters, holiday_map, attendance_map)
+
+ if records:
+ data.append({group_by_column: frappe.bold(value)})
+ data.extend(records)
+ else:
+ data = get_rows(employee_details, filters, holiday_map, attendance_map)
+
+ return data
+
+
+def get_attendance_map(filters: Filters) -> Dict:
+ """Returns a dictionary of employee wise attendance map as per shifts for all the days of the month like
+ {
+ 'employee1': {
+ 'Morning Shift': {1: 'Present', 2: 'Absent', ...}
+ 'Evening Shift': {1: 'Absent', 2: 'Present', ...}
+ },
+ 'employee2': {
+ 'Afternoon Shift': {1: 'Present', 2: 'Absent', ...}
+ 'Night Shift': {1: 'Absent', 2: 'Absent', ...}
+ }
+ }
+ """
+ Attendance = frappe.qb.DocType("Attendance")
+ query = (
+ frappe.qb.from_(Attendance)
+ .select(
+ Attendance.employee,
+ Extract("day", Attendance.attendance_date).as_("day_of_month"),
+ Attendance.status,
+ Attendance.shift,
+ )
+ .where(
+ (Attendance.docstatus == 1)
+ & (Attendance.company == filters.company)
+ & (Extract("month", Attendance.attendance_date) == filters.month)
+ & (Extract("year", Attendance.attendance_date) == filters.year)
+ )
+ )
+ if filters.employee:
+ query = query.where(Attendance.employee == filters.employee)
+ query = query.orderby(Attendance.employee, Attendance.attendance_date)
+
+ attendance_list = query.run(as_dict=1)
+ attendance_map = {}
+
for d in attendance_list:
- att_map.setdefault(d.employee, frappe._dict()).setdefault(d.day_of_month, "")
- att_map[d.employee][d.day_of_month] = d.status
+ attendance_map.setdefault(d.employee, frappe._dict()).setdefault(d.shift, frappe._dict())
+ attendance_map[d.employee][d.shift][d.day_of_month] = d.status
- return att_map
+ return attendance_map
-def get_conditions(filters):
- if not (filters.get("month") and filters.get("year")):
- msgprint(_("Please select month and year"), raise_exception=1)
-
- filters["total_days_in_month"] = monthrange(cint(filters.year), cint(filters.month))[1]
-
- conditions = " and month(attendance_date) = %(month)s and year(attendance_date) = %(year)s"
-
- if filters.get("company"):
- conditions += " and company = %(company)s"
- if filters.get("employee"):
- conditions += " and employee = %(employee)s"
-
- return conditions, filters
-
-
-def get_employee_details(group_by, company):
- emp_map = {}
- query = """select name, employee_name, designation, department, branch, company,
- holiday_list from `tabEmployee` where company = %s """ % frappe.db.escape(
- company
+def get_employee_related_details(group_by: str, company: str) -> Tuple[Dict, List]:
+ """Returns
+ 1. nested dict for employee details
+ 2. list of values for the group by filter
+ """
+ Employee = frappe.qb.DocType("Employee")
+ query = (
+ frappe.qb.from_(Employee)
+ .select(
+ Employee.name,
+ Employee.employee_name,
+ Employee.designation,
+ Employee.grade,
+ Employee.department,
+ Employee.branch,
+ Employee.company,
+ Employee.holiday_list,
+ )
+ .where(Employee.company == company)
)
if group_by:
group_by = group_by.lower()
- query += " order by " + group_by + " ASC"
+ query = query.orderby(group_by)
- employee_details = frappe.db.sql(query, as_dict=1)
+ employee_details = query.run(as_dict=True)
- group_by_parameters = []
+ group_by_param_values = []
+ emp_map = {}
+
if group_by:
+ for parameter, employees in groupby(employee_details, key=lambda d: d[group_by]):
+ group_by_param_values.append(parameter)
+ emp_map.setdefault(parameter, frappe._dict())
- group_by_parameters = list(
- set(detail.get(group_by, "") for detail in employee_details if detail.get(group_by, ""))
- )
- for parameter in group_by_parameters:
- emp_map[parameter] = {}
-
- for d in employee_details:
- if group_by and len(group_by_parameters):
- if d.get(group_by, None):
-
- emp_map[d.get(group_by)][d.name] = d
- else:
- emp_map[d.name] = d
-
- if not group_by:
- return emp_map
+ for emp in employees:
+ emp_map[parameter][emp.name] = emp
else:
- return emp_map, group_by_parameters
+ for emp in employee_details:
+ emp_map[emp.name] = emp
+
+ return emp_map, group_by_param_values
-def get_holiday(holiday_list, month):
+def get_holiday_map(filters: Filters) -> Dict[str, List[Dict]]:
+ """
+ Returns a dict of holidays falling in the filter month and year
+ with list name as key and list of holidays as values like
+ {
+ 'Holiday List 1': [
+ {'day_of_month': '0' , 'weekly_off': 1},
+ {'day_of_month': '1', 'weekly_off': 0}
+ ],
+ 'Holiday List 2': [
+ {'day_of_month': '0' , 'weekly_off': 1},
+ {'day_of_month': '1', 'weekly_off': 0}
+ ]
+ }
+ """
+ # add default holiday list too
+ holiday_lists = frappe.db.get_all("Holiday List", pluck="name")
+ default_holiday_list = frappe.get_cached_value("Company", filters.company, "default_holiday_list")
+ holiday_lists.append(default_holiday_list)
+
holiday_map = frappe._dict()
- for d in holiday_list:
- if d:
- holiday_map.setdefault(
- d,
- frappe.db.sql(
- """select day(holiday_date), weekly_off from `tabHoliday`
- where parent=%s and month(holiday_date)=%s""",
- (d, month),
- ),
+ Holiday = frappe.qb.DocType("Holiday")
+
+ for d in holiday_lists:
+ if not d:
+ continue
+
+ holidays = (
+ frappe.qb.from_(Holiday)
+ .select(Extract("day", Holiday.holiday_date).as_("day_of_month"), Holiday.weekly_off)
+ .where(
+ (Holiday.parent == d)
+ & (Extract("month", Holiday.holiday_date) == filters.month)
+ & (Extract("year", Holiday.holiday_date) == filters.year)
)
+ ).run(as_dict=True)
+
+ holiday_map.setdefault(d, holidays)
return holiday_map
-@frappe.whitelist()
-def get_attendance_years():
- year_list = frappe.db.sql_list(
- """select distinct YEAR(attendance_date) from tabAttendance ORDER BY YEAR(attendance_date) DESC"""
+def get_rows(
+ employee_details: Dict, filters: Filters, holiday_map: Dict, attendance_map: Dict
+) -> List[Dict]:
+ records = []
+ default_holiday_list = frappe.get_cached_value("Company", filters.company, "default_holiday_list")
+
+ for employee, details in employee_details.items():
+ emp_holiday_list = details.holiday_list or default_holiday_list
+ holidays = holiday_map.get(emp_holiday_list)
+
+ if filters.summarized_view:
+ attendance = get_attendance_status_for_summarized_view(employee, filters, holidays)
+ if not attendance:
+ continue
+
+ leave_summary = get_leave_summary(employee, filters)
+ entry_exits_summary = get_entry_exits_summary(employee, filters)
+
+ row = {"employee": employee, "employee_name": details.employee_name}
+ set_defaults_for_summarized_view(filters, row)
+ row.update(attendance)
+ row.update(leave_summary)
+ row.update(entry_exits_summary)
+
+ records.append(row)
+ else:
+ employee_attendance = attendance_map.get(employee)
+ if not employee_attendance:
+ continue
+
+ attendance_for_employee = get_attendance_status_for_detailed_view(
+ employee, filters, employee_attendance, holidays
+ )
+ # set employee details in the first row
+ attendance_for_employee[0].update(
+ {"employee": employee, "employee_name": details.employee_name}
+ )
+
+ records.extend(attendance_for_employee)
+
+ return records
+
+
+def set_defaults_for_summarized_view(filters, row):
+ for entry in get_columns(filters):
+ if entry.get("fieldtype") == "Float":
+ row[entry.get("fieldname")] = 0.0
+
+
+def get_attendance_status_for_summarized_view(
+ employee: str, filters: Filters, holidays: List
+) -> Dict:
+ """Returns dict of attendance status for employee like
+ {'total_present': 1.5, 'total_leaves': 0.5, 'total_absent': 13.5, 'total_holidays': 8, 'unmarked_days': 5}
+ """
+ summary, attendance_days = get_attendance_summary_and_days(employee, filters)
+ if not any(summary.values()):
+ return {}
+
+ total_days = get_total_days_in_month(filters)
+ total_holidays = total_unmarked_days = 0
+
+ for day in range(1, total_days + 1):
+ if day in attendance_days:
+ continue
+
+ status = get_holiday_status(day, holidays)
+ if status in ["Weekly Off", "Holiday"]:
+ total_holidays += 1
+ elif not status:
+ total_unmarked_days += 1
+
+ return {
+ "total_present": summary.total_present + summary.total_half_days,
+ "total_leaves": summary.total_leaves + summary.total_half_days,
+ "total_absent": summary.total_absent + summary.total_half_days,
+ "total_holidays": total_holidays,
+ "unmarked_days": total_unmarked_days,
+ }
+
+
+def get_attendance_summary_and_days(employee: str, filters: Filters) -> Tuple[Dict, List]:
+ Attendance = frappe.qb.DocType("Attendance")
+
+ present_case = (
+ frappe.qb.terms.Case()
+ .when(((Attendance.status == "Present") | (Attendance.status == "Work From Home")), 1)
+ .else_(0)
)
- if not year_list:
+ sum_present = Sum(present_case).as_("total_present")
+
+ absent_case = frappe.qb.terms.Case().when(Attendance.status == "Absent", 1).else_(0)
+ sum_absent = Sum(absent_case).as_("total_absent")
+
+ leave_case = frappe.qb.terms.Case().when(Attendance.status == "On Leave", 1).else_(0)
+ sum_leave = Sum(leave_case).as_("total_leaves")
+
+ half_day_case = frappe.qb.terms.Case().when(Attendance.status == "Half Day", 0.5).else_(0)
+ sum_half_day = Sum(half_day_case).as_("total_half_days")
+
+ summary = (
+ frappe.qb.from_(Attendance)
+ .select(
+ sum_present,
+ sum_absent,
+ sum_leave,
+ sum_half_day,
+ )
+ .where(
+ (Attendance.docstatus == 1)
+ & (Attendance.employee == employee)
+ & (Attendance.company == filters.company)
+ & (Extract("month", Attendance.attendance_date) == filters.month)
+ & (Extract("year", Attendance.attendance_date) == filters.year)
+ )
+ ).run(as_dict=True)
+
+ days = (
+ frappe.qb.from_(Attendance)
+ .select(Extract("day", Attendance.attendance_date).as_("day_of_month"))
+ .distinct()
+ .where(
+ (Attendance.docstatus == 1)
+ & (Attendance.employee == employee)
+ & (Attendance.company == filters.company)
+ & (Extract("month", Attendance.attendance_date) == filters.month)
+ & (Extract("year", Attendance.attendance_date) == filters.year)
+ )
+ ).run(pluck=True)
+
+ return summary[0], days
+
+
+def get_attendance_status_for_detailed_view(
+ employee: str, filters: Filters, employee_attendance: Dict, holidays: List
+) -> List[Dict]:
+ """Returns list of shift-wise attendance status for employee
+ [
+ {'shift': 'Morning Shift', 1: 'A', 2: 'P', 3: 'A'....},
+ {'shift': 'Evening Shift', 1: 'P', 2: 'A', 3: 'P'....}
+ ]
+ """
+ total_days = get_total_days_in_month(filters)
+ attendance_values = []
+
+ for shift, status_dict in employee_attendance.items():
+ row = {"shift": shift}
+
+ for day in range(1, total_days + 1):
+ status = status_dict.get(day)
+ if status is None and holidays:
+ status = get_holiday_status(day, holidays)
+
+ abbr = status_map.get(status, "")
+ row[day] = abbr
+
+ attendance_values.append(row)
+
+ return attendance_values
+
+
+def get_holiday_status(day: int, holidays: List) -> str:
+ status = None
+ for holiday in holidays:
+ if day == holiday.get("day_of_month"):
+ if holiday.get("weekly_off"):
+ status = "Weekly Off"
+ else:
+ status = "Holiday"
+ break
+ return status
+
+
+def get_leave_summary(employee: str, filters: Filters) -> Dict[str, float]:
+ """Returns a dict of leave type and corresponding leaves taken by employee like:
+ {'leave_without_pay': 1.0, 'sick_leave': 2.0}
+ """
+ Attendance = frappe.qb.DocType("Attendance")
+ day_case = frappe.qb.terms.Case().when(Attendance.status == "Half Day", 0.5).else_(1)
+ sum_leave_days = Sum(day_case).as_("leave_days")
+
+ leave_details = (
+ frappe.qb.from_(Attendance)
+ .select(Attendance.leave_type, sum_leave_days)
+ .where(
+ (Attendance.employee == employee)
+ & (Attendance.docstatus == 1)
+ & (Attendance.company == filters.company)
+ & ((Attendance.leave_type.isnotnull()) | (Attendance.leave_type != ""))
+ & (Extract("month", Attendance.attendance_date) == filters.month)
+ & (Extract("year", Attendance.attendance_date) == filters.year)
+ )
+ .groupby(Attendance.leave_type)
+ ).run(as_dict=True)
+
+ leaves = {}
+ for d in leave_details:
+ leave_type = frappe.scrub(d.leave_type)
+ leaves[leave_type] = d.leave_days
+
+ return leaves
+
+
+def get_entry_exits_summary(employee: str, filters: Filters) -> Dict[str, float]:
+ """Returns total late entries and total early exits for employee like:
+ {'total_late_entries': 5, 'total_early_exits': 2}
+ """
+ Attendance = frappe.qb.DocType("Attendance")
+
+ late_entry_case = frappe.qb.terms.Case().when(Attendance.late_entry == "1", "1")
+ count_late_entries = Count(late_entry_case).as_("total_late_entries")
+
+ early_exit_case = frappe.qb.terms.Case().when(Attendance.early_exit == "1", "1")
+ count_early_exits = Count(early_exit_case).as_("total_early_exits")
+
+ entry_exits = (
+ frappe.qb.from_(Attendance)
+ .select(count_late_entries, count_early_exits)
+ .where(
+ (Attendance.docstatus == 1)
+ & (Attendance.employee == employee)
+ & (Attendance.company == filters.company)
+ & (Extract("month", Attendance.attendance_date) == filters.month)
+ & (Extract("year", Attendance.attendance_date) == filters.year)
+ )
+ ).run(as_dict=True)
+
+ return entry_exits[0]
+
+
+@frappe.whitelist()
+def get_attendance_years() -> str:
+ """Returns all the years for which attendance records exist"""
+ Attendance = frappe.qb.DocType("Attendance")
+ year_list = (
+ frappe.qb.from_(Attendance)
+ .select(Extract("year", Attendance.attendance_date).as_("year"))
+ .distinct()
+ ).run(as_dict=True)
+
+ if year_list:
+ year_list.sort(key=lambda d: d.year, reverse=True)
+ else:
year_list = [getdate().year]
- return "\n".join(str(year) for year in year_list)
+ return "\n".join(cstr(entry.year) for entry in year_list)
+
+
+def get_chart_data(attendance_map: Dict, filters: Filters) -> Dict:
+ days = get_columns_for_days(filters)
+ labels = []
+ absent = []
+ present = []
+ leave = []
+
+ for day in days:
+ labels.append(day["label"])
+ total_absent_on_day = total_leaves_on_day = total_present_on_day = 0
+
+ for employee, attendance_dict in attendance_map.items():
+ for shift, attendance in attendance_dict.items():
+ attendance_on_day = attendance.get(day["fieldname"])
+
+ if attendance_on_day == "Absent":
+ total_absent_on_day += 1
+ elif attendance_on_day in ["Present", "Work From Home"]:
+ total_present_on_day += 1
+ elif attendance_on_day == "Half Day":
+ total_present_on_day += 0.5
+ total_leaves_on_day += 0.5
+ elif attendance_on_day == "On Leave":
+ total_leaves_on_day += 1
+
+ absent.append(total_absent_on_day)
+ present.append(total_present_on_day)
+ leave.append(total_leaves_on_day)
+
+ return {
+ "data": {
+ "labels": labels,
+ "datasets": [
+ {"name": "Absent", "values": absent},
+ {"name": "Present", "values": present},
+ {"name": "Leave", "values": leave},
+ ],
+ },
+ "type": "line",
+ "colors": ["red", "green", "blue"],
+ }
diff --git a/erpnext/hr/report/monthly_attendance_sheet/test_monthly_attendance_sheet.py b/erpnext/hr/report/monthly_attendance_sheet/test_monthly_attendance_sheet.py
index 91da08e..2f3cb53 100644
--- a/erpnext/hr/report/monthly_attendance_sheet/test_monthly_attendance_sheet.py
+++ b/erpnext/hr/report/monthly_attendance_sheet/test_monthly_attendance_sheet.py
@@ -1,18 +1,24 @@
import frappe
from dateutil.relativedelta import relativedelta
from frappe.tests.utils import FrappeTestCase
-from frappe.utils import now_datetime
+from frappe.utils import get_year_ending, get_year_start, getdate, now_datetime
from erpnext.hr.doctype.attendance.attendance import mark_attendance
from erpnext.hr.doctype.employee.test_employee import make_employee
+from erpnext.hr.doctype.holiday_list.test_holiday_list import set_holiday_list
+from erpnext.hr.doctype.leave_application.test_leave_application import make_allocation_record
from erpnext.hr.report.monthly_attendance_sheet.monthly_attendance_sheet import execute
+from erpnext.payroll.doctype.salary_slip.test_salary_slip import make_leave_application
+
+test_dependencies = ["Shift Type"]
class TestMonthlyAttendanceSheet(FrappeTestCase):
def setUp(self):
- self.employee = make_employee("test_employee@example.com")
- frappe.db.delete("Attendance", {"employee": self.employee})
+ self.employee = make_employee("test_employee@example.com", company="_Test Company")
+ frappe.db.delete("Attendance")
+ @set_holiday_list("Salary Slip Test Holiday List", "_Test Company")
def test_monthly_attendance_sheet_report(self):
now = now_datetime()
previous_month = now.month - 1
@@ -33,14 +39,146 @@
}
)
report = execute(filters=filters)
- employees = report[1][0]
+
+ record = report[1][0]
datasets = report[3]["data"]["datasets"]
absent = datasets[0]["values"]
present = datasets[1]["values"]
leaves = datasets[2]["values"]
# ensure correct attendance is reflect on the report
- self.assertIn(self.employee, employees)
+ self.assertEqual(self.employee, record.get("employee"))
self.assertEqual(absent[0], 1)
self.assertEqual(present[1], 1)
self.assertEqual(leaves[2], 1)
+
+ @set_holiday_list("Salary Slip Test Holiday List", "_Test Company")
+ def test_monthly_attendance_sheet_with_detailed_view(self):
+ now = now_datetime()
+ previous_month = now.month - 1
+ previous_month_first = now.replace(day=1).replace(month=previous_month).date()
+
+ company = frappe.db.get_value("Employee", self.employee, "company")
+
+ # attendance with shift
+ mark_attendance(self.employee, previous_month_first, "Absent", "Day Shift")
+ mark_attendance(
+ self.employee, previous_month_first + relativedelta(days=1), "Present", "Day Shift"
+ )
+
+ # attendance without shift
+ mark_attendance(self.employee, previous_month_first + relativedelta(days=2), "On Leave")
+ mark_attendance(self.employee, previous_month_first + relativedelta(days=3), "Present")
+
+ filters = frappe._dict(
+ {
+ "month": previous_month,
+ "year": now.year,
+ "company": company,
+ }
+ )
+ report = execute(filters=filters)
+
+ day_shift_row = report[1][0]
+ row_without_shift = report[1][1]
+
+ self.assertEqual(day_shift_row["shift"], "Day Shift")
+ self.assertEqual(day_shift_row[1], "A") # absent on the 1st day of the month
+ self.assertEqual(day_shift_row[2], "P") # present on the 2nd day
+
+ self.assertEqual(row_without_shift["shift"], None)
+ self.assertEqual(row_without_shift[3], "L") # on leave on the 3rd day
+ self.assertEqual(row_without_shift[4], "P") # present on the 4th day
+
+ @set_holiday_list("Salary Slip Test Holiday List", "_Test Company")
+ def test_monthly_attendance_sheet_with_summarized_view(self):
+ now = now_datetime()
+ previous_month = now.month - 1
+ previous_month_first = now.replace(day=1).replace(month=previous_month).date()
+
+ company = frappe.db.get_value("Employee", self.employee, "company")
+
+ # attendance with shift
+ mark_attendance(self.employee, previous_month_first, "Absent", "Day Shift")
+ mark_attendance(
+ self.employee, previous_month_first + relativedelta(days=1), "Present", "Day Shift"
+ )
+
+ mark_attendance(
+ self.employee, previous_month_first + relativedelta(days=3), "Present"
+ ) # attendance without shift
+ mark_attendance(
+ self.employee, previous_month_first + relativedelta(days=4), "Present", late_entry=1
+ ) # late entry
+ mark_attendance(
+ self.employee, previous_month_first + relativedelta(days=5), "Present", early_exit=1
+ ) # early exit
+
+ leave_application = get_leave_application(self.employee)
+
+ filters = frappe._dict(
+ {"month": previous_month, "year": now.year, "company": company, "summarized_view": 1}
+ )
+ report = execute(filters=filters)
+
+ row = report[1][0]
+ self.assertEqual(row["employee"], self.employee)
+ self.assertEqual(row["total_present"], 4)
+ self.assertEqual(row["total_absent"], 1)
+ self.assertEqual(row["total_leaves"], leave_application.total_leave_days)
+
+ self.assertEqual(row["_test_leave_type"], leave_application.total_leave_days)
+ self.assertEqual(row["total_late_entries"], 1)
+ self.assertEqual(row["total_early_exits"], 1)
+
+ @set_holiday_list("Salary Slip Test Holiday List", "_Test Company")
+ def test_attendance_with_group_by_filter(self):
+ now = now_datetime()
+ previous_month = now.month - 1
+ previous_month_first = now.replace(day=1).replace(month=previous_month).date()
+
+ company = frappe.db.get_value("Employee", self.employee, "company")
+
+ # attendance with shift
+ mark_attendance(self.employee, previous_month_first, "Absent", "Day Shift")
+ mark_attendance(
+ self.employee, previous_month_first + relativedelta(days=1), "Present", "Day Shift"
+ )
+
+ # attendance without shift
+ mark_attendance(self.employee, previous_month_first + relativedelta(days=2), "On Leave")
+ mark_attendance(self.employee, previous_month_first + relativedelta(days=3), "Present")
+
+ filters = frappe._dict(
+ {"month": previous_month, "year": now.year, "company": company, "group_by": "Department"}
+ )
+ report = execute(filters=filters)
+
+ department = frappe.db.get_value("Employee", self.employee, "department")
+ department_row = report[1][0]
+ self.assertIn(department, department_row["department"])
+
+ day_shift_row = report[1][1]
+ row_without_shift = report[1][2]
+
+ self.assertEqual(day_shift_row["shift"], "Day Shift")
+ self.assertEqual(day_shift_row[1], "A") # absent on the 1st day of the month
+ self.assertEqual(day_shift_row[2], "P") # present on the 2nd day
+
+ self.assertEqual(row_without_shift["shift"], None)
+ self.assertEqual(row_without_shift[3], "L") # on leave on the 3rd day
+ self.assertEqual(row_without_shift[4], "P") # present on the 4th day
+
+
+def get_leave_application(employee):
+ now = now_datetime()
+ previous_month = now.month - 1
+
+ date = getdate()
+ year_start = getdate(get_year_start(date))
+ year_end = getdate(get_year_ending(date))
+ make_allocation_record(employee=employee, from_date=year_start, to_date=year_end)
+
+ from_date = now.replace(day=7).replace(month=previous_month).date()
+ to_date = now.replace(day=8).replace(month=previous_month).date()
+ return make_leave_application(employee, from_date, to_date, "_Test Leave Type")