from flask import Blueprint, render_template, request, redirect, url_for, flash from datetime import datetime, timedelta, date from sqlalchemy import and_, func, or_ from flask_login import login_required, current_user from extensions import db from models import ( Employee, Attendance, Holiday, LeaveRequest, DesktimeLog, SyncLog, ) from cli.attendance_cli import fetch_and_store_desktime import json from collections import defaultdict desktime_logs_bp = Blueprint("desktime_logs", __name__, url_prefix="/desktime") # --------------------------------------------------------- # 1. Manual Desktime re-process page (existing behaviour) # --------------------------------------------------------- @desktime_logs_bp.route("/logs", methods=["GET", "POST"]) @login_required def desktime_logs(): """ UI page to manually (re)fetch Desktime data for selected employees for a specific date or full month. Also shows a table of the attendance rows that were updated in this run. """ employees = ( Employee.query.filter(Employee.desktime_id.isnot(None)) .order_by(Employee.full_name) .all() ) summary = None logs = [] if request.method == "POST": selected_emp_ids = request.form.getlist("employee_ids") specific_date_str = request.form.get("specific_date") # YYYY-MM-DD month_str = request.form.get("month") # YYYY-MM if not selected_emp_ids: flash("Please select at least one employee.", "danger") return redirect(url_for("desktime_logs.desktime_logs")) if not specific_date_str and not month_str: flash("Please select either a date or a month.", "danger") return redirect(url_for("desktime_logs.desktime_logs")) # ---------------------------- # Determine date range # ---------------------------- if specific_date_str: try: d = datetime.strptime(specific_date_str, "%Y-%m-%d").date() except ValueError: flash("Invalid date format.", "danger") return redirect(url_for("desktime_logs.desktime_logs")) start_date = end_date = d else: try: m = datetime.strptime(month_str, "%Y-%m") except ValueError: flash("Invalid month format.", "danger") return redirect(url_for("desktime_logs.desktime_logs")) year = m.year month = m.month start_date = date(year, month, 1) # compute first day of next month safely if month == 12: next_month_first = date(year + 1, 1, 1) else: next_month_first = date(year, month + 1, 1) end_date = next_month_first - timedelta(days=1) updated_records = 0 errors = [] updated_attendance_rows = [] today = datetime.today().date() for emp_id in selected_emp_ids: emp = Employee.query.get(emp_id) if not emp or not emp.desktime_id: continue curr_date = start_date while curr_date <= end_date and curr_date <= today: # Delete existing attendance + desktime log for that day Attendance.query.filter_by(employee_id=emp.id, date=curr_date).delete() DesktimeLog.query.filter_by(employee_id=emp.id, date=curr_date).delete() db.session.commit() try: hours = fetch_and_store_desktime(emp, curr_date) # Basic rule for remark based on hours if hours >= 7: remarks = "Present" elif hours >= 4: remarks = "Half Day" else: remarks = "LWP" # Override to Holiday/Leave if exists holiday = Holiday.query.filter_by(date=curr_date).first() leave = ( LeaveRequest.query.filter_by( employee_id=emp.id, status="final_approved" ) .filter( LeaveRequest.start_date <= curr_date, LeaveRequest.end_date >= curr_date, ) .first() ) if holiday: remarks = "Holiday" elif leave: remarks = "Leave" attendance = Attendance( employee_id=emp.id, date=curr_date, check_in_time=None, check_out_time=None, working_hours=hours, remarks=remarks, source="Desktime", ) db.session.add(attendance) db.session.commit() updated_attendance_rows.append(attendance) updated_records += 1 except Exception as e: # noqa: BLE001 errors.append(f"{emp.full_name} on {curr_date}: {str(e)}") curr_date += timedelta(days=1) summary = { "total_updated": updated_records, "errors": errors, } logs = updated_attendance_rows return render_template( "attendance/desktime_logs.html", employees=employees, summary=summary, logs=logs, ) # --------------------------------------------------------- # 2. Overview page – one row per employee (you already saw) # --------------------------------------------------------- @desktime_logs_bp.route("/overview") @login_required def desktime_overview(): """ Overview page: one row per employee with desktime_id, showing the latest DesktimeLog row (last log date, activity, productive hours, and when data was fetched). """ search = request.args.get("search", "").strip() emp_q = Employee.query.filter(Employee.desktime_id.isnot(None)) if search: emp_q = emp_q.filter( or_( Employee.full_name.ilike(f"%{search}%"), Employee.employee_code.ilike(f"%search%"), ) ) # small fix: correct ilike pattern for code if search: emp_q = Employee.query.filter( Employee.desktime_id.isnot(None), or_( Employee.full_name.ilike(f"%{search}%"), Employee.employee_code.ilike(f"%{search}%"), ), ) employees = emp_q.order_by(Employee.full_name).all() # Subquery: latest log DATE per employee subq = ( db.session.query( DesktimeLog.employee_id.label("emp_id"), func.max(DesktimeLog.date).label("last_log_date"), ) .group_by(DesktimeLog.employee_id) .subquery() ) # Join to get the actual DesktimeLog row for that last_log_date latest_logs = ( db.session.query(DesktimeLog) .join( subq, and_( DesktimeLog.employee_id == subq.c.emp_id, DesktimeLog.date == subq.c.last_log_date, ), ) .all() ) latest_by_emp = {log.employee_id: log for log in latest_logs} rows = [] for emp in employees: rows.append( { "employee": emp, "log": latest_by_emp.get(emp.id), } ) return render_template( "attendance/desktime_overview.html", rows=rows, search=search, ) # --------------------------------------------------------- # 3. NEW – Per-employee daily details page # --------------------------------------------------------- @desktime_logs_bp.route("/employee/") @login_required def desktime_employee_logs(employee_id: int): """ Shows day-by-day Desktime data for a single employee, for a selected month. Includes first/last activity, productive hours and mapped attendance info. """ emp = Employee.query.get_or_404(employee_id) # Month selection (YYYY-MM) month_str = request.args.get("month", "").strip() if month_str: try: m = datetime.strptime(month_str, "%Y-%m") year, month = m.year, m.month except ValueError: # fallback to latest log month if invalid month_str = "" year = month = None else: year = month = None # If month not provided / invalid, use month of latest log, else current month if not year or not month: latest_log = ( DesktimeLog.query.filter_by(employee_id=employee_id) .order_by(DesktimeLog.date.desc()) .first() ) if latest_log: year = latest_log.date.year month = latest_log.date.month else: today = date.today() year, month = today.year, today.month selected_month = f"{year:04d}-{month:02d}" start_date = date(year, month, 1) if month == 12: next_month_first = date(year + 1, 1, 1) else: next_month_first = date(year, month + 1, 1) end_date = next_month_first - timedelta(days=1) # Query DesktimeLog + Attendance for the range q = ( db.session.query(DesktimeLog, Attendance) .outerjoin( Attendance, and_( Attendance.employee_id == DesktimeLog.employee_id, Attendance.date == DesktimeLog.date, ), ) .filter( DesktimeLog.employee_id == employee_id, DesktimeLog.date >= start_date, DesktimeLog.date <= end_date, ) .order_by(DesktimeLog.date) ) rows = q.all() return render_template( "attendance/desktime_employee_logs.html", employee=emp, rows=rows, selected_month=selected_month, ) # --------------------------------------------------------- # 4. Sync history page # --------------------------------------------------------- @desktime_logs_bp.route("/sync-history") @login_required def view_sync_logs(): """ Shows the last 100 sync batch runs recorded in SyncLog. """ logs = SyncLog.query.order_by(SyncLog.started_at.desc()).limit(100).all() return render_template("desktime_logs/logs.html", logs=logs) @desktime_logs_bp.route("/dashboard/") @login_required def desktime_employee_dashboard(employee_id: int): """ Desktime-style dashboard for a single employee. Shows: - Monthly summary cards - Daily productivity bar + table - Aggregated productive / unproductive apps for month - Optional day view (mini dashboard for a single date) """ import json from collections import defaultdict emp = Employee.query.get_or_404(employee_id) # Month selection (YYYY-MM) and optional day selection (YYYY-MM-DD) month_str = request.args.get("month", "").strip() day_str = request.args.get("day", "").strip() # Helpers ------------------------------------------------- def sec_to_hm(seconds): if not seconds or seconds <= 0: return "0h 0m" s = int(seconds) h = s // 3600 m = (s % 3600) // 60 return f"{h}h {m}m" def time_str_from_ts(ts): if not ts: return None try: t = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S").time() return t.strftime("%H:%M") except Exception: return None def _default_year_month(): latest_log = ( DesktimeLog.query.filter_by(employee_id=employee_id) .order_by(DesktimeLog.date.desc()) .first() ) if latest_log: return latest_log.date.year, latest_log.date.month today = date.today() return today.year, today.month # ---- Resolve month ---- if month_str: try: m = datetime.strptime(month_str, "%Y-%m") year, month = m.year, m.month except ValueError: year, month = _default_year_month() else: year, month = _default_year_month() selected_month = f"{year:04d}-{month:02d}" # Date range for the month start_date = date(year, month, 1) if month == 12: next_month_first = date(year + 1, 1, 1) else: next_month_first = date(year, month + 1, 1) end_date = next_month_first - timedelta(days=1) # ---- Monthly logs ---- logs = ( DesktimeLog.query.filter( DesktimeLog.employee_id == employee_id, DesktimeLog.date >= start_date, DesktimeLog.date <= end_date, ) .order_by(DesktimeLog.date) .all() ) # Aggregates ---------------------------------------------- total_productive = 0 total_desktime = 0 total_at_work = 0 total_online = 0 total_offline = 0 arrival_minutes = [] left_minutes = [] productivity_values = [] efficiency_values = [] apps_agg = { "productive": defaultdict(int), "neutral": defaultdict(int), "unproductive": defaultdict(int), } daily_rows = [] for log in logs: day_info = { "date": log.date, "arrival": None, "left": None, "productive_sec": 0, "desktime_sec": 0, "at_work_sec": 0, "productivity": None, "efficiency": None, } data = {} if log.raw_data: try: data = json.loads(log.raw_data) except Exception: data = {} # Arrival / left times -------------------------------- arrival = time_str_from_ts(data.get("arrived")) left = time_str_from_ts(data.get("left")) day_info["arrival"] = arrival day_info["left"] = left if arrival: h, m = map(int, arrival.split(":")) arrival_minutes.append(h * 60 + m) if left: h, m = map(int, left.split(":")) left_minutes.append(h * 60 + m) # Time values (seconds) ------------------------------- prod_sec = data.get("productiveTime") or 0 desk_sec = data.get("desktimeTime") or 0 at_work_sec = data.get("atWorkTime") or 0 online_sec = data.get("onlineTime") or 0 offline_sec = data.get("offlineTime") or 0 day_info["productive_sec"] = prod_sec day_info["desktime_sec"] = desk_sec day_info["at_work_sec"] = at_work_sec total_productive += prod_sec total_desktime += desk_sec total_at_work += at_work_sec total_online += online_sec total_offline += offline_sec # Productivity / efficiency --------------------------- prod_pct = data.get("productivity") eff_pct = data.get("efficiency") # Bar height: use productivity % if available, otherwise fallback to time ratio if isinstance(prod_pct, (int, float)): bar_ratio = float(prod_pct) productivity_values.append(float(prod_pct)) elif at_work_sec > 0: bar_ratio = (prod_sec / at_work_sec) * 100.0 else: bar_ratio = 0.0 # clamp 0–100 bar_ratio = max(0.0, min(100.0, bar_ratio)) if isinstance(eff_pct, (int, float)): efficiency_values.append(float(eff_pct)) day_info["productivity"] = prod_pct day_info["efficiency"] = eff_pct # Apps aggregation ------------------------------------ apps_json = data.get("apps", {}) or {} for cat_key, cat_name in (("1", "productive"), ("0", "neutral"), ("-1", "unproductive")): cat_apps = apps_json.get(cat_key) or {} if isinstance(cat_apps, dict): for app_key, app_obj in cat_apps.items(): name = app_obj.get("name") or app_key dur = app_obj.get("duration") or 0 apps_agg[cat_name][name] += int(dur) # Human-readable strings & bar ratio ------------------ day_info["productive_str"] = sec_to_hm(prod_sec) day_info["desktime_str"] = sec_to_hm(desk_sec) day_info["at_work_str"] = sec_to_hm(at_work_sec) day_info["bar_ratio"] = round(bar_ratio, 1) daily_rows.append(day_info) days_count = len(daily_rows) if daily_rows else 1 if arrival_minutes: avg_arrival_m = sum(arrival_minutes) // len(arrival_minutes) avg_arrival_str = f"{avg_arrival_m // 60:02d}:{avg_arrival_m % 60:02d}" else: avg_arrival_str = "-" if left_minutes: avg_left_m = sum(left_minutes) // len(left_minutes) avg_left_str = f"{avg_left_m // 60:02d}:{avg_left_m % 60:02d}" else: avg_left_str = "-" avg_productivity = ( round(sum(productivity_values) / len(productivity_values), 2) if productivity_values else None ) avg_efficiency = ( round(sum(efficiency_values) / len(efficiency_values), 2) if efficiency_values else None ) overall = { "total_productive_str": sec_to_hm(total_productive), "total_desktime_str": sec_to_hm(total_desktime), "total_at_work_str": sec_to_hm(total_at_work), "avg_productive_str": sec_to_hm(total_productive / days_count) if days_count else "0h 0m", "avg_desktime_str": sec_to_hm(total_desktime / days_count) if days_count else "0h 0m", "avg_at_work_str": sec_to_hm(total_at_work / days_count) if days_count else "0h 0m", "avg_arrival_str": avg_arrival_str, "avg_left_str": avg_left_str, "avg_productivity": avg_productivity, "avg_efficiency": avg_efficiency, "total_online_str": sec_to_hm(total_online), "total_offline_str": sec_to_hm(total_offline), "days_count": days_count, } def _apps_list(cat: str): items = apps_agg[cat] arr = [ {"name": name, "duration_str": sec_to_hm(sec), "seconds": sec} for name, sec in items.items() ] arr.sort(key=lambda x: x["seconds"], reverse=True) return arr productive_apps = _apps_list("productive") unproductive_apps = _apps_list("unproductive") neutral_apps = _apps_list("neutral") # ----------------------------------------------------- # Day view (mini dashboard) # ----------------------------------------------------- day_view = None day_productive_apps = [] day_unproductive_apps = [] day_neutral_apps = [] if day_str: try: day_date = datetime.strptime(day_str, "%Y-%m-%d").date() except ValueError: day_date = None if day_date: day_log = DesktimeLog.query.filter_by( employee_id=employee_id, date=day_date ).first() if day_log and day_log.raw_data: try: ddata = json.loads(day_log.raw_data) except Exception: ddata = {} darrival = time_str_from_ts(ddata.get("arrived")) dleft = time_str_from_ts(ddata.get("left")) dprod = ddata.get("productiveTime") or 0 ddesk = ddata.get("desktimeTime") or 0 datwork = ddata.get("atWorkTime") or 0 dprod_pct = ddata.get("productivity") deff_pct = ddata.get("efficiency") day_view = { "date": day_date, "arrival": darrival, "left": dleft, "productive_str": sec_to_hm(dprod), "desktime_str": sec_to_hm(ddesk), "at_work_str": sec_to_hm(datwork), "productivity": dprod_pct, "efficiency": deff_pct, } dapps = ddata.get("apps", {}) or {} def _day_apps_list(cat_key: str): cat = dapps.get(cat_key) or {} out = [] if isinstance(cat, dict): for app_key, app_obj in cat.items(): name = app_obj.get("name") or app_key dur = app_obj.get("duration") or 0 out.append( { "name": name, "duration_str": sec_to_hm(dur), "seconds": int(dur), } ) out.sort(key=lambda x: x["seconds"], reverse=True) return out day_productive_apps = _day_apps_list("1") day_neutral_apps = _day_apps_list("0") day_unproductive_apps = _day_apps_list("-1") return render_template( "attendance/desktime_employee_dashboard.html", employee=emp, selected_month=selected_month, overall=overall, daily_rows=daily_rows, productive_apps=productive_apps, unproductive_apps=unproductive_apps, neutral_apps=neutral_apps, day_view=day_view, day_productive_apps=day_productive_apps, day_unproductive_apps=day_unproductive_apps, day_neutral_apps=day_neutral_apps, selected_day=day_str, ) @desktime_logs_bp.route("/my-desktime") @login_required def my_desktime_dashboard(): """ Menu entry for normal employees. Redirects to their own Desktime dashboard. """ return redirect( url_for( "desktime_logs.desktime_employee_dashboard", employee_id=current_user.id, ) ) # --------------------------------------------------------- # 5. Sync history with parsed payload # --------------------------------------------------------- @desktime_logs_bp.route("/sync-history") @login_required def desktime_sync_history(): logs = SyncLog.query.order_by(SyncLog.started_at.desc()).limit(100).all() parsed_logs = [] for log in logs: payload = {} try: if log.message: payload = json.loads(log.message) except Exception: # keep payload empty if parsing fails payload = {} parsed_logs.append({ "log": log, "payload": payload, }) return render_template( "attendance/desktime_sync_history.html", rows=parsed_logs, )