import mysql.connector import csv import os import datetime import calendar from franai.services.config import Config from franai.models.team_model import Team class AttendanceLog: """Model for managing attendance log data in the database.""" @staticmethod def get_db_connection(): """Establishes a database connection.""" try: conn = mysql.connector.connect( host=Config.DB_HOST, port=Config.DB_PORT, user=Config.DB_USER, password=Config.DB_PASSWORD, database=Config.DB_NAME ) return conn except mysql.connector.Error as e: print(f"Error connecting to MySQL Database: {e}") return None @staticmethod def create_table_if_not_exists(): """Creates the attendance_logs table if it doesn't exist.""" conn = AttendanceLog.get_db_connection() if not conn: return try: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS attendance_logs ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, pid INT UNSIGNED NOT NULL, recognition_timestamp DATETIME NOT NULL, status VARCHAR(50) NOT NULL, location_id INT UNSIGNED NULL, similarity FLOAT NULL, UNIQUE KEY unique_attendance (pid, recognition_timestamp), FOREIGN KEY (location_id) REFERENCES locations(id) ) """ ) conn.commit() print("[OK] 'attendance_logs' table exists or was successfully created.") except mysql.connector.Error as e: print(f"Error creating 'attendance_logs' table: {e}") finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def sync_from_csv_incremental(csv_path="attendance_log.csv", last_sync_time=None): """ Synchronizes records from the CSV incrementally based on the last sync time. Returns the timestamp of the latest processed record. """ if not os.path.exists(csv_path): return None conn = AttendanceLog.get_db_connection() if not conn: return None records_to_insert = [] latest_timestamp_in_batch = last_sync_time try: with open(csv_path, 'r', newline='') as file: reader = csv.DictReader(file) for row in reader: try: timestamp_str = row.get('Timestamp') if not timestamp_str: continue record_time = datetime.datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S") # If we have a last_sync_time, skip records that are not new if last_sync_time and record_time <= last_sync_time: continue records_to_insert.append(( int(row['User ID']), timestamp_str, row.get('Status', 'Check-in'), int(row['Location']) )) # Keep track of the latest timestamp in this batch if latest_timestamp_in_batch is None or record_time > latest_timestamp_in_batch: latest_timestamp_in_batch = record_time except (ValueError, TypeError): continue # Skip rows with bad data except (IOError, csv.Error) as e: print(f"Error reading CSV file: {e}") return None if not records_to_insert: print("Scheduler: No new attendance records to sync.") return latest_timestamp_in_batch try: cursor = conn.cursor() query = "INSERT IGNORE INTO attendance_logs (pid, recognition_timestamp, status) VALUES (%s, %s, %s)" cursor.executemany(query, records_to_insert) conn.commit() inserted_count = cursor.rowcount print(f"🔄 Incremental Sync: Processed {len(records_to_insert)} new records. Inserted {inserted_count} into DB.") return latest_timestamp_in_batch except mysql.connector.Error as e: print(f"Error during incremental bulk insert: {e}") conn.rollback() return last_sync_time # Return the old time on failure finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def sync_from_csv(csv_path="attendance_log.csv"): """ Synchronizes records from the attendance CSV file to the database. Uses 'INSERT IGNORE' to efficiently skip duplicates. """ if not os.path.exists(csv_path): print(f"Info: CSV file not found at {csv_path}. Skipping sync.") return 0, 0 conn = AttendanceLog.get_db_connection() if not conn: return 0, 0 records_to_insert = [] try: with open(csv_path, 'r', newline='') as file: reader = csv.DictReader(file) for row in reader: # Basic data validation if row.get('User ID') and row.get('Timestamp'): records_to_insert.append(( int(row['User ID']), row['Timestamp'], row.get('Status', 'Check-in') )) except (IOError, csv.Error) as e: print(f"Error reading CSV file: {e}") return 0, 0 if not records_to_insert: return 0, 0 inserted_count = 0 try: cursor = conn.cursor() # 'INSERT IGNORE' will skip inserting rows that would cause a duplicate entry # in a UNIQUE index (pid, recognition_timestamp). query = """ INSERT IGNORE INTO attendance_logs (pid, recognition_timestamp, status) VALUES (%s, %s, %s) """ cursor.executemany(query, records_to_insert) conn.commit() inserted_count = cursor.rowcount print(f"🔄 Attendance Sync: Processed {len(records_to_insert)} records. Inserted {inserted_count} new records.") return len(records_to_insert), inserted_count except mysql.connector.Error as e: print(f"Error fetching attendance status for user {user_id}: {e}") return attendance_info finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_status_for_user(user_id): """ Checks the database for a 'Check-in' record for a specific user for the current day. Returns the status and the check-in time. """ attendance_info = {'has_attendance': False, 'check_in_time': None} conn = AttendanceLog.get_db_connection() if not conn: return attendance_info try: cursor = conn.cursor(dictionary=True) query = """ SELECT recognition_timestamp FROM attendance_logs WHERE pid = %s AND status = 'Check-in' AND DATE(recognition_timestamp) = CURDATE() ORDER BY recognition_timestamp DESC LIMIT 1 """ cursor.execute(query, (user_id,)) result = cursor.fetchone() if result: attendance_info['has_attendance'] = True timestamp = result['recognition_timestamp'] # Format the time to a more readable h:i:s A format (e.g., 09:30:00 AM) attendance_info['check_in_time'] = timestamp.strftime('%I:%M:%S %p') return attendance_info except mysql.connector.Error as e: print(f"Error fetching attendance status for user {user_id}: {e}") return attendance_info finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def add_log(pid, status, timestamp, location_id=None, similarity=None): """ Adds a new attendance record to the database with a specific timestamp. Returns True on success, False on failure. """ conn = AttendanceLog.get_db_connection() if not conn: return False try: cursor = conn.cursor() query = """ INSERT INTO attendance_logs (pid, recognition_timestamp, status, location_id, similarity) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(query, (pid, timestamp, status, location_id, similarity)) conn.commit() return True except mysql.connector.Error as e: print(f"Error adding attendance log for PID {pid}: {e}") conn.rollback() return False finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def add_record(pid, location_id=None, similarity=None): """ Determines the attendance status (Check-in/Check-out) and adds a new record. Returns the status, a message, and the timestamp used. """ status_info = AttendanceLog.get_status_for_user(pid) if not status_info['has_attendance']: status = "Check-in" else: status = "Check-out" timestamp = Config.get_ph_time().replace(microsecond=0) if AttendanceLog.add_log(pid, status, timestamp, location_id, similarity): message = f"Staff {status.lower()} successfully" return status, message, timestamp else: return None, "Failed to log attendance", None @staticmethod def get_all_attendance_with_details(start_date=None, end_date=None, page=1, limit=10, exporting=False, search_term=None): """ Retrieves a list of attendance logs with user and location details, supporting date filtering and pagination. """ conn = AttendanceLog.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) params = [] query = """ SELECT al.id, al.pid, CONCAT(s.first_name, ' ', s.last_name) AS name, al.recognition_timestamp, al.status, loc.name AS location_name FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid LEFT JOIN locations loc ON al.location_id = loc.id """ where_clauses = [] if start_date: where_clauses.append("al.recognition_timestamp >= %s") params.append(f"{start_date} 00:00:00") if end_date: where_clauses.append("al.recognition_timestamp <= %s") params.append(f"{end_date} 23:59:59") if search_term: where_clauses.append("CONCAT(s.first_name, ' ', s.last_name) LIKE %s") params.append(f"%{search_term}%") if where_clauses: query += " WHERE " + " AND ".join(where_clauses) query += " ORDER BY al.recognition_timestamp DESC" if not exporting: offset = (page - 1) * limit query += " LIMIT %s OFFSET %s" params.extend([limit, offset]) cursor.execute(query, tuple(params)) results = cursor.fetchall() return results except mysql.connector.Error as e: print(f"Error fetching all attendance data: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_total_logs_count(start_date=None, end_date=None, search_term=None): """ Counts the total number of raw attendance logs, supporting date filtering and search. """ conn = AttendanceLog.get_db_connection() if not conn: return 0 try: cursor = conn.cursor() params = [] query = "SELECT COUNT(*) FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid" where_clauses = [] if start_date: where_clauses.append("al.recognition_timestamp >= %s") params.append(f"{start_date} 00:00:00") if end_date: where_clauses.append("al.recognition_timestamp <= %s") params.append(f"{end_date} 23:59:59") if search_term: where_clauses.append("CONCAT(s.first_name, ' ', s.last_name) LIKE %s") params.append(f"%{search_term}%") if where_clauses: query += " WHERE " + " AND ".join(where_clauses) cursor.execute(query, tuple(params)) count = cursor.fetchone()[0] return count finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_attendance_summary(start_date=None, end_date=None, view='summary', search_term=None): """ Calculates summary statistics. The unique_users count is always the total number of registered users, regardless of filters. """ conn = AttendanceLog.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) # 1. Get the total number of unique registered users (unfiltered) cursor.execute("SELECT COUNT(pid) as total_users FROM registered_users") total_users_result = cursor.fetchone() total_unique_users = total_users_result['total_users'] if total_users_result else 0 # 2. Get the filtered check-in and check-out counts params = [] from_join = "FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid" if view == 'logs': query = f""" SELECT COALESCE(SUM(CASE WHEN status = 'Check-in' THEN 1 ELSE 0 END), 0) AS checkin_count, COALESCE(SUM(CASE WHEN status = 'Check-out' THEN 1 ELSE 0 END), 0) AS checkout_count {from_join} """ else: # 'summary' view query = f""" SELECT COUNT(DISTINCT CONCAT(al.pid, DATE(al.recognition_timestamp))) AS checkin_count, COUNT(DISTINCT CASE WHEN al.status = 'Check-out' THEN CONCAT(al.pid, DATE(al.recognition_timestamp)) END) AS checkout_count {from_join} """ where_clauses = [] if start_date: where_clauses.append("al.recognition_timestamp >= %s") params.append(f"{start_date} 00:00:00") if end_date: where_clauses.append("al.recognition_timestamp <= %s") params.append(f"{end_date} 23:59:59") if search_term: where_clauses.append("CONCAT(s.first_name, ' ', s.last_name) LIKE %s") params.append(f"%{search_term}%") if where_clauses: query += " WHERE " + " AND ".join(where_clauses) cursor.execute(query, tuple(params)) summary = cursor.fetchone() # 3. Combine the results if summary: summary['unique_users'] = total_unique_users else: # In case the filtered query returns nothing summary = {'checkin_count': 0, 'checkout_count': 0, 'unique_users': total_unique_users} return summary except mysql.connector.Error as e: print(f"Error getting attendance summary: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_daily_attendance(start_date=None, end_date=None, page=1, limit=15, exporting=False, search_term=None, team_gid=None): """ Retrieves a paginated or full, summarized list of daily attendance for each user, showing the location of their first check-in of the day. """ conn = AttendanceLog.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) params = [] query = """ WITH FirstCheckIn AS ( SELECT pid, DATE(recognition_timestamp) AS attendance_date, location_id, ROW_NUMBER() OVER(PARTITION BY pid, DATE(recognition_timestamp) ORDER BY recognition_timestamp ASC) as rn FROM attendance_logs WHERE status = 'Check-in' ) SELECT DATE(al.recognition_timestamp) AS attendance_date, al.pid, CONCAT(s.first_name, ' ', s.last_name) AS name, s.company_email AS email, ug.group_name AS team_name, MIN(CASE WHEN al.status = 'Check-in' THEN al.recognition_timestamp END) AS check_in, MAX(CASE WHEN al.status = 'Check-out' THEN al.recognition_timestamp END) AS check_out, MAX(loc.name) AS location_name FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid LEFT JOIN intra.tag_group tg ON s.pid = tg.user_id AND tg.workspace_default = 'Y' LEFT JOIN intra.user_group ug ON tg.group_id = ug.GID AND ug.workspace_status = 1 LEFT JOIN FirstCheckIn fci ON al.pid = fci.pid AND DATE(al.recognition_timestamp) = fci.attendance_date AND fci.rn = 1 LEFT JOIN locations loc ON fci.location_id = loc.id """ where_clauses = [] if start_date: where_clauses.append("al.recognition_timestamp >= %s") params.append(f"{start_date} 00:00:00") if end_date: where_clauses.append("al.recognition_timestamp <= %s") params.append(f"{end_date} 23:59:59") if search_term: where_clauses.append("CONCAT(s.first_name, ' ', s.last_name) LIKE %s") params.append(f"%{search_term}%") if team_gid: where_clauses.append("ug.GID = %s") params.append(team_gid) if where_clauses: query += " WHERE " + " AND ".join(where_clauses) query += " GROUP BY al.pid, DATE(al.recognition_timestamp), name, email, team_name" query += " ORDER BY attendance_date DESC, check_in DESC" # Only apply pagination if not exporting if not exporting: offset = (page - 1) * limit query += " LIMIT %s OFFSET %s" params.extend([limit, offset]) cursor.execute(query, tuple(params)) return cursor.fetchall() except mysql.connector.Error as e: print(f"Error fetching daily attendance: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_total_daily_count(start_date=None, end_date=None, search_term=None): """Counts the total number of daily summary records.""" conn = AttendanceLog.get_db_connection() if not conn: return 0 try: cursor = conn.cursor() params = [] # We need to join with the staff table to search by name query = "SELECT COUNT(DISTINCT al.pid, DATE(al.recognition_timestamp)) FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid" where_clauses = [] if start_date: where_clauses.append("al.recognition_timestamp >= %s") params.append(f"{start_date} 00:00:00") if end_date: where_clauses.append("al.recognition_timestamp <= %s") params.append(f"{end_date} 23:59:59") if search_term: where_clauses.append("CONCAT(s.first_name, ' ', s.last_name) LIKE %s") params.append(f"%{search_term}%") if where_clauses: query += " WHERE " + " AND ".join(where_clauses) cursor.execute(query, tuple(params)) return cursor.fetchone()[0] except mysql.connector.Error as e: print(f"Error counting daily attendance: {e}") return 0 finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_last_attendance_for_user(pid): """ Retrieves the timestamp of the most recent attendance record for a given user. """ conn = AttendanceLog.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) query = """ SELECT recognition_timestamp FROM attendance_logs WHERE pid = %s ORDER BY recognition_timestamp DESC LIMIT 1 """ cursor.execute(query, (pid,)) result = cursor.fetchone() return result['recognition_timestamp'] if result else None except mysql.connector.Error as e: print(f"Error fetching last attendance for user {pid}: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_all_users_recognition_scores(): """ Retrieves the min, max, and average recognition scores for all users. """ conn = AttendanceLog.get_db_connection() if not conn: return {} scores = {} try: cursor = conn.cursor(dictionary=True) query = """ SELECT pid, MIN(similarity) as min_score, MAX(similarity) as max_score, AVG(similarity) as avg_score FROM attendance_logs WHERE similarity IS NOT NULL GROUP BY pid """ cursor.execute(query) results = cursor.fetchall() for row in results: scores[row['pid']] = { 'min_score': row['min_score'], 'max_score': row['max_score'], 'avg_score': row['avg_score'] } return scores except mysql.connector.Error as e: print(f"Error fetching all users recognition scores: {e}") return {} finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_user_recognition_scores_since(pid, start_date): """ Retrieves the min, max, and average recognition scores for a single user since a specific start date. """ conn = AttendanceLog.get_db_connection() if not conn or not start_date: return {} try: cursor = conn.cursor(dictionary=True) query = """ SELECT MIN(similarity) as min_score, MAX(similarity) as max_score, AVG(similarity) as avg_score FROM attendance_logs WHERE pid = %s AND recognition_timestamp >= %s AND similarity IS NOT NULL """ cursor.execute(query, (pid, start_date)) scores = cursor.fetchone() return scores or {} except mysql.connector.Error as e: print(f"Error fetching recognition scores for PID {pid} since {start_date}: {e}") return {} finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_quick_view_summary(start_date=None, end_date=None): """ Calculates the quick view summary. - Unique FTE is the average number of unique employees who check in per week. - Required Visits is based on the total number of hired employees. - Assumes a static 'Hybrid 1' (once per week) schedule. """ conn = AttendanceLog.get_db_connection() if not conn: return None summary_data = {'total_visits': 0, 'unique_fte': 0, 'required_visits': 0, 'performance': 0} try: from collections import defaultdict cursor = conn.cursor(dictionary=True) # Query 1: Get all hired employees to calculate requirements individually. hired_employees_query = "SELECT pid FROM hris_core.employee_information_tbl WHERE status = 'Hired'" cursor.execute(hired_employees_query) hired_employees = cursor.fetchall() print(f"DEBUG: Found {len(hired_employees)} hired employees from HRIS.") # Query 2: Get actual attendance data for the period. actuals_query = """ SELECT DISTINCT al.pid, DATE(al.recognition_timestamp) AS attendance_date FROM attendance_logs al WHERE al.recognition_timestamp >= %s AND al.recognition_timestamp <= %s AND al.status = 'Check-in' """ params = (f"{start_date} 00:00:00", f"{end_date} 23:59:59") cursor.execute(actuals_query, params) check_in_records = cursor.fetchall() summary_data['total_visits'] = len(check_in_records) # Calculate unique FTE by summing up unique check-ins per week if check_in_records: weekly_ftes = defaultdict(set) for record in check_in_records: week_number = record['attendance_date'].isocalendar().week weekly_ftes[week_number].add(record['pid']) if weekly_ftes: total_weekly_unique_fte = sum(len(pids) for pids in weekly_ftes.values()) summary_data['unique_fte'] = total_weekly_unique_fte print(f"DEBUG: Total Visits (unique daily check-ins): {summary_data['total_visits']}") print(f"DEBUG: Sum of unique FTEs per week: {summary_data['unique_fte']}") # Calculate required visits by applying the hybrid logic to each hired employee if start_date and end_date: try: start = datetime.datetime.strptime(start_date, "%Y-%m-%d") end = datetime.datetime.strptime(end_date, "%Y-%m-%d") # Calculate number of weeks (by Mondays) and months for the period num_weeks = 0 current_day = start while current_day <= end: if current_day.weekday() == 0: # 0 is Monday num_weeks += 1 current_day += datetime.timedelta(days=1) num_months = (end.year - start.year) * 12 + (end.month - start.month) + 1 # Explicitly apply the logic for each employee total_required_visits = 0 for emp in hired_employees: # Statically assign the hybrid setup for now setup = 'Hybrid 1' if setup == 'Hybrid 1': total_required_visits += num_weeks # Once a week elif setup == 'Hybrid 2': total_required_visits += (2 * num_months) # Twice a month elif setup in ['Hybrid 3', 'Hybrid 4']: total_required_visits += num_months # Once a month summary_data['required_visits'] = total_required_visits except (ValueError, TypeError): summary_data['required_visits'] = 0 # Calculate performance if summary_data['required_visits'] > 0: summary_data['performance'] = round((summary_data['total_visits'] / summary_data['required_visits']) * 100) print(f"DEBUG: Final summary to be returned: {summary_data}") return summary_data except mysql.connector.Error as e: print(f"Error getting quick view summary: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_recent_attendance_logs(limit=5): """ Retrieves the most recent, unique attendance logs by user and status. This prevents showing multiple check-outs for the same person in the recent list. """ conn = AttendanceLog.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) query = """ WITH RankedLogs AS ( SELECT al.pid, CONCAT(s.first_name, ' ', s.last_name) AS name, al.recognition_timestamp, al.status, ROW_NUMBER() OVER(PARTITION BY al.pid, al.status ORDER BY al.recognition_timestamp DESC) as rn FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid ) SELECT pid, name, recognition_timestamp, status FROM RankedLogs WHERE rn = 1 ORDER BY recognition_timestamp DESC LIMIT %s """ cursor.execute(query, (limit,)) results = cursor.fetchall() return results except mysql.connector.Error as e: print(f"Error fetching recent attendance logs: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_weekly_summary(start_date=None, end_date=None, team_gid=None): """ Calculates a week-by-week summary of attendance metrics, filterable by team. This function now calls the team attendance summary and aggregates the results to ensure consistent calculations. """ # Get the detailed team-by-team breakdown team_summary = AttendanceLog.get_team_attendance_summary( start_date=start_date, end_date=end_date, team_gid=team_gid ) if team_summary is None: return None # Aggregate the results from the team summary total_visits = 0 unique_fte = 0 actual = 0 required = 0 if not team_summary: required = 0 else: for team_data in team_summary: total_visits += team_data.get('totalVisits', 0) unique_fte += team_data.get('uniqueFTE', 0) actual += team_data.get('actual', 0) required += team_data.get('required', 0) start_dt = datetime.datetime.strptime(start_date, "%Y-%m-%d") aggregated_summary = [{ 'week_identifier': start_dt.strftime('%Y%U'), 'totalVisits': total_visits, 'uniqueFTE': unique_fte, 'actual': actual, 'required': required }] return aggregated_summary @staticmethod def get_team_attendance_summary(start_date=None, end_date=None, team_gid=None, prev_start_date=None, prev_end_date=None): """ Calculates a team-by-team summary of attendance, including comparison to a previous period, with dynamic carry-over logic for required attendance. """ conn = AttendanceLog.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) def _get_data_for_period(start, end, gid): """Helper function to fetch and process data for a single period.""" if not start or not end: return {} # 1. Get actual attendance for the specified period (e.g., one week) actuals_query = """ SELECT COALESCE(ug.group_name, 'Unassigned') AS team_name, COUNT(al.id) AS total_visits_raw, COUNT(DISTINCT al.pid, DATE(al.recognition_timestamp)) AS actual_unique_days, COUNT(DISTINCT al.pid) AS unique_fte FROM attendance_logs al LEFT JOIN hris_core.employee_information_tbl s ON al.pid = s.pid LEFT JOIN intra.tag_group tg ON s.pid = tg.user_id AND tg.workspace_default = 'Y' LEFT JOIN intra.user_group ug ON tg.group_id = ug.GID AND ug.workspace_status = 1 WHERE al.status = 'Check-in' AND al.recognition_timestamp >= %s AND al.recognition_timestamp <= %s """ params = [f"{start} 00:00:00", f"{end} 23:59:59"] if gid is not None and gid != '': actuals_query += " AND ug.GID = %s" params.append(gid) actuals_query += " GROUP BY team_name;" cursor.execute(actuals_query, tuple(params)) actuals_by_team = {team['team_name']: team for team in cursor.fetchall()} # 2. Calculate required attendance with "make-up" logic required_map = {} try: start_dt = datetime.datetime.strptime(start, "%Y-%m-%d") year, month = start_dt.year, start_dt.month month_start_date = start_dt.replace(day=1) month_end_day = calendar.monthrange(year, month)[1] month_end_date = month_start_date.replace(day=month_end_day) def get_week_of_month(date): # Use ISO week numbers for a consistent week-of-the-month calculation. first_day_of_month = date.replace(day=1) # Get the ISO week number for the first day of the month and the given date. # isocalendar() returns a tuple (year, week, weekday) first_day_iso_week = first_day_of_month.isocalendar()[1] given_date_iso_week = date.isocalendar()[1] # Handle the edge case where a month starts in a week from the previous year (e.g., Jan 1st). if given_date_iso_week < first_day_iso_week: # This can happen if the first day of the month is in week 52 or 53 # and the given date is in week 1 of the new year. # A simple way to handle this is to count weeks from the start of the year. return given_date_iso_week return (given_date_iso_week - first_day_iso_week) + 1 current_week_of_month = get_week_of_month(start_dt) # Fetch all hired employees active during the reporting period required_query = """ SELECT s.pid, s.hybrid_level, COALESCE(ug.group_name, 'Unassigned') AS team_name, r.last_day FROM hris_core.employee_information_tbl s LEFT JOIN intra.tag_group tg ON s.pid = tg.user_id AND tg.workspace_default = 'Y' LEFT JOIN intra.user_group ug ON tg.group_id = ug.GID AND ug.workspace_status = 1 LEFT JOIN hris_core.resignation_tbl r ON s.pid = r.pid AND r.status = 'Resigned' WHERE s.status = 'Hired' AND STR_TO_DATE(s.date_joined, '%m/%d/%Y') <= %s """ params_required = [end] # Use the end date of the reporting period if gid is not None and gid != '': required_query += " AND ug.GID = %s" params_required.append(gid) cursor.execute(required_query, tuple(params_required)) hired_employees = cursor.fetchall() # Fetch all monthly check-ins for these employees pids = [emp['pid'] for emp in hired_employees] monthly_logs = {} if pids: logs_query = "SELECT pid, DATE(recognition_timestamp) as attendance_date FROM attendance_logs WHERE status = 'Check-in' AND pid IN ({}) AND recognition_timestamp BETWEEN %s AND %s".format(','.join(map(str, pids))) params_logs = [month_start_date.strftime('%Y-%m-%d 00:00:00'), month_end_date.strftime('%Y-%m-%d 23:59:59')] cursor.execute(logs_query, tuple(params_logs)) for row in cursor.fetchall(): pid = row['pid'] if pid not in monthly_logs: monthly_logs[pid] = set() monthly_logs[pid].add(get_week_of_month(row['attendance_date'])) # Calculate baseline and adjustments for each team team_baselines = {} team_adjustments = {} for emp in hired_employees: team_name = emp.get('team_name', 'Unassigned') team_baselines.setdefault(team_name, {1:0, 2:0, 3:0, 4:0, 5:0}) team_adjustments.setdefault(team_name, {1:0, 2:0, 3:0, 4:0, 5:0}) pid = emp['pid'] setup = emp.get('hybrid_level') last_day_str = emp.get('last_day') # Default to 'Hybrid Group 1' if hybrid_level is not explicitly set or is empty if not setup: setup = 'Hybrid Group 1' attended_weeks = monthly_logs.get(pid, set()) required_weeks = [] if setup == 'Hybrid Group 1': required_weeks = [1, 2, 3, 4, 5] elif setup == 'Hybrid Group 2': required_weeks = [1, 3] elif setup in ['Hybrid Group 3', 'Hybrid Group 3-4']: required_weeks = [1] elif setup == 'Hybrid Group 4' and month in [1, 4, 7, 10]: required_weeks = [1] else: # If setup is a string but not recognized, default to full attendance as before required_weeks = [1, 2, 3, 4, 5] # Determine the last week of employment if resigned this month last_week_of_employment = 5 # Default to all weeks if last_day_str: try: last_day_date = datetime.datetime.strptime(last_day_str, '%m/%d/%Y').date() if last_day_date.year == year and last_day_date.month == month: last_week_of_employment = get_week_of_month(last_day_date) except ValueError: pass # Ignore invalid date formats for week_num in required_weeks: if week_num <= get_week_of_month(month_end_date) and week_num <= last_week_of_employment: team_baselines[team_name][week_num] += 1 missed_target_weeks = sorted(list(set(required_weeks) - attended_weeks)) makeup_weeks = sorted(list(attended_weeks - set(required_weeks))) for missed_week in missed_target_weeks: for makeup_week in makeup_weeks: if makeup_week > missed_week: team_adjustments[team_name][missed_week] -= 1 team_adjustments[team_name][makeup_week] += 1 makeup_weeks.remove(makeup_week) break # Finalize required map for the current week for team_name in team_baselines: required_map[team_name] = team_baselines[team_name][current_week_of_month] + team_adjustments[team_name][current_week_of_month] except Exception as e: print(f"Warning: Could not calculate hybrid requirements for team summary. Error: {e}") # 3. Get a definitive list of all teams all_teams_from_model = Team.get_all_teams() all_team_names = {team['group_name'] for team in all_teams_from_model} if gid: filtered_team = next((t for t in all_teams_from_model if str(t.get('team_gid')) == str(gid)), None) if filtered_team: all_team_names = {filtered_team['group_name']} if 'Unassigned' in actuals_by_team or 'Unassigned' in required_map: all_team_names.add('Unassigned') # 4. Merge data period_data = {} for team_name in sorted(list(all_team_names)): actuals = actuals_by_team.get(team_name, {}) required = required_map.get(team_name, 0) period_data[team_name] = { 'totalVisits': actuals.get('total_visits_raw', 0), 'uniqueFTE': actuals.get('unique_fte', 0), 'required': required, 'actual': actuals.get('actual_unique_days', 0), 'performance': round((actuals.get('actual_unique_days', 0) / required) * 100) if required > 0 else 0, } return period_data # Fetch data for both periods current_data = _get_data_for_period(start_date, end_date, team_gid) prev_data = _get_data_for_period(prev_start_date, prev_end_date, team_gid) # Combine the results all_team_names = sorted(list(set(current_data.keys()) | set(prev_data.keys()))) formatted_summary = [] for team_name in all_team_names: current = current_data.get(team_name, {}) prev = prev_data.get(team_name, {}) actual = current.get('actual', 0) required = current.get('required', 0) formatted_summary.append({ 'name': team_name, 'totalVisits': current.get('totalVisits', 0), 'uniqueFTE': current.get('uniqueFTE', 0), 'required': required, 'actual': actual, 'performance': current.get('performance', 0), 'isDanger': actual < required, 'prevTotalVisits': prev.get('totalVisits', 0), 'prevUniqueFTE': prev.get('uniqueFTE', 0), 'prevRequired': prev.get('required', 0), 'prevActual': prev.get('actual', 0), 'prevPerformance': prev.get('performance', 0), }) return formatted_summary except mysql.connector.Error as e: print(f"Error getting team attendance summary: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close()