from flask import Blueprint, render_template, request, redirect, url_for, flash, current_app, make_response from flask_login import login_required, current_user, login_user from werkzeug.utils import secure_filename import os from datetime import datetime, timedelta, date, time from extensions import db import bcrypt from models import Employee, Department, Designation, EmployeeChangeLog, PublicLink, PendingEmployee, PendingApprovalLog, BroadcastMessage, Role, Holiday,LeaveRequest,BiometricPunch from sqlalchemy import extract,func from sqlalchemy.exc import IntegrityError from common_utils import employee_required from typing import Optional from routes.biometric import get_biometric_day_summary import requests from flask import current_app import re read_broadcasts = db.relationship('BroadcastRead', backref='employee', lazy='dynamic') employee_bp = Blueprint('employee', __name__, url_prefix='/employee') from sqlalchemy import func, literal # Reusable filter for employee visibility ALLOWED_EMPLOYEE_STATUSES = ('active', 'notice period', 'notice_period') def _extract_emp_prefix(code: Optional[str]) -> Optional[str]: """ Extract an 'institute' prefix from an employee_code. Example: 'CHL00123' -> 'CHL' Falls back to the full code if no alphabetic prefix. """ if not code: return None code = code.strip() if not code: return None prefix_chars = [] for ch in code: if ch.isalpha(): prefix_chars.append(ch) else: break prefix = "".join(prefix_chars).upper() return prefix or code def status_allowed_filter(model_cls=None): """ Returns a SQLAlchemy filter that matches employees whose status is in ['active', 'notice period', 'notice_period'] (case-insensitive). """ col = (model_cls.status if model_cls is not None else Employee.status) return func.lower(func.coalesce(col, literal(''))).in_(ALLOWED_EMPLOYEE_STATUSES) def _approx_location_from_latlng(lat, lng): """Return a short human-readable location from lat/lng (or None).""" if lat is None or lng is None: return None try: resp = requests.get( "https://nominatim.openstreetmap.org/reverse", params={"lat": lat, "lon": lng, "format": "json", "zoom": 14}, headers={"User-Agent": "hrsoft-self-punch/1.0"}, timeout=5, ) resp.raise_for_status() data = resp.json() addr = data.get("address", {}) or {} parts = [ addr.get("suburb") or addr.get("neighbourhood"), addr.get("city") or addr.get("town") or addr.get("village"), addr.get("state"), addr.get("country"), ] nice = ", ".join([p for p in parts if p]) return nice or data.get("display_name") except Exception: current_app.logger.exception("Reverse geocode failed") return None @employee_bp.route('/add', methods=['GET', 'POST']) @login_required def add_employee(): if request.method == 'POST': full_name = request.form.get('full_name') email = (request.form.get('email') or "").strip() phone = request.form.get('phone') employee_code = request.form.get('employee_code') reporting_manager_id = request.form.get('reporting_manager_id') department_id = request.form.get('department_id') designation_id = request.form.get('designation_id') doj = request.form.get('doj') uan = request.form.get('uan') qualification = request.form.get('qualification') institute = request.form.get('institute') work_mode = request.form.get('work_mode') shift_start = request.form.get('shift_start') if work_mode in ['WFO', 'Hybrid'] else None shift_end = request.form.get('shift_end') if work_mode in ['WFO', 'Hybrid'] else None desktime_id = request.form.get('desktime_id') resume = request.files.get('resume') id_proof = request.files.get('id_proof') uploads_dir = os.path.join('static', 'uploads') os.makedirs(uploads_dir, exist_ok=True) resume_path = None id_proof_path = None if resume and resume.filename: resume_filename = secure_filename(resume.filename) resume_path = os.path.join(uploads_dir, resume_filename) resume.save(resume_path) if id_proof and id_proof.filename: id_proof_filename = secure_filename(id_proof.filename) id_proof_path = os.path.join(uploads_dir, id_proof_filename) id_proof.save(id_proof_path) # 🔐 Email reuse allowed only for ACTIVE / NOTICE employees if email: email_conflict = Employee.query.filter( func.lower(Employee.email) == email.lower(), func.lower(func.coalesce(Employee.status, "")). in_(["active", "notice_period", "notice period"]) ).first() if email_conflict: flash("Email already used by an active employee.", "danger") return redirect(url_for('employee.add_employee')) new_employee = Employee( full_name=full_name, email=email or None, # ✅ store email phone=phone, employee_code=employee_code, reporting_manager_id=reporting_manager_id or None, department_id=department_id, designation_id=designation_id, date_of_joining=doj, uan=uan, qualification=qualification, institute=institute, resume_path=resume_path, id_proof_path=id_proof_path, work_mode=work_mode, shift_start=shift_start, shift_end=shift_end, desktime_id=desktime_id, created_by=current_user.id, updated_by=current_user.id, created_on=datetime.utcnow() ) db.session.add(new_employee) try: db.session.commit() flash('Employee added successfully.', 'success') return redirect(url_for('employee.view_employees')) except IntegrityError: db.session.rollback() flash('Employee Code already exists. Please use a unique code.', 'danger') return redirect(url_for('employee.add_employee')) departments = Department.query.all() designations = Designation.query.all() managers = Employee.query.all() return render_template('employee/add_employee.html', departments=departments, designations=designations, managers=managers, employee=None) from datetime import datetime def parse_optional_date(value): try: return datetime.strptime(value, '%Y-%m-%d').date() if value else None except (ValueError, TypeError): return None from flask import render_template, request, redirect, url_for, flash from flask_login import login_required from datetime import datetime import os from werkzeug.utils import secure_filename from models import Employee, Department, Designation @employee_bp.route('/edit/', methods=['GET', 'POST']) @login_required def edit_employee(employee_id): employee = Employee.query.get_or_404(employee_id) departments = Department.query.all() designations = Designation.query.all() managers = Employee.query.filter_by(status='active').all() # ⬇️ NEW: read-only flag for resigned profiles is_readonly = (employee.status == "resigned") if request.method == 'POST': # ⬇️ NEW: block edits for resigned employees if is_readonly: flash("This employee has completed exit. Profile is read-only.", "warning") return redirect(request.url) f = request.form errors = {} def val(name): return (f.get(name) or "").strip() def int_or_none(name): v = val(name) return int(v) if v.isdigit() else None # ---- REQUIRED FIELDS (adjust to your policy) ---- required = { "full_name": "Full name is required.", "email": "Work email is required.", "employee_code": "Employee code is required.", "department_id": "Please select a department.", "designation_id": "Please select a designation.", "reporting_manager_id": "Please select a reporting manager.", } for field, msg in required.items(): if not val(field): errors[field] = msg # ---- Email format ---- email = val("email") if email and "@" not in email: errors["email"] = "Please enter a valid email address." # ---- Uniqueness checks ---- if email and "email" not in errors: existing_email_user = Employee.query.filter( func.lower(Employee.email) == email.lower(), Employee.id != employee.id, func.lower(func.coalesce(Employee.status, "")). in_(["active", "notice_period", "notice period"]) ).first() if existing_email_user: errors["email"] = "Email already exists for an active employee." desktime_id = val("desktime_id") if desktime_id: existing_desktime_user = Employee.query.filter( Employee.desktime_id == desktime_id, Employee.id != employee.id ).first() if existing_desktime_user: errors["desktime_id"] = "Desktime ID already exists for another employee." if errors: flash(f"Please fix {len(errors)} field(s) highlighted below.", "danger") return render_template( "employee/edit_employee.html", employee=employee, departments=departments, designations=designations, managers=managers, errors=errors, form_values=f.to_dict(flat=True), is_readonly=is_readonly, # ⬅️ pass through ), 400 # ---- NO ERRORS: proceed to save ---- uploads_dir = os.path.join("static", "uploads") os.makedirs(uploads_dir, exist_ok=True) def save_file(field_name, current_path): file = request.files.get(field_name) if file and file.filename: filename = secure_filename(file.filename) full_path = os.path.join(uploads_dir, filename) file.save(full_path) return os.path.join("uploads", filename).replace("\\", "/") return current_path # Text fields employee.full_name = val("full_name") employee.email = email or None employee.phone = val("phone") employee.employee_code = val("employee_code") or employee.employee_code employee.department_id = int_or_none("department_id") employee.designation_id = int_or_none("designation_id") employee.reporting_manager_id = int_or_none("reporting_manager_id") employee.location = val("location") employee.work_mode = val("work_mode") employee.shift_start = val("shift_start") or None employee.shift_end = val("shift_end") or None employee.qualification = val("qualification") employee.institute = val("institute") employee.employment_type = (request.form.get("employment_type") or "payroll").lower() # Dates employee.date_of_joining = parse_optional_date(val("date_of_joining")) employee.dob = parse_optional_date(val("dob")) employee.last_company_join_date = parse_optional_date(val("last_company_doj")) employee.last_company_exit_date = parse_optional_date(val("last_company_doe")) # Personal employee.gender = val("gender") employee.father_spouse_name = val("father_spouse_name") employee.marital_status = val("marital_status") employee.address = val("address") employee.temporary_address = val("temporary_address") employee.personal_number = val("personal_number") employee.emergency_contact_name = val("emergency_contact_name") employee.emergency_contact_number = val("emergency_contact_number") employee.aadhar_name = val("aadhar_name") employee.aadhar_number = val("aadhar_number") employee.pan_name = val("pan_name") employee.pan_number = val("pan_number") employee.personal_email = val("personal_email") # Banking employee.bank_account_name = val("bank_account_name") employee.bank_account_number = val("bank_account_number") employee.ifsc_code = val("ifsc_code") employee.bank_name = val("bank_name") employee.uan = val("uan") employee.last_company_pf_number = val("last_company_pf_number") # Education employee.secondary_school_name = val("secondary_name") employee.secondary_school_marks = val("secondary_marks") employee.secondary_school_year = val("secondary_year") employee.higher_secondary_school_name = val("higher_secondary_name") employee.higher_secondary_school_marks = val("higher_secondary_marks") employee.higher_secondary_school_year = val("higher_secondary_year") employee.graduation_name = val("graduation_name") employee.graduation_marks = val("graduation_marks") employee.graduation_year = val("graduation_year") employee.post_graduation_name = val("post_graduation_name") employee.post_graduation_marks = val("post_graduation_marks") employee.post_graduation_year = val("post_graduation_year") # Professional employee.last_company_name = val("last_company_name") employee.last_company_code = val("last_company_code") employee.last_company_designation = val("last_company_designation") # Desktime employee.desktime_id = desktime_id or None # Files employee.resume_path = save_file("resume", employee.resume_path) employee.id_proof_path = save_file("id_proof", employee.id_proof_path) employee.doc_qualification = save_file("doc_qualification", employee.doc_qualification) employee.doc_employment_proof = save_file("doc_employment_proof", employee.doc_employment_proof) employee.doc_photographs = save_file("doc_photographs", employee.doc_photographs) employee.doc_pan_card = save_file("doc_pan_card", employee.doc_pan_card) employee.doc_aadhar_card = save_file("doc_aadhar_card", employee.doc_aadhar_card) employee.doc_parents_aadhar = save_file("doc_parents_aadhar", employee.doc_parents_aadhar) employee.doc_uan_card = save_file("doc_uan_card", employee.doc_uan_card) employee.doc_cancelled_cheque = save_file("doc_cancelled_cheque", employee.doc_cancelled_cheque) # Audit employee.updated_on = datetime.utcnow() employee.updated_by = current_user.id db.session.commit() flash("✅ Employee details updated successfully.", "success") return redirect(url_for("employee.view_employees")) # GET return render_template( "employee/edit_employee.html", employee=employee, departments=departments, designations=designations, managers=managers, errors={}, # none initially form_values={}, # use model values in template is_readonly=is_readonly, # ⬅️ NEW ) @employee_bp.route('/view') @login_required def view_employees(): from sqlalchemy import asc, desc from sqlalchemy.orm import joinedload # --- Inputs --- name = request.args.get('name', '').strip() code = request.args.get('code', '').strip() dept_id = request.args.get('department_id', type=int) role_name = request.args.get('role', '').strip() status = request.args.get('status', '').strip().lower() # active/notice/resigned institute = request.args.get('institute', '').strip() # emp-code prefix location = request.args.get('location', '').strip() # workplace designation_id = request.args.get('designation_id', type=int) page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 20, type=int) sort = request.args.get('sort', 'name') # whitelist key direction = request.args.get('dir', 'asc') # asc|desc # --- Base query (eager-load to avoid N+1) --- q = (Employee.query .options(joinedload(Employee.department), joinedload(Employee.designation), joinedload(Employee.role))) # --- Filters --- if name: q = q.filter(Employee.full_name.ilike(f"%{name}%")) if code: q = q.filter(Employee.employee_code.ilike(f"%{code}%")) if dept_id: q = q.filter(Employee.department_id == dept_id) if designation_id: q = q.filter(Employee.designation_id == designation_id) if role_name: q = q.join(Role, Employee.role_id == Role.id).filter(Role.name == role_name) if status: q = q.filter(Employee.status.ilike(status)) if institute: # treat "institute" as employee_code prefix q = q.filter(Employee.employee_code.ilike(f"{institute}%")) if location: q = q.filter(Employee.location == location) # --- Sorting (whitelist) --- sort_map = { 'name': Employee.full_name, 'code': Employee.employee_code, 'department': Employee.department_id, 'designation': Employee.designation_id, 'status': Employee.status, } sort_col = sort_map.get(sort, Employee.full_name) order_fn = asc if direction.lower() != 'desc' else desc q = q.order_by(order_fn(sort_col), asc(Employee.id)) # --- Pagination --- pagination = q.paginate(page=page, per_page=per_page, error_out=False) employees = pagination.items # Precompute page number window for Jinja (avoid using max/min in template) total_pages = pagination.pages or 1 start = max(1, pagination.page - 2) end = min(total_pages, pagination.page + 2) page_numbers = list(range(start, end + 1)) # --- Dropdown data --- # Departments & roles (already existed) departments = Department.query.order_by(Department.name).all() roles = Role.query.order_by(Role.name).all() # Designations designations = Designation.query.order_by(Designation.title).all() # Institute list (from employee_code prefix) code_rows = db.session.query(Employee.employee_code).filter( Employee.employee_code.isnot(None) ).all() inst_set = set() for (c,) in code_rows: p = _extract_emp_prefix(c) if p: inst_set.add(p) institutes = sorted(inst_set) # Locations (workplace) loc_rows = db.session.query(Employee.location).filter( Employee.location.isnot(None), Employee.location != "" ).distinct().all() locations = sorted({loc for (loc,) in loc_rows}) return render_template( 'employee/view_employees.html', employees=employees, roles=roles, departments=departments, designations=designations, institutes=institutes, locations=locations, # pagination helpers pagination=pagination, page_numbers=page_numbers, # echo filters name=name, code=code, dept_id=dept_id, role_name=role_name, status=status, institute=institute, location=location, designation_id=designation_id, per_page=per_page, sort=sort, direction=direction ) @employee_bp.route('/logs', methods=['GET']) @login_required def view_change_log(): user_filter = request.args.get('user', '').strip() field_filter = request.args.get('field', '').strip() start_date = request.args.get('start_date') end_date = request.args.get('end_date') page = request.args.get('page', 1, type=int) from sqlalchemy.orm import aliased emp = aliased(Employee) changer = aliased(Employee) query = db.session.query( EmployeeChangeLog, emp.full_name.label('employee_name'), changer.full_name.label('changer_name') ).outerjoin(emp, emp.id == EmployeeChangeLog.record_id ).outerjoin(changer, changer.id == EmployeeChangeLog.changed_by) if user_filter: query = query.filter(changer.full_name.ilike(f"%{user_filter}%")) if field_filter: query = query.filter(EmployeeChangeLog.field_name.ilike(f"%{field_filter}%")) if start_date: query = query.filter(EmployeeChangeLog.changed_on >= start_date) if end_date: query = query.filter(EmployeeChangeLog.changed_on <= end_date) logs = query.order_by(EmployeeChangeLog.changed_on.desc()).paginate(page=page, per_page=10) return render_template( 'employee/view_change_log.html', logs=logs, user_filter=user_filter, field_filter=field_filter, start_date=start_date, end_date=end_date ) @employee_bp.route('/generate-public-link', methods=['POST']) @login_required def generate_public_link(): from datetime import timedelta import uuid description = request.form.get('description') expires_days = int(request.form.get('expires_days', 7)) expires_on = datetime.utcnow() + timedelta(days=expires_days) new_link = PublicLink( token=str(uuid.uuid4()), description=description, expires_on=expires_on ) db.session.add(new_link) db.session.commit() flash(f'Public link generated: {new_link.token}', 'success') return redirect(url_for('employee.list_public_links')) @employee_bp.route('/admin/public-links', methods=['GET', 'POST']) @login_required def list_public_links(): if current_user.role.name.lower() not in ['admin', 'hr','hr admin','demo admin']: flash('Access denied', 'danger') return redirect(url_for('employee.dashboard')) if request.method == 'POST': import uuid description = request.form.get('description', '').strip() expires_days = int(request.form.get('expires_days', 7)) expires_on = datetime.utcnow() + timedelta(days=expires_days) new_link = PublicLink( token=str(uuid.uuid4()), created_on=datetime.utcnow(), expires_on=expires_on, is_used=False, description=description ) db.session.add(new_link) db.session.commit() flash('New public link generated!', 'success') return redirect(url_for('employee.list_public_links', q=request.args.get('q',''), page=request.args.get('page', 1), per_page=request.args.get('per_page', 10))) # GET: filter + paginate page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) q = request.args.get('q', '', type=str).strip() query = PublicLink.query if q: query = query.filter(func.lower(PublicLink.description).like(f"%{q.lower()}%")) query = query.order_by(PublicLink.created_on.desc()) pagination = query.paginate(page=page, per_page=per_page, error_out=False) links = pagination.items # compute page_numbers for Jinja total_pages = pagination.pages or 1 start = 1 if pagination.page - 2 < 1 else pagination.page - 2 end = total_pages if pagination.page + 2 > total_pages else pagination.page + 2 page_numbers = list(range(start, end + 1)) return render_template( 'employee/public_links.html', links=links, pagination=pagination, q=q, per_page=per_page, page_numbers=page_numbers, # << pass to template ) #------------------------- from datetime import datetime from werkzeug.utils import secure_filename import os, uuid @employee_bp.route('/public/add', methods=['GET', 'POST']) def public_add_employee(): token_value = request.args.get('token') if not token_value: return "Invalid or missing token", 403 link = PublicLink.query.filter_by(token=token_value).first() if not link or link.is_used or (link.expires_on and link.expires_on < datetime.utcnow()): return "This link is invalid or expired.", 403 departments = Department.query.all() # kept for future (safe even if not shown) designations = Designation.query.all() if request.method == 'POST': # --- basic email validation (required) --- email = (request.form.get('email') or '').strip() if not email or '@' not in email: # keep a friendly message; you can render a template with flash if you prefer return "Valid email is required.", 400 # --- optional company fields (removed from form) --- dept_id = request.form.get('department_id') or None desig_id = request.form.get('designation_id') or None doj_raw = request.form.get('date_of_joining') or None doj = None if doj_raw: try: doj = datetime.strptime(doj_raw, '%Y-%m-%d').date() except Exception: pass # ignore invalid; stays None uploads_dir = os.path.join('static', 'uploads') os.makedirs(uploads_dir, exist_ok=True) def save_file(field_name): f = request.files.get(field_name) if f and f.filename: filename = secure_filename(f.filename) full_path = os.path.join(uploads_dir, filename) f.save(full_path) return os.path.join('uploads', filename).replace("\\", "/") return None pending = PendingEmployee( # required full_name=request.form.get('full_name'), email=email, # optional phone (form didn’t include, but keep read) phone=request.form.get('phone'), # company now optional department_id=dept_id, designation_id=desig_id, date_of_joining=doj, # education summary (optional) qualification=request.form.get('qualification'), institute=request.form.get('institute'), # single-file resume_path=save_file('resume'), id_proof_path=save_file('id_proof'), # personal gender=request.form.get('gender'), address=request.form.get('address'), dob=request.form.get('dob'), father_spouse_name=request.form.get('father_spouse_name'), marital_status=request.form.get('marital_status'), temporary_address=request.form.get('temporary_address'), personal_number=request.form.get('personal_number'), emergency_contact_name=request.form.get('emergency_contact_name'), emergency_contact_number=request.form.get('emergency_contact_number'), # id/bank aadhar_name=request.form.get('aadhar_name'), aadhar_number=request.form.get('aadhar_number'), pan_name=request.form.get('pan_name'), pan_number=request.form.get('pan_number'), uan=request.form.get('uan'), last_company_pf_number=request.form.get('last_company_pf_number'), personal_email=request.form.get('personal_email'), bank_account_name=request.form.get('bank_account_name'), bank_account_number=request.form.get('bank_account_number'), ifsc_code=request.form.get('ifsc_code'), bank_name=request.form.get('bank_name'), # detailed education secondary_school_name=request.form.get('secondary_name'), secondary_school_marks=request.form.get('secondary_marks'), secondary_school_year=request.form.get('secondary_year'), higher_secondary_school_name=request.form.get('higher_secondary_name'), higher_secondary_school_marks=request.form.get('higher_secondary_marks'), higher_secondary_school_year=request.form.get('higher_secondary_year'), graduation_name=request.form.get('graduation_name'), graduation_marks=request.form.get('graduation_marks'), graduation_year=request.form.get('graduation_year'), post_graduation_name=request.form.get('post_graduation_name'), post_graduation_marks=request.form.get('post_graduation_marks'), post_graduation_year=request.form.get('post_graduation_year'), # professional last_company_name=request.form.get('last_company_name'), last_company_code=request.form.get('last_company_code'), last_company_designation=request.form.get('last_company_designation'), # multi-docs doc_qualification=save_file('doc_qualification'), doc_employment_proof=save_file('doc_employment_proof'), doc_photographs=save_file('doc_photographs'), doc_pan_card=save_file('doc_pan_card'), doc_aadhar_card=save_file('doc_aadhar_card'), doc_parents_aadhar=save_file('doc_parents_aadhar'), doc_uan_card=save_file('doc_uan_card'), doc_cancelled_cheque=save_file('doc_cancelled_cheque') ) try: db.session.add(pending) db.session.flush() # ensure id allocated link.is_used = True db.session.commit() except Exception as e: db.session.rollback() # don’t consume the link on failure return f"Failed to submit: {e}", 500 return redirect(url_for('employee.public_thank_you')) return render_template('employee/public_add_employee.html', departments=departments, designations=designations, token_value=token_value, hide_sidebar=True) #------------------------- @employee_bp.route('/public/thank-you') def public_thank_you(): return render_template('employee/public_thank_you.html', hide_sidebar=True) # routes/employee.py from sqlalchemy import func, or_ from flask import request, render_template from flask_login import login_required from models import db, PendingEmployee, Employee, Department, Designation # adjust imports to your project from flask import request, render_template from flask_login import login_required from sqlalchemy import func, or_ from models import db, PendingEmployee, Employee, Department, Designation @employee_bp.route('/pending-approvals', methods=['GET']) @login_required def view_pending_employees(): q = (request.args.get('q') or '').strip() dept_id = request.args.get('department_id', type=int) desig_id = request.args.get('designation_id', type=int) doj_from = request.args.get('doj_from') # YYYY-MM-DD doj_to = request.args.get('doj_to') # YYYY-MM-DD page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) query = PendingEmployee.query # ---- dynamic search over existing columns only ---- if q: like = f"%{q.lower()}%" search_fields = [] if hasattr(PendingEmployee, 'full_name'): search_fields.append(func.lower(PendingEmployee.full_name).like(like)) if hasattr(PendingEmployee, 'email'): search_fields.append(func.lower(PendingEmployee.email).like(like)) if hasattr(PendingEmployee, 'phone'): search_fields.append(func.lower(PendingEmployee.phone).like(like)) # Add more if they exist on your model: # if hasattr(PendingEmployee, 'some_field'): # search_fields.append(func.lower(PendingEmployee.some_field).like(like)) if search_fields: query = query.filter(or_(*search_fields)) # ---- filters (guard with hasattr to avoid attribute errors) ---- if dept_id and hasattr(PendingEmployee, 'department_id'): query = query.filter(PendingEmployee.department_id == dept_id) if desig_id and hasattr(PendingEmployee, 'designation_id'): query = query.filter(PendingEmployee.designation_id == desig_id) if doj_from and hasattr(PendingEmployee, 'date_of_joining'): query = query.filter(PendingEmployee.date_of_joining >= doj_from) if doj_to and hasattr(PendingEmployee, 'date_of_joining'): query = query.filter(PendingEmployee.date_of_joining <= doj_to) query = query.order_by( getattr(PendingEmployee, 'submitted_on', None) or func.lower(PendingEmployee.full_name) ) pagination = query.paginate(page=page, per_page=per_page, error_out=False) pending = pagination.items managers = (Employee.query .filter(func.lower(Employee.status).in_(['active', 'notice'])) .order_by(func.lower(Employee.full_name)) .all()) departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() total_pages = pagination.pages or 1 start = max(1, pagination.page - 2) end = min(total_pages, pagination.page + 2) page_numbers = list(range(start, end + 1)) return render_template( 'employee/pending_approvals.html', pending=pending, managers=managers, pagination=pagination, page_numbers=page_numbers, per_page=per_page, q=q, dept_id=dept_id, desig_id=desig_id, doj_from=doj_from, doj_to=doj_to, departments=departments, designations=designations, ) @employee_bp.route('/approve/', methods=['POST']) @login_required def approve_pending_employee(pending_id): pending = PendingEmployee.query.get_or_404(pending_id) employee_code = request.form['employee_code'] reporting_manager_id = request.form.get('reporting_manager_id') or None doj = None doj_raw = request.form.get('date_of_joining') # may be blank or absent if doj_raw: try: doj = datetime.strptime(doj_raw, '%Y-%m-%d').date() except Exception: doj = None # ignore invalid formats import secrets from sqlalchemy.exc import IntegrityError email_conflict = None if pending.email: email_conflict = Employee.query.filter( func.lower(Employee.email) == pending.email.lower(), func.lower(func.coalesce(Employee.status, "")). in_(["active", "notice_period", "notice period"]) ).first() if email_conflict: flash("❌ Email already belongs to an active employee.", "danger") return redirect(url_for('employee.view_pending_employees')) department = Department.query.get(pending.department_id) if department and department.name.lower() != 'management' and not reporting_manager_id: flash('❌ Reporting Manager is required for non-management departments.', 'danger') return redirect(url_for('employee.view_pending_employees')) approved = Employee( employee_code=employee_code, full_name=pending.full_name, email=pending.email, phone=pending.phone, department_id=pending.department_id, designation_id=pending.designation_id, reporting_manager_id=reporting_manager_id, date_of_joining=doj or pending.date_of_joining or None, # <-- OPTIONAL NOW qualification=pending.qualification, institute=pending.institute, resume_path=pending.resume_path, id_proof_path=pending.id_proof_path, # New fields mapped gender=pending.gender, address=pending.address, dob=pending.dob, father_spouse_name=pending.father_spouse_name, marital_status=pending.marital_status, temporary_address=pending.temporary_address, personal_number=pending.personal_number, emergency_contact_name=pending.emergency_contact_name, emergency_contact_number=pending.emergency_contact_number, aadhar_name=pending.aadhar_name, aadhar_number=pending.aadhar_number, pan_name=pending.pan_name, pan_number=pending.pan_number, uan=pending.uan, last_company_pf_number=pending.last_company_pf_number, personal_email=pending.personal_email, bank_account_name=pending.bank_account_name, bank_account_number=pending.bank_account_number, ifsc_code=pending.ifsc_code, bank_name=pending.bank_name, secondary_school_name=pending.secondary_school_name, secondary_school_marks=pending.secondary_school_marks, secondary_school_year=pending.secondary_school_year, higher_secondary_school_name=pending.higher_secondary_school_name, higher_secondary_school_marks=pending.higher_secondary_school_marks, higher_secondary_school_year=pending.higher_secondary_school_year, graduation_name=pending.graduation_name, graduation_marks=pending.graduation_marks, graduation_year=pending.graduation_year, post_graduation_name=pending.post_graduation_name, post_graduation_marks=pending.post_graduation_marks, post_graduation_year=pending.post_graduation_year, last_company_name=pending.last_company_name, last_company_code=pending.last_company_code, last_company_designation=pending.last_company_designation, doc_qualification=pending.doc_qualification, doc_employment_proof=pending.doc_employment_proof, doc_photographs=pending.doc_photographs, doc_pan_card=pending.doc_pan_card, doc_aadhar_card=pending.doc_aadhar_card, doc_parents_aadhar=pending.doc_parents_aadhar, doc_uan_card=pending.doc_uan_card, doc_cancelled_cheque=pending.doc_cancelled_cheque, created_by=current_user.id, set_password_token=secrets.token_urlsafe(32) ) try: db.session.add(approved) db.session.flush() # required to fetch approved.id before creating log log = PendingApprovalLog( pending_employee_id=pending.id, approved_employee_id=approved.id, approved_by=current_user.id, status='approved' ) db.session.add(log) db.session.delete(pending) db.session.commit() flash('✅ Employee approved and added.', 'success') except IntegrityError as e: db.session.rollback() if 'employee_code' in str(e.orig): flash('❌ Employee code already exists. Please enter a new code.', 'danger') else: flash(f'❌ An error occurred while approving: {str(e.orig)}', 'danger') return redirect(url_for('employee.view_pending_employees')) @employee_bp.route('/approval-logs') @login_required def view_approval_logs(): page = request.args.get('page', 1, type=int) logs = PendingApprovalLog.query.order_by(PendingApprovalLog.approved_on.desc()).paginate(page=page, per_page=20) return render_template('employee/view_approval_log.html', logs=logs) @employee_bp.route('/reject/', methods=['POST']) @login_required def reject_pending_employee(pending_id): pending = PendingEmployee.query.get_or_404(pending_id) log = PendingApprovalLog( pending_employee_id=pending.id, pending_employee_name=pending.full_name, approved_employee_id=None, approved_by=current_user.id, status='rejected' ) db.session.add(log) db.session.delete(pending) db.session.commit() flash('❌ Pending employee rejected.', 'warning') return redirect(url_for('employee.view_pending_employees')) @employee_bp.route('/employee/set-password', methods=['GET', 'POST']) def set_password(): token = request.args.get('token') employee = Employee.query.filter_by(set_password_token=token).first() if not employee: flash('❌ Invalid or expired token.', 'danger') return redirect(url_for('employee.employee_login')) if request.method == 'POST': password = request.form['password'] confirm_password = request.form['confirm_password'] if password != confirm_password: flash('❌ Passwords do not match.', 'danger') return redirect(request.url) employee.set_password(password) employee.set_password_token = None # Invalidate token db.session.commit() flash('✅ Password set successfully! You can now log in.', 'success') return redirect(url_for('employee.employee_login')) return render_template('employee/set_password.html', employee=employee, hide_sidebar=True) import uuid @employee_bp.route('/employee/regenerate-token/', methods=['POST']) @login_required def regenerate_token(emp_id): emp = Employee.query.get_or_404(emp_id) emp.set_password_token = str(uuid.uuid4()) db.session.commit() flash(f"🔁 New password link generated for {emp.full_name}", "success") return redirect(url_for('employee.view_employees')) # or wherever you list employees from datetime import date # add this if not present @employee_bp.route('/employee/login', methods=['GET', 'POST']) def employee_login(): if current_user.is_authenticated: return redirect(url_for('employee.dashboard')) if request.method == 'POST': employee_code = request.form['employee_code'] password = request.form['password'] employee = Employee.query.filter_by(employee_code=employee_code).first() if employee and employee.check_password(password): # 🔐 Block login based on employment status / exit date (non-admins) emp_status = (employee.status or '').strip().lower() lwd = getattr(employee, 'exit_date', None) # If already RESIGNED -> never allow login if emp_status == 'resigned': flash("Your account is inactive after separation. Please contact HR.", "danger") return redirect(url_for('employee.employee_login')) # If on NOTICE PERIOD and today is after LWD -> block if emp_status == 'notice_period' and lwd and date.today() > lwd: flash("Your last working day has passed. Login is disabled. Please contact HR.", "danger") return redirect(url_for('employee.employee_login')) # ⛔ Check if role is assigned if not employee.role or not employee.role.name: flash("❌ Your role is not yet assigned. Please contact HR.", "danger") return redirect(url_for('employee.employee_login')) # ✅ Check salary structure only for non-admins role_name = employee.role.name.strip().lower() if role_name != 'admin': from models import EmployeeSalaryStructures has_salary = EmployeeSalaryStructures.query.filter( EmployeeSalaryStructures.employee_id == employee.id, EmployeeSalaryStructures.amount > 0 ).first() if not has_salary: flash("❌ You cannot log in until your salary structure is assigned. Please contact HR.", "danger") return redirect(url_for('employee.employee_login')) # All good -> login login_user(employee) flash('✅ Logged in successfully!', 'success') next_page = request.args.get('next') return redirect(next_page or url_for('employee.dashboard')) else: flash('❌ Invalid employee code or password.', 'danger') return render_template('employee/employee_login.html', hide_sidebar=True) from sqlalchemy import extract, and_, or_ @employee_bp.route('/employee/dashboard') @employee_required def dashboard(): today = date.today() current_year = today.year current_month = today.month # --- Broadcasts (no change) --- broadcasts = BroadcastMessage.query.filter( ((BroadcastMessage.start_date == None) | (BroadcastMessage.start_date <= today)), ((BroadcastMessage.end_date == None) | (BroadcastMessage.end_date >= today)), (~BroadcastMessage.departments.any() | BroadcastMessage.departments.any(id=current_user.department_id)), (~BroadcastMessage.roles.any() | BroadcastMessage.roles.any(id=current_user.role_id)) ).order_by(BroadcastMessage.created_on.desc()).all() read_ids = [r.broadcast_id for r in current_user.read_broadcasts.all()] # --- Birthdays (Active + Notice) --- birthdays = ( Employee.query .filter(status_allowed_filter(Employee)) .filter(extract('month', Employee.dob) == current_month) .order_by(extract('day', Employee.dob).asc()) .all() ) # --- New Joinees (Active + Notice) --- new_joinees = ( Employee.query .filter(status_allowed_filter(Employee)) .filter( extract('month', Employee.date_of_joining) == current_month, extract('year', Employee.date_of_joining) == current_year ) .order_by(Employee.date_of_joining.asc()) .all() ) # --- Holidays --- holidays = Holiday.query.filter( extract('month', Holiday.date) == current_month, extract('year', Holiday.date) == current_year ).order_by(Holiday.date.asc()).all() # --- Employees on Leave --- first_day = date(current_year, current_month, 1) next_month = date(current_year + (1 if current_month == 12 else 0), 1 if current_month == 12 else current_month + 1, 1) employees_on_leave = ( LeaveRequest.query .join(Employee, Employee.id == LeaveRequest.employee_id) .filter(status_allowed_filter(Employee)) .filter( or_( LeaveRequest.status == 'final_approved', LeaveRequest.status == 'manager_approved' ), or_( and_(LeaveRequest.start_date >= today, LeaveRequest.start_date < next_month), and_(LeaveRequest.end_date >= today, LeaveRequest.end_date < next_month), and_(LeaveRequest.start_date <= today, LeaveRequest.end_date >= today) ) ) .order_by(LeaveRequest.start_date.asc()) .all() ) return render_template( 'employee/dashboard.html', broadcasts=broadcasts, employee=current_user, read_ids=read_ids, birthdays=birthdays, new_joinees=new_joinees, holidays=holidays, employees_on_leave=employees_on_leave, current_year=current_year, current_month=current_month ) @employee_bp.route('/update-role/', methods=['POST']) @login_required def update_role(emp_id): if current_user.role.name not in ['admin','hr','hr admin']: flash('❌ Only admins and HR can update roles.', 'danger') return redirect(url_for('employee.view_employees')) emp = Employee.query.get_or_404(emp_id) new_role = request.form.get('role') role_obj = Role.query.filter_by(name=new_role).first() if not new_role: flash('❌ No role selected.', 'danger') return redirect(url_for('employee.view_employees')) emp.role = role_obj db.session.commit() flash(f'✅ Role updated to {new_role} for {emp.full_name}.', 'success') return redirect(url_for('employee.view_employees')) @employee_bp.route('/employee/forgot-password', methods=['GET', 'POST']) def forgot_password(): if request.method == 'POST': email = request.form['email'] employee = (Employee.query .filter( func.lower(Employee.email) == email.lower(), func.lower(func.coalesce(Employee.status, "")). in_(["active", "notice_period", "notice period"]) ) .order_by(Employee.id.desc()) # latest if multiple .first() ) if employee: import secrets employee.set_password_token = secrets.token_urlsafe(32) db.session.commit() # Here: send email or flash link (for testing) reset_link = url_for('employee.set_password', token=employee.set_password_token, _external=True) flash(f'Reset link: {reset_link}', 'info') # Replace with actual email send else: flash('❌ Email not found.', 'danger') return render_template('employee/forgot_password.html',hide_sidebar=True) @employee_bp.route("/me/profile", methods=["GET", "POST"]) @login_required def my_profile(): # Current user is an Employee record in your app emp: Employee = Employee.query.get_or_404(current_user.id) # Block fully-exited users from edits (optional) if request.method == "POST" and (emp.status or "").lower() == "resigned": flash("Your profile is read‑only after exit is finalised.", "warning") return redirect(url_for("employee.my_profile")) if request.method == "POST": f = request.form def val(name: str) -> Optional[str]: return (f.get(name) or "").strip() or None # ---- Allowed editable fields (self‑service) ---- emp.phone = val("phone") emp.personal_email = val("personal_email") emp.address = val("address") emp.temporary_address = val("temporary_address") emp.personal_number = val("personal_number") emp.emergency_contact_name = val("emergency_contact_name") emp.emergency_contact_number = val("emergency_contact_number") # Government IDs (if you allow) emp.aadhar_name = val("aadhar_name") emp.aadhar_number = val("aadhar_number") emp.pan_name = val("pan_name") emp.pan_number = val("pan_number") # Banking (optional to expose in self‑service; remove if HR-only) emp.bank_account_name = val("bank_account_name") emp.bank_account_number = val("bank_account_number") emp.ifsc_code = val("ifsc_code") emp.bank_name = val("bank_name") # ⚠️ Company‑controlled fields deliberately NOT editable here: # employee_code / department_id / designation_id / # reporting_manager_id / work_mode / shift_start / shift_end / desktime_id / date_of_joining / etc. # ---------- File uploads (personal docs) ---------- # inside my_profile() upload_root = os.path.join(current_app.config['UPLOAD_FOLDER'], str(emp.id)) os.makedirs(upload_root, exist_ok=True) def save_file(field_name: str, current_path: Optional[str]) -> Optional[str]: file = request.files.get(field_name) if file and file.filename: safe = secure_filename(file.filename) full_path = os.path.join(upload_root, safe) # absolute path file.save(full_path) # store a route-friendly relative path (either "uploads/..." or just "/...") return f"uploads/{emp.id}/{safe}" return current_path # Map only the docs you want to expose for self‑service emp.resume_path = save_file("resume", emp.resume_path) emp.id_proof_path = save_file("id_proof", emp.id_proof_path) emp.doc_qualification = save_file("doc_qualification", emp.doc_qualification) emp.doc_photographs = save_file("doc_photographs", emp.doc_photographs) emp.doc_pan_card = save_file("doc_pan_card", emp.doc_pan_card) emp.doc_aadhar_card = save_file("doc_aadhar_card", emp.doc_aadhar_card) emp.doc_parents_aadhar = save_file("doc_parents_aadhar", emp.doc_parents_aadhar) emp.doc_uan_card = save_file("doc_uan_card", emp.doc_uan_card) emp.doc_cancelled_cheque = save_file("doc_cancelled_cheque", emp.doc_cancelled_cheque) # Audit emp.updated_on = datetime.utcnow() emp.updated_by = current_user.id db.session.commit() flash("✅ Your profile has been updated.", "success") return redirect(url_for("employee.my_profile")) # GET return render_template("employee/my_profile.html", emp=emp) @employee_bp.route('/documents/') @login_required def employee_documents(employee_id): emp = Employee.query.get_or_404(employee_id) documents = [ ("Resume", emp.resume_path), ("ID Proof", emp.id_proof_path), ("Qualification Certificate", emp.doc_qualification), ("Employment Proof", emp.doc_employment_proof), ("Photographs", emp.doc_photographs), ("PAN Card", emp.doc_pan_card), ("Aadhar Card", emp.doc_aadhar_card), ("Parents Aadhar", emp.doc_parents_aadhar), ("UAN Card", emp.doc_uan_card), ("Cancelled Cheque", emp.doc_cancelled_cheque), ] return render_template('employee/employee_documents.html', employee=emp, documents=documents) # --- Employee Import (CSV/XLSX) ----------------------------------------------- from io import BytesIO import csv, uuid import pandas as pd from werkzeug.utils import secure_filename from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import joinedload from calendar import monthrange ALLOWED_EMP_IMPORT_EXTS = {".csv", ".xlsx", ".xls"} def _is_hr_admin(): role = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() return role in ("hr", "hr admin", "admin") def _safe_ext(filename: str) -> str: f = filename or "" dot = f.rfind(".") return f[dot:].lower() if dot != -1 else "" def _read_employee_file(file_path: str) -> pd.DataFrame: ext = _safe_ext(file_path) if ext == ".csv": return pd.read_csv(file_path, dtype=str).fillna("") elif ext in (".xlsx", ".xls"): return pd.read_excel(file_path, dtype=str).fillna("") raise ValueError("Unsupported file type") def _expected_columns(): # minimal set; you can add more fields later return [ "employee_code", "full_name", "email", "phone", "department", "designation", "reporting_manager_code", "date_of_joining", # YYYY-MM-DD "status", # active | notice_period | resigned "work_mode", # WFO | WFH | Hybrid (optional) "location" # optional ] from io import StringIO from flask import make_response @employee_bp.route("/import-employees/template") @login_required def download_employee_import_template(): if not _is_hr_admin(): flash("Access denied.", "danger") return redirect(url_for("employee.view_employees")) expected = _expected_columns() sample = [{ "employee_code": "E0001", "full_name": "John Doe", "email": "john.doe@company.com", "phone": "9876543210", "department": "Sales", "designation": "Executive", "reporting_manager_code": "M0001", "date_of_joining": "2025-10-01", "status": "active", "work_mode": "Hybrid", "location": "Kolkata", }] # Write CSV to a text buffer first sio = StringIO() writer = csv.DictWriter(sio, fieldnames=expected) writer.writeheader() for r in sample: writer.writerow(r) # Encode with UTF-8 BOM so Excel opens cleanly data = sio.getvalue().encode("utf-8-sig") resp = make_response(data) resp.headers["Content-Disposition"] = "attachment; filename=employee_import_template.csv" resp.headers["Content-Type"] = "text/csv; charset=utf-8" return resp @employee_bp.route("/import-employees", methods=["GET", "POST"]) @login_required def import_employees(): """ Step 1 (GET): show upload form Step 2 (POST action=preview): upload file, parse, validate, render preview Step 3 (POST action=confirm): actually insert rows that are valid """ if not _is_hr_admin(): flash("Access denied.", "danger") return redirect(url_for("employee.view_employees")) action = (request.form.get("action") or request.args.get("action") or "form").strip() # Where to store temp files tmp_dir = os.path.join(current_app.root_path, "static", "uploads", "tmp") os.makedirs(tmp_dir, exist_ok=True) if request.method == "GET" or action == "form": departments = Department.query.order_by(Department.name).all() designations = Designation.query.order_by(Designation.title).all() return render_template("employee/import_employees.html", step="form", departments=departments, designations=designations) # POST handlers if action == "preview": f = request.files.get("file") auto_create = bool(request.form.get("auto_create_dept_desig")) if not f or not f.filename: flash("Please choose a CSV/XLSX file.", "warning") return redirect(url_for("employee.import_employees")) ext = _safe_ext(f.filename) if ext not in ALLOWED_EMP_IMPORT_EXTS: flash("Unsupported file type. Please upload .csv or .xlsx", "danger") return redirect(url_for("employee.import_employees")) temp_name = f"emp_import_{uuid.uuid4().hex}{ext}" temp_path = os.path.join(tmp_dir, secure_filename(temp_name)) f.save(temp_path) try: df = _read_employee_file(temp_path) except Exception as e: flash(f"Failed to read file: {e}", "danger") return redirect(url_for("employee.import_employees")) # Normalize columns (lowercase, underscores) df.columns = [c.strip().lower() for c in df.columns] # Ensure required columns exist missing = [c for c in _expected_columns() if c not in df.columns] if missing: flash(f"Missing columns: {', '.join(missing)}", "danger") return redirect(url_for("employee.import_employees")) # Build quick lookups emp_by_code = {e.employee_code: e for e in Employee.query.all()} dept_by_name = {d.name.strip().lower(): d for d in Department.query.all()} desig_by_name = {d.title.strip().lower(): d for d in Designation.query.all()} preview_rows = [] any_errors = False row_index = 0 for _, row in df.iterrows(): row_index += 1 r = {k: (row.get(k, "") or "").strip() for k in _expected_columns()} errors = [] # basic validations if not r["employee_code"]: errors.append("employee_code required") if r["employee_code"] and r["employee_code"] in emp_by_code: errors.append("employee_code already exists") if not r["full_name"]: errors.append("full_name required") if r["email"] and "@" not in r["email"]: errors.append("invalid email") # Department dept_key = r["department"].strip().lower() dept_obj = dept_by_name.get(dept_key) if not dept_obj: if auto_create and r["department"]: dept_obj = Department(name=r["department"]) db.session.add(dept_obj) db.session.flush() dept_by_name[dept_key] = dept_obj else: errors.append("department not found") # Designation desig_key = r["designation"].strip().lower() desig_obj = desig_by_name.get(desig_key) if not desig_obj: if auto_create and r["designation"]: desig_obj = Designation(title=r["designation"]) db.session.add(desig_obj) db.session.flush() desig_by_name[desig_key] = desig_obj else: errors.append("designation not found") # Reporting Manager by employee_code (optional) rm_code = r["reporting_manager_code"] rm = emp_by_code.get(rm_code) if rm_code else None if rm_code and not rm: errors.append("reporting_manager_code not found") # date_of_joining doj = r["date_of_joining"] if doj: try: _ = datetime.strptime(doj, "%Y-%m-%d").date() except Exception: errors.append("date_of_joining must be YYYY-MM-DD") # status status = (r["status"] or "active").lower() if status not in ("active", "notice_period", "resigned"): errors.append("status must be active|notice_period|resigned") # Compile row_status = "OK" if not errors else "ERROR" any_errors = any_errors or bool(errors) preview_rows.append({ "idx": row_index, "data": r, "dept_id": getattr(dept_obj, "id", None), "desig_id": getattr(desig_obj, "id", None), "rm_id": getattr(rm, "id", None), "errors": errors, "status": row_status, }) if any_errors: flash("Some rows have issues. Fix the file or tick Auto-Create for Master data where applicable.", "warning") # Render preview with temp file token return render_template( "employee/import_employees.html", step="preview", rows=preview_rows, temp_file=os.path.basename(temp_path), auto_create=auto_create ) if action == "confirm": temp_file = request.form.get("temp_file") auto_create = bool(request.form.get("auto_create")) if not temp_file: flash("Import session expired. Please re-upload.", "warning") return redirect(url_for("employee.import_employees")) temp_path = os.path.join(tmp_dir, secure_filename(temp_file)) if not os.path.exists(temp_path): flash("Temp file not found. Please re-upload.", "warning") return redirect(url_for("employee.import_employees")) try: df = _read_employee_file(temp_path) except Exception as e: flash(f"Failed to read file: {e}", "danger") return redirect(url_for("employee.import_employees")) df.columns = [c.strip().lower() for c in df.columns] emp_by_code = {e.employee_code: e for e in Employee.query.all()} dept_by_name = {d.name.strip().lower(): d for d in Department.query.all()} desig_by_name = {d.title.strip().lower(): d for d in Designation.query.all()} inserted, skipped = 0, 0 errs = [] for _, row in df.iterrows(): r = {k: (row.get(k, "") or "").strip() for k in _expected_columns()} # Validate again quickly (defensive) if not r["employee_code"] or r["employee_code"] in emp_by_code: skipped += 1 errs.append(f"{r['employee_code'] or ''}: duplicate/missing employee_code") continue dept_key = r["department"].strip().lower() dept = dept_by_name.get(dept_key) if not dept: if auto_create and r["department"]: dept = Department(name=r["department"]) db.session.add(dept); db.session.flush() dept_by_name[dept_key] = dept else: skipped += 1; errs.append(f"{r['employee_code']}: department not found") continue desig_key = r["designation"].strip().lower() desig = desig_by_name.get(desig_key) if not desig: if auto_create and r["designation"]: desig = Designation(title=r["designation"]) db.session.add(desig); db.session.flush() desig_by_name[desig_key] = desig else: skipped += 1; errs.append(f"{r['employee_code']}: designation not found") continue rm_id = None if r["reporting_manager_code"]: rm = emp_by_code.get(r["reporting_manager_code"]) if not rm: skipped += 1; errs.append(f"{r['employee_code']}: reporting manager code not found") continue rm_id = rm.id doj = None if r["date_of_joining"]: try: doj = datetime.strptime(r["date_of_joining"], "%Y-%m-%d").date() except Exception: skipped += 1; errs.append(f"{r['employee_code']}: invalid date_of_joining") continue status = (r["status"] or "active").lower() if status not in ("active", "notice_period", "resigned"): status = "active" # Default password policy (you can change) default_pwd = "Welcome@123" try: new_emp = Employee( employee_code=r["employee_code"], full_name=r["full_name"], email=r["email"] or None, phone=r["phone"] or None, department_id=dept.id, designation_id=desig.id, reporting_manager_id=rm_id, date_of_joining=doj, work_mode=r.get("work_mode") or None, location=r.get("location") or None, status=status, created_by=current_user.id, updated_by=current_user.id, created_on=datetime.utcnow() ) # set default password (if your model supports set_password) try: new_emp.set_password(default_pwd) except Exception: # if model doesn’t have set_password, ignore pass db.session.add(new_emp) db.session.flush() # to get id emp_by_code[new_emp.employee_code] = new_emp # for subsequent RM mapping if needed inserted += 1 except IntegrityError as e: db.session.rollback() skipped += 1 errs.append(f"{r['employee_code']}: DB integrity error ({str(e.orig)})") except Exception as e: db.session.rollback() skipped += 1 errs.append(f"{r['employee_code']}: {e}") try: db.session.commit() except Exception as e: db.session.rollback() flash(f"Commit failed: {e}", "danger") return redirect(url_for("employee.import_employees")) # Cleanup temp try: os.remove(temp_path) except Exception: pass msg = f"Import complete. Inserted={inserted}, Skipped={skipped}." flash(msg, "success" if inserted and not skipped else "warning" if inserted else "danger") if errs: flash(" | ".join(errs[:10]) + (" ..." if len(errs) > 10 else ""), "warning") return redirect(url_for("employee.view_employees")) # Fallback return redirect(url_for("employee.import_employees")) # --- Employee Export (CSV/XLSX) ---------------------------------------------- from io import BytesIO import pandas as pd from sqlalchemy.orm import joinedload from sqlalchemy import asc, desc from datetime import date, datetime as dt @employee_bp.route("/export-employees", methods=["GET"]) @login_required def export_employees(): # (optional) restrict to HR/Admin role = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() if role not in ("hr", "hr admin", "admin"): flash("Access denied.", "danger") return redirect(url_for("employee.view_employees")) # ---- read format ---- fmt = (request.args.get("format") or "csv").lower() # csv | xlsx if fmt not in ("csv", "xlsx"): fmt = "csv" # ---- same filters as view_employees ---- name = request.args.get('name', '').strip() code = request.args.get('code', '').strip() dept_id = request.args.get('department_id', type=int) role_name = request.args.get('role', '').strip() status = request.args.get('status', '').strip().lower() institute = request.args.get('institute', '').strip() location = request.args.get('location', '').strip() designation_id = request.args.get('designation_id', type=int) sort = request.args.get('sort', 'name') direction = request.args.get('dir', 'asc') q = (Employee.query .options(joinedload(Employee.department), joinedload(Employee.designation), joinedload(Employee.role))) if name: q = q.filter(Employee.full_name.ilike(f"%{name}%")) if code: q = q.filter(Employee.employee_code.ilike(f"%{code}%")) if dept_id: q = q.filter(Employee.department_id == dept_id) if designation_id: q = q.filter(Employee.designation_id == designation_id) if role_name: q = q.join(Role, Employee.role_id == Role.id).filter(Role.name == role_name) if status: q = q.filter(Employee.status.ilike(status)) if institute: q = q.filter(Employee.employee_code.ilike(f"{institute}%")) if location: q = q.filter(Employee.location == location) sort_map = { 'name': Employee.full_name, 'code': Employee.employee_code, 'department': Employee.department_id, 'designation': Employee.designation_id, 'status': Employee.status, } sort_col = sort_map.get(sort, Employee.full_name) order_fn = asc if direction.lower() != 'desc' else desc q = q.order_by(order_fn(sort_col), asc(Employee.id)) employees = q.all() # quick lookup for reporting manager code/name all_emp = Employee.query.with_entities(Employee.id, Employee.employee_code, Employee.full_name).all() emp_lookup = {e.id: (e.employee_code, e.full_name) for e in all_emp} # gather ALL db columns dynamically db_cols = [c.name for c in Employee.__table__.columns] rows = [] def _val(v): # normalize dates/datetimes for export if isinstance(v, dt): return v.strftime("%Y-%m-%d %H:%M:%S") if isinstance(v, date): return v.strftime("%Y-%m-%d") return v for emp in employees: row = {col: _val(getattr(emp, col, None)) for col in db_cols} # computed / friendly columns row.update({ "department_name": getattr(getattr(emp, "department", None), "name", None), "designation_title": getattr(getattr(emp, "designation", None), "title", None), "role_name": getattr(getattr(emp, "role", None), "name", None), "reporting_manager_code": emp_lookup.get(emp.reporting_manager_id, (None, None))[0], "reporting_manager_name": emp_lookup.get(emp.reporting_manager_id, (None, None))[1], }) rows.append(row) df = pd.DataFrame(rows) file_ts = dt.utcnow().strftime("%Y%m%d_%H%M%S") fn_base = f"employees_{file_ts}" if fmt == "xlsx": output = BytesIO() with pd.ExcelWriter(output, engine="xlsxwriter") as writer: df.to_excel(writer, index=False, sheet_name="Employees") output.seek(0) resp = make_response(output.read()) resp.headers["Content-Disposition"] = f"attachment; filename={fn_base}.xlsx" resp.headers["Content-Type"] = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" return resp # default CSV (UTF-8 BOM for Excel) csv_bytes = df.to_csv(index=False).encode("utf-8-sig") resp = make_response(csv_bytes) resp.headers["Content-Disposition"] = f"attachment; filename={fn_base}.csv" resp.headers["Content-Type"] = "text/csv; charset=utf-8" return resp # --- Employee Self Punch (IN/OUT) --------------------------------------------- def _today_punch_direction(emp_id: int) -> str: """Decide next direction for today: first punch = in, then alternate.""" today = date.today() start_dt = datetime.combine(today, time.min) end_dt = datetime.combine(today + timedelta(days=1), time.min) punches = (BiometricPunch.query .filter(BiometricPunch.employee_id == emp_id, BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt) .order_by(BiometricPunch.punch_time.asc()) .all()) if not punches: return "in" # alternate; last punch decides next return "out" if punches[-1].direction == "in" else "in" from PIL import Image, ExifTags import base64 def _get_exif_datetime(path): """ Try to read DateTimeOriginal / DateTime from image EXIF metadata. Returns a datetime or None. """ try: img = Image.open(path) exif = img._getexif() or {} tag_map = {ExifTags.TAGS.get(k): v for k, v in exif.items()} dt_raw = tag_map.get("DateTimeOriginal") or tag_map.get("DateTime") if not dt_raw: return None # EXIF uses "YYYY:MM:DD HH:MM:SS" return datetime.strptime(dt_raw, "%Y:%m:%d %H:%M:%S") except Exception: current_app.logger.exception("Failed to read EXIF datetime") return None from datetime import date, datetime, timedelta, time from sqlalchemy import func def _day_punch_direction(emp_id: int, punch_date: date) -> str: """Decide next direction for a given date: first punch = in, then alternate.""" start_dt = datetime.combine(punch_date, time.min) end_dt = datetime.combine(punch_date + timedelta(days=1), time.min) punches = (BiometricPunch.query .filter(BiometricPunch.employee_id == emp_id, BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt) .order_by(BiometricPunch.punch_time.asc()) .all()) if not punches: return "in" return "out" if punches[-1].direction == "in" else "in" def _month_key(d: date) -> str: """YYYY-MM string""" return f"{d.year:04d}-{d.month:02d}" def _is_payroll_processed_for_month(month_str: str, employee_id: int = None) -> bool: """ Returns True if payroll is processed/frozen for the given month. - month_str: 'YYYY-MM' - employee_id: optional, for per-employee payroll tables IMPORTANT: Adjust model/table names below based on your payroll.py. This function is defensive and won't crash if a model doesn't exist. """ try: import models # your project models module # 1) If you have a monthly freeze table (common) if hasattr(models, "PayrollFreeze"): q = models.PayrollFreeze.query.filter_by(month=month_str) return q.first() is not None # 2) If you store per-employee payroll rows (common) if hasattr(models, "Payroll"): q = models.Payroll.query.filter_by(month=month_str) if employee_id is not None and hasattr(models.Payroll, "employee_id"): q = q.filter_by(employee_id=employee_id) return q.first() is not None # 3) If you have a frozen payroll header table if hasattr(models, "FrozenPayroll"): q = models.FrozenPayroll.query.filter_by(month=month_str) return q.first() is not None # If none of the models exist, treat as "not processed" return False except Exception: current_app.logger.exception("Payroll processed check failed; allowing punch by default.") return False @employee_bp.route("/self-punch", methods=["GET", "POST"]) @login_required def self_punch(): emp: Employee = current_user # type: ignore # ---- Date selection (GET + POST) ---- today = date.today() # POST: date comes from form; GET: date comes from query param selected_date = None if request.method == "POST": d = request.form.get("punch_date") if d: try: selected_date = datetime.strptime(d, "%Y-%m-%d").date() except Exception: selected_date = None else: d = request.args.get("date") if d: try: selected_date = datetime.strptime(d, "%Y-%m-%d").date() except Exception: selected_date = None selected_date = selected_date or today # Block future dates if selected_date > today: flash("❌ You cannot punch for a future date.", "danger") return redirect(url_for("employee.self_punch")) # Block dates before DOJ (optional but recommended) doj = getattr(emp, "date_of_joining", None) if doj and selected_date < doj: flash("❌ You cannot punch for dates before your Date of Joining.", "danger") return redirect(url_for("employee.self_punch")) # Payroll lock: if month is processed, no punches for any date in that month month_str = _month_key(selected_date) if _is_payroll_processed_for_month(month_str, employee_id=emp.id): flash(f"⛔ Salary is already processed for {month_str}. Self punch is locked for this month.", "danger") return redirect(url_for("employee.self_punch")) if request.method == "POST": now = datetime.now() # Use selected date + current time punch_time = datetime.combine(selected_date, now.time()) lat_raw = request.form.get("lat") lon_raw = request.form.get("lon") note = (request.form.get("location_note") or "").strip() lat = float(lat_raw) if lat_raw else None lon = float(lon_raw) if lon_raw else None # 🔁 decide IN / OUT for selected_date (not just today) direction = _day_punch_direction(emp.id, selected_date) # 🗺️ reverse geocode if needed if not note and lat is not None and lon is not None: approx = _approx_location_from_latlng(lat, lon) if approx: note = approx # 📷 save photo (same as your existing code) photo_file = request.files.get("photo") captured_b64 = request.form.get("captured_image") photo_path = None photo_taken_at = None photo_is_fresh = None upload_root = os.path.join( current_app.root_path, current_app.config.get("UPLOAD_FOLDER", "uploads/documents"), "self_punch", str(emp.id), ) os.makedirs(upload_root, exist_ok=True) if captured_b64: try: header, _, b64data = captured_b64.partition(",") img_bytes = base64.b64decode(b64data) safe_name = secure_filename( f"{selected_date.strftime('%Y%m%d')}_{now.strftime('%H%M%S')}_cam.jpg" ) full_path = os.path.join(upload_root, safe_name) with open(full_path, "wb") as f: f.write(img_bytes) photo_path = os.path.join("self_punch", str(emp.id), safe_name).replace("\\", "/") photo_taken_at = now photo_is_fresh = True except Exception: current_app.logger.exception("Failed to save captured camera image") elif photo_file and photo_file.filename: safe_name = secure_filename( f"{selected_date.strftime('%Y%m%d')}_{now.strftime('%H%M%S')}_{photo_file.filename}" ) full_path = os.path.join(upload_root, safe_name) photo_file.save(full_path) photo_path = os.path.join("self_punch", str(emp.id), safe_name).replace("\\", "/") exif_dt = _get_exif_datetime(full_path) photo_taken_at = exif_dt if exif_dt: delta_sec = abs((now - exif_dt).total_seconds()) photo_is_fresh = (delta_sec <= 600) else: photo_is_fresh = None try: punch = BiometricPunch( sync_run_id=None, employee_id=emp.id, external_employee_code=emp.employee_code, punch_time=punch_time, # ✅ selected day direction=direction, device_code="WEB", source_type="web", source_name="self", raw_data=None, latitude=lat, longitude=lon, location_note=note, photo_path=photo_path, approval_status="pending", photo_taken_at=photo_taken_at, photo_is_fresh=photo_is_fresh, ) db.session.add(punch) db.session.commit() if selected_date != today: flash(f"✅ Previous day punch ({selected_date.strftime('%d %b %Y')}) sent for approval.", "success") else: flash(f"✅ Punch {direction.upper()} captured and sent for approval.", "success") except Exception as e: db.session.rollback() current_app.logger.exception("Self punch failed") flash(f"Failed to record punch: {e}", "danger") return redirect(url_for("employee.self_punch", date=selected_date.strftime("%Y-%m-%d"))) # ---------------- GET ---------------- next_dir = _day_punch_direction(emp.id, selected_date) start_dt = datetime.combine(selected_date, time.min) end_dt = datetime.combine(selected_date + timedelta(days=1), time.min) todays_punches = ( BiometricPunch.query.filter( BiometricPunch.employee_id == emp.id, BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt, ) .order_by(BiometricPunch.punch_time.asc()) .all() ) day_summary = get_biometric_day_summary(emp.id, selected_date, source_name=None) # Allow date selection within the month (only if month not processed already - already checked above) first_day_of_month = date(selected_date.year, selected_date.month, 1) min_date = first_day_of_month max_date = today return render_template( "employee/self_punch.html", next_dir=next_dir, todays_punches=todays_punches, day_summary=day_summary, today=selected_date, # template uses `today` label selected_date=selected_date, # ✅ new min_date=min_date, # ✅ new max_date=max_date, # ✅ new ) @employee_bp.route("/self-punch/history") @login_required def self_punch_history(): emp: Employee = current_user # type: ignore # month/year params today = date.today() month = request.args.get("month", type=int) or today.month year = request.args.get("year", type=int) or today.year first_day = date(year, month, 1) last_day = date(year, month, monthrange(year, month)[1]) # Preload punches for the whole month for this employee start_dt = datetime.combine(first_day, time.min) end_dt = datetime.combine(last_day + timedelta(days=1), time.min) punches = (BiometricPunch.query .filter(BiometricPunch.employee_id == emp.id, BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt) .order_by(BiometricPunch.punch_time.asc()) .all()) # group by day by_day = {} for p in punches: d = p.punch_time.date() by_day.setdefault(d, []).append(p) def _counts_for_hours(p: BiometricPunch) -> bool: """ For hours calculation: - device punches: always count - self punches: count only if approved """ if (p.source_name or "").lower() == "self": return (p.approval_status or "").lower() == "approved" return True # build per-day summary using first IN / last OUT history_rows = [] cur = first_day while cur <= last_day: day_punches = by_day.get(cur, []) # ✅ for display: show ALL punches (including pending/rejected) display_punches = day_punches # ✅ for hours: only "effective" punches effective = [p for p in day_punches if _counts_for_hours(p)] first_in = next((p for p in effective if p.direction == "in"), None) last_out = next((p for p in reversed(effective) if p.direction == "out"), None) check_in = first_in.punch_time if first_in else None check_out = last_out.punch_time if last_out else None total_hours = None if check_in and check_out and check_out > check_in: total_hours = (check_out - check_in).total_seconds() / 3600.0 history_rows.append({ "date": cur, "check_in": check_in, "check_out": check_out, "total_hours": total_hours, "punches": display_punches, # keep showing all (P/A/R badges in template) }) cur += timedelta(days=1) return render_template( "employee/self_punch_history.html", history_rows=history_rows, month=month, year=year, ) from sqlalchemy.orm import joinedload from sqlalchemy import or_ @employee_bp.route("/self-punch/approvals", methods=["GET"]) @login_required def self_punch_approvals(): role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() is_hr_admin = role_name in ("hr", "hr admin", "admin") # manager scope manager_id_arg = request.args.get("manager_id", type=int) if is_hr_admin and manager_id_arg: manager_id = manager_id_arg else: manager_id = current_user.id # optional date filter date_str = request.args.get("date") if date_str: try: filter_date = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: filter_date = date.today() else: filter_date = date.today() start_dt = datetime.combine(filter_date, time.min) end_dt = datetime.combine(filter_date + timedelta(days=1), time.min) # team of this manager team_ids = [ e.id for e in Employee.query.filter( Employee.reporting_manager_id == manager_id, status_allowed_filter(Employee), # ✅ handles active + notice period variants ).all() ] if not team_ids: punches = [] else: punches = ( BiometricPunch.query.options(joinedload(BiometricPunch.employee)) .filter( BiometricPunch.employee_id.in_(team_ids), BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt, BiometricPunch.source_name == "self", BiometricPunch.approval_status == "pending", ) .order_by(BiometricPunch.punch_time.asc()) .all() ) # list of managers (for HR/Admin dropdown) managers = [] if is_hr_admin: managers = ( Employee.query.filter(status_allowed_filter(Employee)) .order_by(Employee.full_name) .all() ) return render_template( "employee/self_punch_approvals.html", punches=punches, filter_date=filter_date, is_hr_admin=is_hr_admin, managers=managers, selected_manager_id=manager_id, ) @employee_bp.route("/self-punch/approvals//approve", methods=["POST"]) @login_required def approve_self_punch(punch_id): punch = BiometricPunch.query.get_or_404(punch_id) emp = punch.employee role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() is_hr_admin = role_name in ("hr", "hr admin", "admin") if not is_hr_admin and emp.reporting_manager_id != current_user.id: flash("You are not authorised to approve this punch.", "danger") return redirect(url_for("employee.self_punch_approvals")) punch.approval_status = "approved" punch.approved_by_id = current_user.id punch.approved_on = datetime.utcnow() punch.rejection_reason = None db.session.commit() # ✅ Immediately recalc attendance for that date try: from attendance_engine import recalc_attendance_for_day recalc_attendance_for_day(emp, punch.punch_time.date()) flash("✅ Punch approved and attendance updated.", "success") except Exception as e: current_app.logger.exception("Attendance recalc failed after self punch approval") flash("Punch approved, but attendance recalculation failed. Please run attendance engine.", "warning") return redirect(url_for("employee.self_punch_approvals", date=punch.punch_time.date().strftime("%Y-%m-%d"))) @employee_bp.route("/self-punch/approvals//reject", methods=["POST"]) @login_required def reject_self_punch(punch_id): punch = BiometricPunch.query.get_or_404(punch_id) emp = punch.employee role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() is_hr_admin = role_name in ("hr", "hr admin", "admin") if not is_hr_admin and emp.reporting_manager_id != current_user.id: flash("You are not authorised to reject this punch.", "danger") return redirect(url_for("employee.self_punch_approvals")) reason = (request.form.get("reason") or "").strip() punch.approval_status = "rejected" punch.approved_by_id = current_user.id punch.approved_on = datetime.utcnow() punch.rejection_reason = reason or None db.session.commit() flash("Punch rejected. It will not be counted in attendance.", "warning") return redirect(url_for("employee.self_punch_approvals", date=punch.punch_time.date().strftime("%Y-%m-%d"))) # --- HR/Admin: Self Punch Backup Approvals (ALL employees) -------------------- from sqlalchemy.orm import joinedload from sqlalchemy import or_ from datetime import datetime, date, time, timedelta def _is_hr_admin_user() -> bool: role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() return role_name in ("hr", "hr admin", "admin", "demo admin") @employee_bp.route("/self-punch/approvals-admin", methods=["GET"]) @login_required def self_punch_approvals_admin(): if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) # ---------------- Filters ---------------- q = (request.args.get("q") or "").strip() # name/code/email search status = (request.args.get("status") or "pending").strip().lower() # pending|approved|rejected|all direction = (request.args.get("direction") or "").strip().lower() # in|out manager_id = request.args.get("manager_id", type=int) # reporting manager filter dept_id = request.args.get("department_id", type=int) date_from_str = (request.args.get("date_from") or "").strip() date_to_str = (request.args.get("date_to") or "").strip() page = request.args.get("page", 1, type=int) per_page = request.args.get("per_page", 20, type=int) # default: today today = date.today() def _parse_date(s: str): try: return datetime.strptime(s, "%Y-%m-%d").date() except Exception: return None d_from = _parse_date(date_from_str) or today d_to = _parse_date(date_to_str) or d_from if d_to < d_from: d_from, d_to = d_to, d_from start_dt = datetime.combine(d_from, time.min) end_dt = datetime.combine(d_to + timedelta(days=1), time.min) # ---------------- Query ---------------- query = ( BiometricPunch.query .options(joinedload(BiometricPunch.employee)) .filter( BiometricPunch.source_name == "self", BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt, ) ) if status != "all": query = query.filter(BiometricPunch.approval_status == status) if direction in ("in", "out"): query = query.filter(BiometricPunch.direction == direction) if manager_id: query = query.join(Employee, Employee.id == BiometricPunch.employee_id)\ .filter(Employee.reporting_manager_id == manager_id) if dept_id: query = query.join(Employee, Employee.id == BiometricPunch.employee_id)\ .filter(Employee.department_id == dept_id) if q: like = f"%{q}%" query = query.join(Employee, Employee.id == BiometricPunch.employee_id)\ .filter(or_( Employee.full_name.ilike(like), Employee.employee_code.ilike(like), Employee.email.ilike(like), )) query = query.order_by(BiometricPunch.punch_time.asc()) pagination = query.paginate(page=page, per_page=per_page, error_out=False) punches = pagination.items # dropdown data managers = (Employee.query .filter(Employee.status.in_(["active", "notice_period"])) .order_by(Employee.full_name.asc()).all()) departments = Department.query.order_by(Department.name.asc()).all() # page number window total_pages = pagination.pages or 1 start = max(1, pagination.page - 2) end = min(total_pages, pagination.page + 2) page_numbers = list(range(start, end + 1)) return render_template( "employee/self_punch_approvals_admin.html", punches=punches, pagination=pagination, page_numbers=page_numbers, per_page=per_page, # echo filters q=q, status=status, direction=direction, manager_id=manager_id, dept_id=dept_id, date_from=d_from.strftime("%Y-%m-%d"), date_to=d_to.strftime("%Y-%m-%d"), managers=managers, departments=departments, ) @employee_bp.route("/self-punch/approvals-admin/bulk", methods=["POST"]) @login_required def self_punch_approvals_admin_bulk(): if not _is_hr_admin_user(): flash("Access denied.", "danger") return redirect(url_for("employee.dashboard")) action = (request.form.get("action") or "").strip().lower() # approve | reject reason = (request.form.get("reason") or "").strip() approve_all_filtered = (request.form.get("approve_all_filtered") or "").strip() == "1" selected_ids = request.form.getlist("punch_ids") # checkbox list # If no manual selection and user clicked "Approve All (Filtered)" / "Reject All (Filtered)" if approve_all_filtered and not selected_ids: # Rebuild the same filtered query from hidden inputs q = (request.form.get("q") or "").strip() status = (request.form.get("status") or "pending").strip().lower() # pending|approved|rejected|all direction = (request.form.get("direction") or "").strip().lower() # in|out|"" manager_id = request.form.get("manager_id", type=int) dept_id = request.form.get("department_id", type=int) date_from_str = (request.form.get("date_from") or "").strip() date_to_str = (request.form.get("date_to") or "").strip() today = date.today() def _parse_date(s: str): try: return datetime.strptime(s, "%Y-%m-%d").date() except Exception: return None d_from = _parse_date(date_from_str) or today d_to = _parse_date(date_to_str) or d_from if d_to < d_from: d_from, d_to = d_to, d_from start_dt = datetime.combine(d_from, time.min) end_dt = datetime.combine(d_to + timedelta(days=1), time.min) query = ( BiometricPunch.query .options(joinedload(BiometricPunch.employee)) .filter( BiometricPunch.source_name == "self", BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt, ) ) # For bulk actions, "all" means no status filter; otherwise respect it if status != "all": query = query.filter(BiometricPunch.approval_status == status) if direction in ("in", "out"): query = query.filter(BiometricPunch.direction == direction) if manager_id: query = query.join(Employee, Employee.id == BiometricPunch.employee_id)\ .filter(Employee.reporting_manager_id == manager_id) if dept_id: query = query.join(Employee, Employee.id == BiometricPunch.employee_id)\ .filter(Employee.department_id == dept_id) if q: like = f"%{q}%" query = query.join(Employee, Employee.id == BiometricPunch.employee_id)\ .filter(or_( Employee.full_name.ilike(like), Employee.employee_code.ilike(like), Employee.email.ilike(like), )) # ✅ Only act on PENDING by default even if someone filtered "all" (safety) # If you want true-all behavior, comment this out. query = query.filter(BiometricPunch.approval_status == "pending") selected_ids = [str(p.id) for p in query.all()] if action not in ("approve", "reject"): flash("Invalid bulk action.", "danger") return redirect(request.referrer or url_for("employee.self_punch_approvals_admin")) if not selected_ids: flash("Please select at least one punch (or use Approve All Filtered).", "warning") return redirect(request.referrer or url_for("employee.self_punch_approvals_admin")) # 1) Load the selected punches (just to know which employee+date combos are affected) selected_punches = (BiometricPunch.query .options(joinedload(BiometricPunch.employee)) .filter( BiometricPunch.id.in_(selected_ids), BiometricPunch.source_name == "self" ) .all() ) if not selected_punches: flash("No matching punches found.", "warning") return redirect(request.referrer or url_for("employee.self_punch_approvals_admin")) # 2) Build set of (employee_id, date) keys touched touched_keys = {(p.employee_id, p.punch_time.date()) for p in selected_punches} # 3) Expand: fetch ALL self punches for those employee+date keys (so IN + OUT both get approved) # Only pending ones are actionable. expanded_punches = [] for (emp_id, day) in touched_keys: start_dt = datetime.combine(day, time.min) end_dt = datetime.combine(day + timedelta(days=1), time.min) day_punches = (BiometricPunch.query .filter( BiometricPunch.employee_id == emp_id, BiometricPunch.source_name == "self", BiometricPunch.punch_time >= start_dt, BiometricPunch.punch_time < end_dt, BiometricPunch.approval_status == "pending", ) .all() ) expanded_punches.extend(day_punches) # If user chose reject, you may want to reject ONLY selected punches (not expand). # For approve, expansion is what fixes LWP. punches_to_update = expanded_punches if action == "approve" else selected_punches updated = 0 recalc_fail = 0 for p in punches_to_update: # safety if (p.approval_status or "").lower() != "pending": continue if action == "approve": p.approval_status = "approved" p.rejection_reason = None else: p.approval_status = "rejected" p.rejection_reason = reason or None p.approved_by_id = current_user.id p.approved_on = datetime.utcnow() updated += 1 db.session.commit() # Attendance recalc only for approvals if action == "approve" and updated: from attendance_engine import recalc_attendance_for_day for (emp_id, day) in touched_keys: try: emp = Employee.query.get(emp_id) # fresh from DB/session recalc_attendance_for_day(emp, day) except Exception: recalc_fail += 1 current_app.logger.exception("Attendance recalc failed in bulk approval") if action == "approve": msg = f"✅ Approved {updated} punch(es)." if recalc_fail: msg += f" Attendance recalc failed for {recalc_fail} day(s)." flash(msg, "success" if not recalc_fail else "warning") else: flash(f"✅ Rejected {updated} punch(es).", "success") return redirect(request.referrer or url_for("employee.self_punch_approvals_admin")) @employee_bp.route('/update-mandatory-profile', methods=['POST']) @login_required def update_mandatory_profile(): import re from datetime import datetime from flask import current_app dob = request.form.get('dob') or None gender = (request.form.get('gender') or '').strip() marital_status = (request.form.get('marital_status') or '').strip() father_spouse_name = (request.form.get('father_spouse_name') or '').strip() emergency_contact_name = (request.form.get('emergency_contact_name') or '').strip() emergency_contact_number = (request.form.get('emergency_contact_number') or '').strip() address = (request.form.get('address') or '').strip() temporary_address = (request.form.get('temporary_address') or '').strip() aadhar_name = (request.form.get('aadhar_name') or '').strip() aadhar_number = (request.form.get('aadhar_number') or '').strip() pan_name = (request.form.get('pan_name') or '').strip() pan_number = (request.form.get('pan_number') or '').strip().upper() bank_account_name = (request.form.get('bank_account_name') or '').strip() bank_account_number = (request.form.get('bank_account_number') or '').strip() ifsc_code = (request.form.get('ifsc_code') or '').strip().upper() bank_name = (request.form.get('bank_name') or '').strip() required = [ dob, gender, marital_status, father_spouse_name, emergency_contact_name, emergency_contact_number, address, temporary_address, aadhar_name, aadhar_number, pan_name, pan_number, bank_account_name, bank_account_number, ifsc_code, bank_name ] if any(v is None or (isinstance(v, str) and not v.strip()) for v in required): flash("❌ Please fill all mandatory fields.", "danger") return redirect(request.referrer or url_for('dashboard.index')) # ---- Validations BEFORE setting current_user ---- if not re.match(r'^[A-Z]{5}[0-9]{4}[A-Z]$', pan_number): flash("❌ Invalid PAN format (ABCDE1234F).", "danger") return redirect(request.referrer or url_for('dashboard.index')) if not re.match(r'^\d{12}$', aadhar_number): flash("❌ Aadhaar must be 12 digits.", "danger") return redirect(request.referrer or url_for('dashboard.index')) if not re.match(r'^[A-Z]{4}0[A-Z0-9]{6}$', ifsc_code): flash("❌ Invalid IFSC code.", "danger") return redirect(request.referrer or url_for('dashboard.index')) if not re.match(r'^\d{10}$', emergency_contact_number): flash("❌ Emergency contact must be 10 digits.", "danger") return redirect(request.referrer or url_for('dashboard.index')) try: current_user.dob = datetime.strptime(dob, "%Y-%m-%d").date() current_user.gender = gender current_user.marital_status = marital_status current_user.father_spouse_name = father_spouse_name current_user.emergency_contact_name = emergency_contact_name current_user.emergency_contact_number = emergency_contact_number current_user.address = address current_user.temporary_address = temporary_address current_user.aadhar_name = aadhar_name current_user.aadhar_number = aadhar_number current_user.pan_name = pan_name current_user.pan_number = pan_number current_user.bank_account_name = bank_account_name current_user.bank_account_number = bank_account_number current_user.ifsc_code = ifsc_code current_user.bank_name = bank_name # ---- Audit log: best-effort, never block saving ---- try: log = EmployeeChangeLog( employee_id=current_user.id, changed_by=current_user.id, change_type="Mandatory Profile Completion", old_value="Incomplete mandatory profile", new_value="Mandatory profile completed", ) db.session.add(log) except Exception: current_app.logger.exception("EmployeeChangeLog insert failed (field mismatch). Continuing without audit log.") db.session.commit() flash("✅ Profile updated successfully.", "success") except Exception: db.session.rollback() current_app.logger.exception("Mandatory profile save failed") flash("❌ Could not save your profile. Please try again.", "danger") nxt = request.form.get("next") or url_for('dashboard.index') return redirect(nxt)