import mysql.connector import datetime from franai.services.config import Config class RegisteredUser: """ Model for managing registered users in the database. This table stores metadata about the registration itself, while user_data.json stores HRIS details. """ @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 registered_users table if it doesn't already exist.""" conn = RegisteredUser.get_db_connection() if not conn: return try: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS registered_users ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, pid INT UNSIGNED NOT NULL, registered_at DATETIME NOT NULL, last_modified DATETIME NOT NULL, last_recognition_timestamp DATETIME NULL, self_register BOOLEAN DEFAULT FALSE, email_timestamp DATETIME NULL, UNIQUE KEY (pid) ) """) conn.commit() print("[OK] 'registered_users' table exists or was successfully created.") except mysql.connector.Error as e: print(f"Error creating 'registered_users' table: {e}") finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def add_or_update(pid, self_register=False): """ Adds a new registered user or updates an existing one. Returns True on success, False on failure. """ conn = RegisteredUser.get_db_connection() if not conn: return False now = Config.get_ph_time() try: cursor = conn.cursor() # Use INSERT ... ON DUPLICATE KEY UPDATE for an atomic operation query = """ INSERT INTO registered_users (pid, registered_at, last_modified, self_register) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE last_modified = VALUES(last_modified), self_register = VALUES(self_register) """ cursor.execute(query, (pid, now, now, self_register)) conn.commit() print(f"[OK] Successfully saved registration for PID {pid} to the database.") return True except mysql.connector.Error as e: print(f"Error saving registration for PID {pid}: {e}") conn.rollback() return False finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def delete(pid): """Deletes a registered user from the database by their PID.""" conn = RegisteredUser.get_db_connection() if not conn: return False try: cursor = conn.cursor() cursor.execute("DELETE FROM registered_users WHERE pid = %s", (pid,)) conn.commit() # It's okay if no rows were affected (user might not be in DB) print(f"[OK] Successfully deleted PID {pid} from the database (if they existed).") return True except mysql.connector.Error as e: print(f"Error deleting PID {pid} from database: {e}") conn.rollback() return False finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def update_email_timestamp(pid): """Updates the email_timestamp for a given user.""" conn = RegisteredUser.get_db_connection() if not conn: return False now = Config.get_ph_time() try: cursor = conn.cursor() query = "UPDATE registered_users SET email_timestamp = %s WHERE pid = %s" cursor.execute(query, (now, pid)) conn.commit() print(f"[OK] Updated email timestamp for PID {pid}.") return True except mysql.connector.Error as e: print(f"Error updating email timestamp for PID {pid}: {e}") conn.rollback() return False finally: if conn.is_connected(): cursor.close() conn.close() def get_all_with_hris_info(page=1, limit=10, search_term=None, self_register=None, exporting=False): """ Fetches registered users with HRIS info, supporting pagination and search. """ conn = RegisteredUser.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) # Base query base_query = """ FROM registered_users ru JOIN hris_core.employee_information_tbl eit ON ru.pid = eit.pid LEFT JOIN intra.tag_group tg ON ru.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 clause for search and self_register filter where_clauses = [] params = [] if search_term: where_clauses.append("(CONCAT(eit.first_name, ' ', eit.last_name) LIKE %s OR ru.pid LIKE %s)") params.extend([f"%{search_term}%", f"%{search_term}%"]) if self_register is not None and str(self_register) != '': if str(self_register).lower() in ['true', '1']: where_clauses.append("ru.self_register = %s") params.append(1) elif str(self_register).lower() in ['false', '0']: where_clauses.append("(ru.self_register = %s OR ru.self_register IS NULL)") params.append(0) where_clause = "" if where_clauses: where_clause = "WHERE " + " AND ".join(where_clauses) # Get total count for pagination # Note: Counting distinct PIDs in the JOINed result count_query = f"SELECT COUNT(DISTINCT ru.pid) as total {base_query} {where_clause}" cursor.execute(count_query, tuple(params)) total_records = cursor.fetchone()['total'] # Get paginated user data select_fields = """ SELECT ru.pid, ru.registered_at, ru.last_modified, ru.last_recognition_timestamp, ru.self_register, ru.email_timestamp, CONCAT(eit.first_name, ' ', eit.last_name) AS name, eit.company_email as email, MAX(ug.group_name) AS team_name """ group_by_clause = "GROUP BY ru.pid" order_clause = "ORDER BY ru.last_modified DESC" final_query = f"{select_fields} {base_query} {where_clause} {group_by_clause} {order_clause}" if not exporting: limit_clause = "LIMIT %s OFFSET %s" offset = (page - 1) * limit params.extend([limit, offset]) final_query = f"{final_query} {limit_clause}" cursor.execute(final_query, tuple(params)) users = cursor.fetchall() return {'users': users, 'total_records': total_records} except mysql.connector.Error as e: print(f"Error fetching registered users with HRIS info: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def get_all(): """Fetches all registered users from the database.""" conn = RegisteredUser.get_db_connection() if not conn: return None try: cursor = conn.cursor(dictionary=True) cursor.execute("SELECT pid, registered_at, last_modified, last_recognition_timestamp FROM registered_users ORDER BY last_modified DESC") users = cursor.fetchall() return users except mysql.connector.Error as e: print(f"Error fetching all registered users: {e}") return None finally: if conn.is_connected(): cursor.close() conn.close() @staticmethod def is_registered(pid): """Checks if a user is already registered by their PID.""" conn = RegisteredUser.get_db_connection() if not conn: # In case of DB error, assume not registered to be safe return False try: cursor = conn.cursor() cursor.execute("SELECT pid FROM registered_users WHERE pid = %s", (pid,)) result = cursor.fetchone() return result is not None except mysql.connector.Error as e: print(f"Error checking registration for PID {pid}: {e}") return False # Assume not registered on error finally: if conn.is_connected(): cursor.close() conn.close()