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 from extensions import db import bcrypt from models import Employee, Department, Designation, EmployeeChangeLog, PublicLink, PendingEmployee, PendingApprovalLog, BroadcastMessage, Role, Holiday,LeaveRequest from sqlalchemy import extract,func from sqlalchemy.exc import IntegrityError from common_utils import employee_required from typing import Optional 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 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) @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') 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: resume_filename = secure_filename(resume.filename) resume_path = os.path.join(uploads_dir, resume_filename) resume.save(resume_path) if id_proof: 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) new_employee = Employee( full_name=full_name, 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( Employee.email == email, Employee.id != employee.id ).first() if existing_email_user: errors["email"] = "Email already exists for another 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") # Dates employee.date_of_joining = val("date_of_joining") or None employee.dob = val("dob") or None employee.last_company_join_date = val("last_company_doj") or None employee.last_company_exit_date = val("last_company_doe") or None # 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 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 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)) # --- 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)) # For filter dropdowns departments = Department.query.order_by(Department.name).all() roles = Role.query.order_by(Role.name).all() return render_template( 'employee/view_employees.html', employees=employees, roles=roles, departments=departments, # pagination helpers pagination=pagination, page_numbers=page_numbers, # echo filters name=name, code=code, dept_id=dept_id, role_name=role_name, status=status, 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']: flash('Access denied', 'danger') return redirect(url_for('dashboard.index')) 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 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_by(email=email).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) # --- Manager Team Attendance (month-wise) ------------------------------------- from calendar import monthrange from sqlalchemy.orm import joinedload from models import Attendance, LeaveRequest, Holiday, WeekendSettings @employee_bp.route('/team-attendance', methods=['GET']) @login_required def team_attendance(): """ Managers see month-wise attendance for direct reports. HR/Admin can choose any manager via ?manager_id= Filters: ?month=YYYY-MM, ?q= """ # ---- Parse inputs ---- month_str = (request.args.get("month") or "").strip() q = (request.args.get("q") or "").strip() manager_id_arg = request.args.get("manager_id", type=int) # Determine the "viewer" scope role_name = (getattr(getattr(current_user, "role", None), "name", "") or "").lower() is_hr_admin = role_name in ("hr", "hr admin", "admin") # Manager whose team is being viewed if is_hr_admin and manager_id_arg: manager_id = manager_id_arg else: manager_id = current_user.id # Month window (defaults to current month) today = date.today() if month_str: try: y, m = map(int, month_str.split("-")) except Exception: flash("Invalid month format. Use YYYY-MM.", "danger") return redirect(url_for("employee.team_attendance")) else: y, m = today.year, today.month month_str = f"{y:04d}-{m:02d}" first_day = date(y, m, 1) last_day = date(y, m, monthrange(y, m)[1]) # ---- Weekend settings ---- weekend = WeekendSettings.query.order_by(WeekendSettings.updated_on.desc()).first() sunday_off = bool(weekend and (weekend.sunday == "off")) def is_saturday_off(d: date) -> bool: if not weekend: return False week_number = (d.day - 1) // 7 + 1 # 1..5 bucket = ["first","second","third","fourth","fifth"][week_number - 1] return getattr(weekend, f"{bucket}_saturday", None) == "off" # ---- Fetch team (direct reports) ---- team_q = (Employee.query .options(joinedload(Employee.department), joinedload(Employee.designation)) .filter(Employee.reporting_manager_id == manager_id)) # Optional name filter if q: team_q = team_q.filter(Employee.full_name.ilike(f"%{q}%")) # (Default) Show active + notice employees. Tweak if you want resigned as well. team_q = team_q.filter(Employee.status.in_(["active", "notice_period"])) team = team_q.order_by(Employee.full_name.asc()).all() # Preload holidays for the month holiday_dates = { h.date for h in Holiday.query.filter( Holiday.date >= first_day, Holiday.date <= last_day ).all() } # Build per-employee month map all_days = [first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1)] results = [] # [{ employee, counts, rows }, ...] for emp in team: # Clip employment window if needed doj = getattr(emp, "date_of_joining", None) start = max(first_day, doj) if doj else first_day end = last_day # Approved leave date set leave_dates = set() approved = (LeaveRequest.query .filter_by(employee_id=emp.id, status="final_approved") .filter(LeaveRequest.start_date <= end, LeaveRequest.end_date >= start) .all()) for lr in approved: cur = max(lr.start_date, start) stop = min(lr.end_date, end) while cur <= stop: leave_dates.add(cur) cur += timedelta(days=1) # Attendance map for the month att_map = { a.date: (a.remarks or "").strip() for a in Attendance.query .filter(Attendance.employee_id == emp.id, Attendance.date >= start, Attendance.date <= end) .all() } # Rollup counters present = half = leave = holiday = weekend_cnt = lwp = 0 day_rows = [] # [{date, code, label, icon_class, tooltip}, ...] for d in all_days: # Before DOJ -> skip display (show as N/A) if doj and d < doj: day_rows.append({ "date": d, "code": "NA", "label": "Not Joined", "icon": "bi-dash-lg text-muted", "tooltip": "Not joined yet" }) continue # Determine base flags wk = d.weekday() # 0..6 is_weekend = (wk == 6 and sunday_off) or (wk == 5 and is_saturday_off(d)) is_holiday = d in holiday_dates is_leave = d in leave_dates att = att_map.get(d) # could be Present/Half Day/Leave/Holiday/LWP/Weekend/None # Priority: Holiday > Weekend > Approved Leave > Attendance > LWP if is_holiday: holiday += 1 day_rows.append({ "date": d, "code": "H", "label": "Holiday", "icon": "bi-flag-fill text-info", "tooltip": "Holiday" }) continue if is_weekend: weekend_cnt += 1 day_rows.append({ "date": d, "code": "W", "label": "Weekend", "icon": "bi-calendar-week-fill text-secondary", "tooltip": "Weekend" }) continue if is_leave: leave += 1 day_rows.append({ "date": d, "code": "L", "label": "Leave", "icon": "bi-airplane-fill text-warning", "tooltip": "Approved leave" }) continue # Attendance-based if att == "Present": present += 1 day_rows.append({ "date": d, "code": "P", "label": "Present", "icon": "bi-check-circle-fill text-success", "tooltip": "Present" }) elif att == "Half Day": half += 1 day_rows.append({ "date": d, "code": "HD", "label": "Half Day", "icon": "bi-circle-half text-primary", "tooltip": "Half Day" }) elif att == "Leave": leave += 1 day_rows.append({ "date": d, "code": "L", "label": "Leave", "icon": "bi-airplane-fill text-warning", "tooltip": "Leave (attendance)" }) elif att == "Holiday": holiday += 1 day_rows.append({ "date": d, "code": "H", "label": "Holiday", "icon": "bi-flag-fill text-info", "tooltip": "Holiday (attendance)" }) elif (att or "").lower() == "weekend": weekend_cnt += 1 day_rows.append({ "date": d, "code": "W", "label": "Weekend", "icon": "bi-calendar-week-fill text-secondary", "tooltip": "Weekend (attendance)" }) else: lwp += 1 day_rows.append({ "date": d, "code": "A", "label": "LWP", "icon": "bi-x-circle-fill text-danger", "tooltip": "LWP / Absent" }) results.append({ "employee": emp, "counts": { "present": present + 0.5 * half, "present_full": present, "half": half, "leave": leave, "holiday": holiday, "weekend": weekend_cnt, "lwp": lwp, "total_paid": present + 0.5 * half + leave + holiday + weekend_cnt, }, "days": day_rows, "month_str": month_str }) # Managers list for HR/Admin dropdown managers = [] if is_hr_admin: managers = (Employee.query .filter(Employee.id != current_user.id) .order_by(Employee.full_name).all()) return render_template( "employee/team_attendance.html", month=month_str, results=results, query=q, is_hr_admin=is_hr_admin, selected_manager_id=manager_id, managers=managers ) # --- 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() 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 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)) 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) # Keep only mandatory columns for import template? -> No; this is full export. # The import flow already uses a small mandatory set via _expected_columns(). 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