# routes/biometric_views.py from __future__ import annotations from datetime import datetime from flask import Blueprint, render_template, request, redirect, url_for, flash from flask_login import login_required, current_user from extensions import db from models import BiometricSyncRun, BiometricPunch, Employee,BiometricEmployeeMap from routes.biometric import sync_biometric_for_date biometric_bp = Blueprint("biometric", __name__, url_prefix="/biometric") def _is_admin_or_hr(user) -> bool: try: role_name = (user.role.name or "").lower() return role_name in ("admin", "hr admin", "hr") except Exception: return False @biometric_bp.route("/logs", methods=["GET", "POST"]) @login_required def biometric_logs(): if not _is_admin_or_hr(current_user): flash("Access denied.", "danger") return redirect(url_for("dashboard.index")) # Manual trigger from same page (optional) if request.method == "POST": date_str = request.form.get("sync_date") or datetime.utcnow().strftime("%Y-%m-%d") try: target_date = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: flash("Invalid date format. Use YYYY-MM-DD.", "danger") return redirect(url_for("biometric.biometric_logs")) run = sync_biometric_for_date(target_date) flash(f"Biometric sync: {run.status} – {run.message}", "success" if run.status == "success" else "danger") return redirect(url_for("biometric.biometric_logs")) # Filters status = (request.args.get("status") or "").strip() page = request.args.get("page", 1, type=int) per_page = 20 q = BiometricSyncRun.query.order_by(BiometricSyncRun.started_at.desc()) if status: q = q.filter(BiometricSyncRun.status == status) pagination = q.paginate(page=page, per_page=per_page, error_out=False) runs = pagination.items # simple stats total_runs = q.count() last_run = runs[0] if runs else None return render_template( "biometric_logs.html", runs=runs, pagination=pagination, status=status, total_runs=total_runs, last_run=last_run, ) @biometric_bp.route("/logs/") @login_required def biometric_log_detail(run_id): if not _is_admin_or_hr(current_user): flash("Access denied.", "danger") return redirect(url_for("dashboard.index")) run = BiometricSyncRun.query.get_or_404(run_id) # filters inside this run emp_search = (request.args.get("emp") or "").strip() punches_q = BiometricPunch.query.filter(BiometricPunch.sync_run_id == run.id) if emp_search: punches_q = punches_q.join(Employee, BiometricPunch.employee_id == Employee.id)\ .filter( (Employee.full_name.ilike(f"%{emp_search}%")) | (Employee.employee_code.ilike(f"%{emp_search}%")) ) punches = punches_q.order_by(BiometricPunch.punch_time.asc()).all() # group by employee for easier display grouped = {} for p in punches: key = p.employee_id or p.external_employee_code if key not in grouped: grouped[key] = { "employee": p.employee, "external_code": p.external_employee_code, "punches": [], } grouped[key]["punches"].append(p) return render_template( "biometric_log_detail.html", run=run, grouped=grouped, emp_search=emp_search, ) @biometric_bp.route("/mapping", methods=["GET", "POST"]) @login_required def biometric_mapping(): if not _is_admin_or_hr(current_user): flash("Access denied.", "danger") return redirect(url_for("dashboard.index")) # Handle add / update if request.method == "POST": mapping_id = request.form.get("mapping_id") employee_id = request.form.get("employee_id") external_code = (request.form.get("external_code") or "").strip() source_name = (request.form.get("source_name") or "default").strip() or "default" is_active = bool(request.form.get("is_active")) if not (employee_id and external_code): flash("Employee and card number are required.", "danger") return redirect(url_for("biometric.biometric_mapping")) try: employee_id_int = int(employee_id) except ValueError: flash("Invalid employee selected.", "danger") return redirect(url_for("biometric.biometric_mapping")) try: if mapping_id: # update mapping = BiometricEmployeeMap.query.get(int(mapping_id)) if not mapping: flash("Mapping not found.", "danger") return redirect(url_for("biometric.biometric_mapping")) mapping.employee_id = employee_id_int mapping.external_code = external_code mapping.source_name = source_name mapping.is_active = is_active msg = "Mapping updated." else: # create mapping = BiometricEmployeeMap( employee_id=employee_id_int, external_code=external_code, source_name=source_name, is_active=is_active, ) db.session.add(mapping) msg = "Mapping added." db.session.commit() flash(msg, "success") except IntegrityError: db.session.rollback() flash("A mapping for this card number & source already exists.", "danger") except Exception as e: db.session.rollback() current_app.logger.exception("Biometric mapping save failed") flash(f"Save failed: {e}", "danger") return redirect(url_for("biometric.biometric_mapping")) # --- GET: show list + (optional) edit row ---------------------- edit_id = request.args.get("edit_id", type=int) edit_mapping = BiometricEmployeeMap.query.get(edit_id) if edit_id else None # You can add filters later if you want (by department, etc.) mappings = (BiometricEmployeeMap.query .join(Employee, BiometricEmployeeMap.employee_id == Employee.id) .order_by(Employee.full_name.asc()) .all()) employees = Employee.query.order_by(Employee.full_name.asc()).all() return render_template( "biometric_mapping.html", mappings=mappings, employees=employees, edit_mapping=edit_mapping, ) @biometric_bp.post("/mapping/delete/") @login_required def biometric_mapping_delete(mapping_id): if not _is_admin_or_hr(current_user): flash("Access denied.", "danger") return redirect(url_for("dashboard.index")) mapping = BiometricEmployeeMap.query.get_or_404(mapping_id) try: db.session.delete(mapping) db.session.commit() flash("Mapping deleted.", "info") except Exception as e: db.session.rollback() flash(f"Delete failed: {e}", "danger") return redirect(url_for("biometric.biometric_mapping"))