""" Shopfloor Operator Activity Tracker Flask + SQLite. Designed for touch-only operation on shop-floor tablets. Run: python app.py Then open on the tablet's browser: http://:5000/ """ from flask import Flask, request, jsonify, render_template, send_file from datetime import datetime import json import io import database as db import sap_import app = Flask(__name__) db.init_db() # -------------------------------------------------------------------------- # Pages # -------------------------------------------------------------------------- @app.route("/") def select_station(): """Landing menu: pick which workstation/operation this tablet represents.""" return render_template("select.html") @app.route("/board") def board(): """The touch time-booking board for one workstation (?station=KODE).""" return render_template("index.html") @app.route("/admin") def admin(): return render_template("admin.html") @app.route("/dashboard") def dashboard(): return render_template("dashboard.html") # -------------------------------------------------------------------------- # Helpers # -------------------------------------------------------------------------- def _err(msg, code=400): return jsonify({"ok": False, "error": msg}), code def _find_open_session(conn, nik): return conn.execute( "SELECT * FROM work_sessions WHERE nik=? AND is_open=1 ORDER BY id DESC LIMIT 1", (nik,), ).fetchone() def _close_current_activity(conn, session_id, end_time): conn.execute( "UPDATE activity_log SET end_time=? WHERE session_id=? AND end_time IS NULL", (end_time, session_id), ) # -------------------------------------------------------------------------- # Board state (polled every few seconds by the frontend) # -------------------------------------------------------------------------- @app.route("/api/board") def api_board(): station_id = request.args.get("station", "WKS0001") conn = db.get_db() db.sweep_stale_sessions(conn) station = conn.execute( "SELECT * FROM workstations WHERE id=?", (station_id,) ).fetchone() # Which routing(s) this physical workstation can run. Most stations have # exactly one, in which case the frontend never bothers the operator with # an extra "pick a routing" step - it only shows up when there are 2+. routings = conn.execute( f"""SELECT r.* FROM routings r JOIN workstation_routings wr ON wr.routing_id = r.id WHERE wr.workstation_id=? AND r.active=1 ORDER BY {db.ROUTING_ORDER_SQL}""", (station_id,), ).fetchall() mos = conn.execute( """SELECT m.*, p.project_name as project_name, r.name as routing_name FROM manufacturing_orders m JOIN routings r ON r.id = m.routing_id JOIN workstation_routings wr ON wr.routing_id = m.routing_id LEFT JOIN projects p ON p.project_code = m.project_code WHERE wr.workstation_id=? AND m.status=? AND m.active=1 ORDER BY r.sequence, m.created_at""", (station_id, db.MO_SELECTABLE_STATUS), ).fetchall() team = db.get_team_with_members(conn, station["team_id"]) if station else None team_member_niks = {m["nik"] for m in team["members"]} if team else set() open_sessions = conn.execute( """SELECT ws.*, e.name as emp_name FROM work_sessions ws JOIN employees e ON e.nik = ws.nik WHERE ws.is_open=1 AND ws.workstation_id=?""", (station_id,), ).fetchall() session_by_nik = {s["nik"]: dict(s) for s in open_sessions} # elapsed seconds in the CURRENT activity segment, for on-screen timers lanes = {act: [] for act in db.ALL_LANE_STATUSES} now = datetime.now() for s in open_sessions: cur_act = conn.execute( "SELECT * FROM activity_log WHERE session_id=? AND end_time IS NULL ORDER BY id DESC LIMIT 1", (s["id"],), ).fetchone() started = cur_act["start_time"] if cur_act else s["login_time"] started_dt = datetime.strptime(started, "%Y-%m-%d %H:%M:%S") elapsed = int((now - started_dt).total_seconds()) is_helper = bool(team) and s["nik"] not in team_member_niks is_overdue, overdue_minutes = db.session_overdue_info(conn, s["login_time"], now) if s["is_overtime"]: is_overdue = False # sanctioned lembur - don't alarm the board card = { "nik": s["nik"], "name": s["emp_name"], "mo_number": s["mo_number"], "since": started, "elapsed_seconds": elapsed, "session_id": s["id"], "is_helper": is_helper, "is_overdue": is_overdue, "overdue_minutes": overdue_minutes, "is_overtime": bool(s["is_overtime"]), } if s["status"] == db.LOAN_STATUS: elsewhere = db.find_open_sessions_elsewhere(conn, s["nik"], station_id) active = [e for e in elsewhere if e["status"] == "On Duty"] card["loaned_to"] = active[0]["workstation_name"] if active else None lane = s["status"] if s["status"] in lanes else "On Duty" lanes[lane].append(card) payload = { "ok": True, "server_time": now.strftime("%H:%M:%S"), "station": dict(station) if station else {"id": station_id, "name": station_id}, "routings": [dict(r) for r in routings], "mos": [dict(m) for m in mos], "lanes": lanes, "team": team, "session_by_nik": {nik: {"status": s["status"]} for nik, s in session_by_nik.items()}, "activities": db.AWAY_ACTIVITIES, "end_activities": db.END_ACTIVITIES, } conn.close() return jsonify(payload) # -------------------------------------------------------------------------- # Operator identifies by NIK -> lookup name + whether already has open session # -------------------------------------------------------------------------- @app.route("/api/whoami") def api_whoami(): nik = request.args.get("nik", "").strip() station = request.args.get("station", "").strip() conn = db.get_db() db.sweep_stale_sessions(conn) emp = conn.execute("SELECT * FROM employees WHERE nik=? AND active=1", (nik,)).fetchone() if not emp: conn.close() return _err("NIK tidak ditemukan", 404) session_here = db.find_open_session_at(conn, nik, station) if station else _find_open_session(conn, nik) elsewhere = db.find_open_sessions_elsewhere(conn, nik, station) if station else [] result = { "ok": True, "employee": dict(emp), "session": dict(session_here) if session_here else None, "session_elsewhere": dict(elsewhere[0]) if elsewhere else None, } conn.close() return jsonify(result) # -------------------------------------------------------------------------- # Start work: operator enters NIK + selects MO -> creates session # -------------------------------------------------------------------------- @app.route("/api/login", methods=["POST"]) def api_login(): data = request.get_json(force=True) nik = str(data.get("nik", "")).strip() mo_number = str(data.get("mo_number", "")).strip() station_id = str(data.get("station", "WKS0001")).strip() confirm_transfer = bool(data.get("confirm_transfer", False)) conn = db.get_db() db.sweep_stale_sessions(conn) emp = conn.execute("SELECT * FROM employees WHERE nik=? AND active=1", (nik,)).fetchone() if not emp: conn.close() return _err("NIK tidak dikenali") if db.find_open_session_at(conn, nik, station_id): conn.close() return _err("Operator ini masih memiliki sesi kerja aktif di sini") elsewhere = db.find_open_sessions_elsewhere(conn, nik, station_id) active_elsewhere = [s for s in elsewhere if s["status"] == "On Duty"] if active_elsewhere and not confirm_transfer: s = active_elsewhere[0] conn.close() return jsonify({ "ok": False, "needs_confirmation": True, "error": f"{emp['name']} sedang On Duty di {s['workstation_name']} (MO {s['mo_number']}).", "session_elsewhere": dict(s), }), 409 mo = conn.execute( "SELECT * FROM manufacturing_orders WHERE mo_number=? AND status=? AND active=1", (mo_number, db.MO_SELECTABLE_STATUS), ).fetchone() if not mo: conn.close() return _err("Nomor MO tidak ditemukan atau statusnya sedang tidak Active") now = db.now_str() # Park any other still-active session(s) elsewhere - the operator can only # really be in one place at a time, so their home work pauses (not ends) # while they help here. This is what makes the "loan" mechanism work. for s in elsewhere: if s["status"] == "On Duty": _close_current_activity(conn, s["id"], now) conn.execute("UPDATE work_sessions SET status=? WHERE id=?", (db.LOAN_STATUS, s["id"])) conn.execute( "INSERT INTO activity_log (session_id, nik, activity, start_time) VALUES (?,?,?,?)", (s["id"], nik, db.LOAN_STATUS, now), ) cur = conn.execute( """INSERT INTO work_sessions (nik, workstation_id, routing_id, mo_number, login_time, status, is_open) VALUES (?,?,?,?,?, 'On Duty', 1)""", (nik, station_id, mo["routing_id"], mo_number, now), ) session_id = cur.lastrowid conn.execute( "INSERT INTO activity_log (session_id, nik, activity, start_time) VALUES (?,?, 'On Duty', ?)", (session_id, nik, now), ) conn.commit() conn.close() return jsonify({"ok": True, "session_id": session_id}) # -------------------------------------------------------------------------- # Operator taps their card -> chooses a reason to leave the workstation # -------------------------------------------------------------------------- @app.route("/api/activity", methods=["POST"]) def api_activity(): data = request.get_json(force=True) nik = str(data.get("nik", "")).strip() activity = str(data.get("activity", "")).strip() station_id = str(data.get("station", "")).strip() if activity not in db.ALL_ACTIVITIES: return _err("Aktivitas tidak valid") conn = db.get_db() session = db.find_open_session_at(conn, nik, station_id) if station_id else _find_open_session(conn, nik) if not session: conn.close() return _err("Operator tidak memiliki sesi kerja aktif") now = db.now_str() _close_current_activity(conn, session["id"], now) if activity in db.END_ACTIVITIES: # Ends the whole session conn.execute( "UPDATE work_sessions SET is_open=0, logout_time=?, status=? WHERE id=?", (now, activity, session["id"]), ) conn.execute( "INSERT INTO activity_log (session_id, nik, activity, start_time, end_time) VALUES (?,?,?,?,?)", (session["id"], nik, activity, now, now), ) if activity == "MO Selesai": conn.execute( "UPDATE manufacturing_orders SET status='Completed' WHERE mo_number=?", (session["mo_number"],), ) else: # Move card to a different lane, keep session open conn.execute( "UPDATE work_sessions SET status=? WHERE id=?", (activity, session["id"]) ) conn.execute( "INSERT INTO activity_log (session_id, nik, activity, start_time) VALUES (?,?,?,?)", (session["id"], nik, activity, now), ) conn.commit() conn.close() return jsonify({"ok": True}) # -------------------------------------------------------------------------- # Operator returns to workstation -> resumes "On Duty" # -------------------------------------------------------------------------- @app.route("/api/resume", methods=["POST"]) def api_resume(): data = request.get_json(force=True) nik = str(data.get("nik", "")).strip() station_id = str(data.get("station", "")).strip() conn = db.get_db() session = db.find_open_session_at(conn, nik, station_id) if station_id else _find_open_session(conn, nik) if not session: conn.close() return _err("Operator tidak memiliki sesi kerja aktif") if session["status"] == db.LOAN_STATUS: elsewhere = db.find_open_sessions_elsewhere(conn, nik, station_id) active_elsewhere = [s for s in elsewhere if s["status"] == "On Duty"] if active_elsewhere: s = active_elsewhere[0] conn.close() return _err(f"Tidak bisa dilanjutkan: masih On Duty membantu di {s['workstation_name']}. Selesaikan dulu di sana.") now = db.now_str() _close_current_activity(conn, session["id"], now) conn.execute("UPDATE work_sessions SET status='On Duty' WHERE id=?", (session["id"],)) conn.execute( "INSERT INTO activity_log (session_id, nik, activity, start_time) VALUES (?,?, 'On Duty', ?)", (session["id"], nik, now), ) conn.commit() conn.close() return jsonify({"ok": True}) # -------------------------------------------------------------------------- # Admin: manage employees & MOs (simple CRUD, still touch friendly) # -------------------------------------------------------------------------- # -------------------------------------------------------------------------- # Workstation claims: prevent the same routing from being opened for # time-booking on two tablets at once. # -------------------------------------------------------------------------- @app.route("/api/stations/status") def api_stations_status(): conn = db.get_db() stations = conn.execute( f"""SELECT ws.*, t.name as team_name FROM workstations ws LEFT JOIN teams t ON t.id = ws.team_id WHERE ws.active=1 ORDER BY {db.WORKSTATION_ORDER_SQL}""" ).fetchall() result = [] for st in stations: claim = db.get_active_claim(conn, st["id"]) result.append({ **dict(st), "in_use": claim is not None, "held_by_me": False, # frontend fills this in by comparing its own token "client_token": claim["client_token"] if claim else None, }) conn.close() return jsonify({"ok": True, "stations": result}) @app.route("/api/stations/claim", methods=["POST"]) def api_stations_claim(): data = request.get_json(force=True) wsid = str(data.get("workstation_id", "")).strip() token = str(data.get("client_token", "")).strip() if not wsid or not token: return _err("Data tidak lengkap") conn = db.get_db() ws = conn.execute("SELECT * FROM workstations WHERE id=?", (wsid,)).fetchone() if not ws: conn.close() return _err("Workstation tidak ditemukan") if not ws["active"]: conn.close() return _err("Workstation ini sudah dinonaktifkan") ok, msg = db.claim_station(conn, wsid, token) conn.close() if not ok: return _err(msg, 409) return jsonify({"ok": True}) @app.route("/api/stations/heartbeat", methods=["POST"]) def api_stations_heartbeat(): data = request.get_json(force=True) wsid = str(data.get("workstation_id", "")).strip() token = str(data.get("client_token", "")).strip() conn = db.get_db() ok = db.heartbeat_station(conn, wsid, token) conn.close() if not ok: return _err("Klaim tidak berlaku lagi, silakan pilih ulang workstation", 409) return jsonify({"ok": True}) @app.route("/api/stations/release", methods=["POST"]) def api_stations_release(): data = request.get_json(force=True) wsid = str(data.get("workstation_id", "")).strip() token = str(data.get("client_token", "")).strip() conn = db.get_db() db.release_station(conn, wsid, token) conn.close() return jsonify({"ok": True}) def _actor_nik(): """actor_nik can arrive as a query param (DELETE) or JSON body field.""" nik = request.args.get("actor_nik") if nik: return nik if request.is_json: data = request.get_json(silent=True) or {} return data.get("actor_nik") return None @app.route("/api/admin/whoami-role") def api_admin_whoami_role(): nik = request.args.get("nik", "").strip() conn = db.get_db() actor = db.get_actor(conn, nik) conn.close() if not actor: return _err("NIK tidak ditemukan atau tidak aktif", 404) return jsonify({"ok": True, "nik": actor["nik"], "name": actor["name"], "role": actor["role"]}) # -------------------------------------------------------------------------- # Shift master (Shift 1/2/3) - drives the forgotten-clock-out sweep so it # uses the correct shift boundary regardless of which of the 3 shifts an # operator actually logged in during. # -------------------------------------------------------------------------- @app.route("/api/admin/shifts") def api_admin_shifts(): conn = db.get_db() rows = db.get_shifts(conn) conn.close() return jsonify({"ok": True, "shifts": [dict(r) for r in rows]}) @app.route("/api/admin/shifts/", methods=["PUT"]) def api_admin_update_shift(shift_id): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if not conn.execute("SELECT 1 FROM shifts WHERE id=?", (shift_id,)).fetchone(): conn.close() return _err("Shift tidak ditemukan", 404) conn.execute( "UPDATE shifts SET start_time=?, end_time=? WHERE id=?", (str(data.get("start_time", "")).strip(), str(data.get("end_time", "")).strip(), shift_id), ) conn.commit() conn.close() return jsonify({"ok": True}) # -------------------------------------------------------------------------- # Master WKS (physical workstations) - a location/bench/tablet on the shop # floor. One WKS can run 2+ different Routings (see Master Routing below). # -------------------------------------------------------------------------- @app.route("/api/admin/workstations", methods=["GET"]) def api_admin_workstations_list(): conn = db.get_db() rows = conn.execute( f"""SELECT ws.*, t.name as team_name FROM workstations ws LEFT JOIN teams t ON t.id = ws.team_id ORDER BY ws.active DESC, {db.WORKSTATION_ORDER_SQL}""" ).fetchall() result = [] for r in rows: routings = conn.execute( """SELECT rt.id, rt.name FROM routings rt JOIN workstation_routings wr ON wr.routing_id = rt.id WHERE wr.workstation_id=? ORDER BY rt.name""", (r["id"],), ).fetchall() result.append({**dict(r), "routings": [dict(x) for x in routings]}) conn.close() return jsonify({"ok": True, "workstations": result}) @app.route("/api/admin/workstations", methods=["POST"]) def api_admin_workstations_create(): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["supervisor"]) if not ok: conn.close() return _err(msg, 403) wsid = str(data.get("id", "")).strip() name = str(data.get("name", "")).strip() if not wsid or not name: conn.close() return _err("Kode dan nama workstation wajib diisi") if conn.execute("SELECT 1 FROM workstations WHERE id=?", (wsid,)).fetchone(): conn.close() return _err("Kode workstation sudah ada, gunakan Edit untuk mengubahnya") conn.execute( """INSERT INTO workstations (id, name, group_name, shift_start, shift_end, team_id, active, sequence) VALUES (?,?,?,?,?,?,1,?)""", ( wsid, name, data.get("group_name", "A"), data.get("shift_start", "07:00"), data.get("shift_end", "16:00"), data.get("team_id") or None, int(data.get("sequence", 0) or 0), ), ) routing_ids = data.get("routing_ids") or [] for rid in routing_ids: conn.execute( "INSERT OR IGNORE INTO workstation_routings (workstation_id, routing_id) VALUES (?,?)", (wsid, str(rid).strip()), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/workstations/", methods=["PUT"]) def api_admin_workstations_update(wsid): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if not conn.execute("SELECT 1 FROM workstations WHERE id=?", (wsid,)).fetchone(): conn.close() return _err("Workstation tidak ditemukan", 404) conn.execute( """UPDATE workstations SET name=?, group_name=?, shift_start=?, shift_end=?, team_id=?, sequence=?, active=? WHERE id=?""", ( str(data.get("name", "")).strip(), data.get("group_name", "A"), data.get("shift_start", "07:00"), data.get("shift_end", "16:00"), data.get("team_id") or None, int(data.get("sequence", 0) or 0), 1 if str(data.get("active", "1")) in ("1", "true", "True") else 0, wsid, ), ) if "routing_ids" in data: conn.execute("DELETE FROM workstation_routings WHERE workstation_id=?", (wsid,)) for rid in (data.get("routing_ids") or []): conn.execute( "INSERT OR IGNORE INTO workstation_routings (workstation_id, routing_id) VALUES (?,?)", (wsid, str(rid).strip()), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/workstations/", methods=["DELETE"]) def api_admin_delete_workstation(wsid): conn = db.get_db() ok, msg, _ = db.check_role(conn, _actor_nik(), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if db.workstation_is_referenced(conn, wsid): conn.execute("UPDATE workstations SET active=0 WHERE id=?", (wsid,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deactivated"}) conn.execute("DELETE FROM workstations WHERE id=?", (wsid,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deleted"}) # -------------------------------------------------------------------------- # Master Routing (stage/process step). Linked to 1+ physical Workstations. # -------------------------------------------------------------------------- @app.route("/api/admin/routings", methods=["GET"]) def api_admin_routings_list(): conn = db.get_db() rows = conn.execute( f"""SELECT * FROM routings ORDER BY active DESC, {db.ROUTING_ORDER_SQL}""" ).fetchall() result = [] for r in rows: wks = conn.execute( """SELECT ws.id, ws.name FROM workstations ws JOIN workstation_routings wr ON wr.workstation_id = ws.id WHERE wr.routing_id=? ORDER BY ws.name""", (r["id"],), ).fetchall() result.append({**dict(r), "workstations": [dict(x) for x in wks]}) conn.close() return jsonify({"ok": True, "routings": result}) @app.route("/api/admin/routings", methods=["POST"]) def api_admin_routings_create(): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["supervisor"]) if not ok: conn.close() return _err(msg, 403) rid = str(data.get("id", "")).strip() name = str(data.get("name", "")).strip() if not rid or not name: conn.close() return _err("Kode dan nama routing wajib diisi") if conn.execute("SELECT 1 FROM routings WHERE id=?", (rid,)).fetchone(): conn.close() return _err("Kode routing sudah ada, gunakan Edit untuk mengubahnya") conn.execute( "INSERT INTO routings (id, name, product_group, sequence, active) VALUES (?,?,?,?,1)", ( rid, name, str(data.get("product_group", "")).strip() or None, int(data.get("sequence", 0) or 0), ), ) db.set_routing_workstations(conn, rid, data.get("workstation_ids") or []) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/routings/", methods=["PUT"]) def api_admin_routings_update(rid): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if not conn.execute("SELECT 1 FROM routings WHERE id=?", (rid,)).fetchone(): conn.close() return _err("Routing tidak ditemukan", 404) conn.execute( "UPDATE routings SET name=?, product_group=?, sequence=? WHERE id=?", ( str(data.get("name", "")).strip(), str(data.get("product_group", "")).strip() or None, int(data.get("sequence", 0) or 0), rid, ), ) if "workstation_ids" in data: db.set_routing_workstations(conn, rid, data.get("workstation_ids") or []) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/routings/", methods=["DELETE"]) def api_admin_delete_routing(rid): conn = db.get_db() ok, msg, _ = db.check_role(conn, _actor_nik(), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if db.routing_is_referenced(conn, rid): conn.execute("UPDATE routings SET active=0 WHERE id=?", (rid,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deactivated"}) # Not referenced by any MO/session, so safe to hard-delete - but its # workstation links must go first or the FK constraint would block it. conn.execute("DELETE FROM workstation_routings WHERE routing_id=?", (rid,)) conn.execute("DELETE FROM routings WHERE id=?", (rid,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deleted"}) # -------------------------------------------------------------------------- # Master Team # -------------------------------------------------------------------------- @app.route("/api/admin/teams", methods=["GET"]) def api_admin_teams_list(): conn = db.get_db() rows = conn.execute( """SELECT t.*, e.name as leader_name, sup.name as supervisor_name, (SELECT COUNT(*) FROM employees WHERE team_id=t.id AND active=1) as member_count FROM teams t LEFT JOIN employees e ON e.nik = t.leader_nik LEFT JOIN employees sup ON sup.nik = t.supervisor_nik ORDER BY t.active DESC, t.name""" ).fetchall() conn.close() return jsonify({"ok": True, "teams": [dict(r) for r in rows]}) @app.route("/api/admin/teams", methods=["POST"]) def api_admin_teams_create(): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["supervisor"]) if not ok: conn.close() return _err(msg, 403) name = str(data.get("name", "")).strip() if not name: conn.close() return _err("Nama team wajib diisi") if conn.execute("SELECT 1 FROM teams WHERE name=?", (name,)).fetchone(): conn.close() return _err("Nama team sudah ada, gunakan Edit untuk mengubahnya") code = str(data.get("code", "")).strip().upper() or None if code and conn.execute("SELECT 1 FROM teams WHERE code=?", (code,)).fetchone(): conn.close() return _err("Kode team sudah dipakai team lain") conn.execute( "INSERT INTO teams (code, name, leader_nik, supervisor_nik, active) VALUES (?,?,?,?,1)", (code, name, str(data.get("leader_nik", "")).strip() or None, str(data.get("supervisor_nik", "")).strip() or None), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/teams/", methods=["PUT"]) def api_admin_teams_update(team_id): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if not conn.execute("SELECT 1 FROM teams WHERE id=?", (team_id,)).fetchone(): conn.close() return _err("Team tidak ditemukan", 404) conn.execute( "UPDATE teams SET code=?, name=?, leader_nik=?, supervisor_nik=? WHERE id=?", ( str(data.get("code", "")).strip().upper() or None, str(data.get("name", "")).strip(), str(data.get("leader_nik", "")).strip() or None, str(data.get("supervisor_nik", "")).strip() or None, team_id, ), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/teams/", methods=["DELETE"]) def api_admin_delete_team(team_id): conn = db.get_db() ok, msg, _ = db.check_role(conn, _actor_nik(), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if db.team_is_referenced(conn, team_id): conn.execute("UPDATE teams SET active=0 WHERE id=?", (team_id,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deactivated"}) conn.execute("DELETE FROM teams WHERE id=?", (team_id,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deleted"}) # -------------------------------------------------------------------------- # Master Group (A/B/C/Z shift groups) - simple read-only reference list. # -------------------------------------------------------------------------- @app.route("/api/admin/groups", methods=["GET"]) def api_admin_groups_list(): conn = db.get_db() rows = conn.execute("SELECT * FROM groups ORDER BY code").fetchall() conn.close() return jsonify({"ok": True, "groups": [dict(r) for r in rows]}) # -------------------------------------------------------------------------- # Master Karyawan (employees) # -------------------------------------------------------------------------- @app.route("/api/admin/employees", methods=["GET"]) def api_admin_employees_list(): conn = db.get_db() rows = conn.execute( """SELECT e.*, t.name as team_name FROM employees e LEFT JOIN teams t ON t.id = e.team_id ORDER BY e.active DESC, e.group_name, e.name""" ).fetchall() conn.close() return jsonify({"ok": True, "employees": [dict(r) for r in rows]}) @app.route("/api/admin/employees", methods=["POST"]) def api_admin_employees_create(): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["supervisor"]) if not ok: conn.close() return _err(msg, 403) nik = str(data.get("nik", "")).strip() name = str(data.get("name", "")).strip() if not nik or not name: conn.close() return _err("NIK dan nama wajib diisi") if conn.execute("SELECT 1 FROM employees WHERE nik=?", (nik,)).fetchone(): conn.close() return _err("NIK sudah terdaftar, gunakan Edit untuk mengubahnya") conn.execute( "INSERT INTO employees (nik, name, group_name, team_id, role, active) VALUES (?,?,?,?,?,1)", (nik, name, data.get("group_name", "A"), data.get("team_id") or None, data.get("role", "operator")), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/employees/", methods=["PUT"]) def api_admin_employees_update(nik): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if not conn.execute("SELECT 1 FROM employees WHERE nik=?", (nik,)).fetchone(): conn.close() return _err("Karyawan tidak ditemukan", 404) conn.execute( "UPDATE employees SET name=?, group_name=?, team_id=?, role=? WHERE nik=?", ( str(data.get("name", "")).strip(), data.get("group_name", "A"), data.get("team_id") or None, data.get("role", "operator"), nik, ), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/employees/", methods=["DELETE"]) def api_admin_delete_employee(nik): conn = db.get_db() ok, msg, _ = db.check_role(conn, _actor_nik(), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if db.employee_is_team_leader(conn, nik): conn.close() return _err("Tidak bisa dihapus: karyawan ini masih menjadi leader sebuah Team. Ganti leader-nya dulu di Master Team.") if db.employee_has_transactions(conn, nik): conn.execute("UPDATE employees SET active=0 WHERE nik=?", (nik,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deactivated"}) conn.execute("DELETE FROM employees WHERE nik=?", (nik,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deleted"}) # -------------------------------------------------------------------------- # Master WO / MO # -------------------------------------------------------------------------- @app.route("/api/admin/mos", methods=["GET"]) def api_admin_mos_list(): conn = db.get_db() rows = conn.execute( """SELECT m.*, p.project_name as project_name, r.name as routing_name FROM manufacturing_orders m LEFT JOIN projects p ON p.project_code = m.project_code LEFT JOIN routings r ON r.id = m.routing_id ORDER BY m.active DESC, m.created_at DESC""" ).fetchall() conn.close() return jsonify({"ok": True, "mos": [dict(r) for r in rows]}) @app.route("/api/admin/mos", methods=["POST"]) def api_admin_mos_create(): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["supervisor"]) if not ok: conn.close() return _err(msg, 403) mo_number = str(data.get("mo_number", "")).strip() if not mo_number: conn.close() return _err("Nomor WO/MO wajib diisi") if conn.execute("SELECT 1 FROM manufacturing_orders WHERE mo_number=?", (mo_number,)).fetchone(): conn.close() return _err("Nomor WO/MO sudah ada, gunakan Edit untuk mengubahnya") conn.execute( """INSERT INTO manufacturing_orders (mo_number, description, routing_id, status, ideal_time_minutes, active) VALUES (?,?,?,?,?,1)""", ( mo_number, data.get("description", ""), data.get("routing_id", ""), data.get("status", "Active"), int(data.get("ideal_time_minutes", 0) or 0), ), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/mos/", methods=["PUT"]) def api_admin_mos_update(mo_number): data = request.get_json(force=True) conn = db.get_db() ok, msg, _ = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if not conn.execute("SELECT 1 FROM manufacturing_orders WHERE mo_number=?", (mo_number,)).fetchone(): conn.close() return _err("WO/MO tidak ditemukan", 404) conn.execute( """UPDATE manufacturing_orders SET description=?, routing_id=?, status=?, ideal_time_minutes=? WHERE mo_number=?""", ( data.get("description", ""), data.get("routing_id", ""), data.get("status", "Active"), int(data.get("ideal_time_minutes", 0) or 0), mo_number, ), ) conn.commit() conn.close() return jsonify({"ok": True}) @app.route("/api/admin/mos/", methods=["DELETE"]) def api_admin_delete_mo(mo_number): conn = db.get_db() ok, msg, _ = db.check_role(conn, _actor_nik(), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) if db.mo_has_transactions(conn, mo_number): conn.close() return _err( "WO/MO ini sudah pernah digunakan dalam transaksi, tidak bisa dihapus. " "Gunakan status \"Cancel\" lewat tombol Edit untuk menonaktifkannya." ) conn.execute("DELETE FROM manufacturing_orders WHERE mo_number=?", (mo_number,)) conn.commit() conn.close() return jsonify({"ok": True, "mode": "deleted"}) # -------------------------------------------------------------------------- # SAP integration: import Project / MO / Routing / Material-per-MO reference # data from the SAP "Material Readiness" export so SAP and the shopfloor app # share one consistent picture. Upserts by natural key - never duplicates, # never overwrites locally-configured shopfloor fields (Team, shift, our own # MO status/ideal time). See sap_import.py for the exact rules. # -------------------------------------------------------------------------- @app.route("/api/admin/import-sap", methods=["POST"]) def api_admin_import_sap(): actor_nik = request.form.get("actor_nik", "") conn = db.get_db() ok, msg, _ = db.check_role(conn, actor_nik, ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) file = request.files.get("file") if not file or not file.filename: conn.close() return _err("Pilih file export SAP (.txt) terlebih dahulu") try: raw = file.read() result = sap_import.import_sap_export(conn, raw, filename=file.filename) except ValueError as e: conn.close() return _err(str(e)) except Exception as e: # noqa: BLE001 conn.close() return _err(f"Gagal memproses file: {e}") conn.close() return jsonify(result) @app.route("/api/admin/import-status") def api_admin_import_status(): conn = db.get_db() row = conn.execute("SELECT value FROM app_settings WHERE key='last_sap_import'").fetchone() conn.close() if not row: return jsonify({"ok": True, "last_import": None}) return jsonify({"ok": True, "last_import": json.loads(row["value"])}) @app.route("/api/admin/projects") def api_admin_projects(): conn = db.get_db() rows = conn.execute( """SELECT p.*, (SELECT COUNT(*) FROM manufacturing_orders m WHERE m.project_code=p.project_code) as mo_count FROM projects p ORDER BY p.project_code DESC""" ).fetchall() conn.close() return jsonify({"ok": True, "projects": [dict(r) for r in rows]}) @app.route("/api/admin/mos//materials") def api_admin_mo_materials(mo_number): conn = db.get_db() rows = conn.execute( "SELECT * FROM mo_materials WHERE mo_number=? ORDER BY material_name", (mo_number,) ).fetchall() conn.close() return jsonify({"ok": True, "materials": [dict(r) for r in rows]}) _MATERIAL_COLUMNS = [ ("item_code", "Item"), ("material_name", "Nama Material"), ("specification", "Spesifikasi"), ("required_qty", "Butuh"), ("issued_qty", "Terpakai"), ("remaining_need", "Kurang"), ("on_hand", "Stok"), ("warehouse", "Gudang"), ("po_number", "No. PO"), ("status", "Status"), ] @app.route("/api/admin/mos//materials/export.xlsx") def api_export_materials_xlsx(mo_number): import openpyxl from openpyxl.styles import Font, PatternFill conn = db.get_db() rows = conn.execute( "SELECT * FROM mo_materials WHERE mo_number=? ORDER BY material_name", (mo_number,) ).fetchall() conn.close() wb = openpyxl.Workbook() ws = wb.active ws.title = mo_number[:31] ws.append([f"Material untuk MO {mo_number}"]) ws["A1"].font = Font(bold=True, size=13) ws.append([]) header_row = [label for _, label in _MATERIAL_COLUMNS] ws.append(header_row) header_fill = PatternFill(start_color="122A44", end_color="122A44", fill_type="solid") for cell in ws[3]: cell.font = Font(bold=True, color="FFFFFF") cell.fill = header_fill for r in rows: ws.append([r[col] for col, _ in _MATERIAL_COLUMNS]) for col_cells in ws.columns: length = max((len(str(c.value)) for c in col_cells if c.value is not None), default=10) ws.column_dimensions[col_cells[0].column_letter].width = min(max(length + 2, 10), 40) buf = io.BytesIO() wb.save(buf) buf.seek(0) return send_file( buf, as_attachment=True, download_name=f"Material_{mo_number}.xlsx", mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) @app.route("/api/admin/mos//materials/export.pdf") def api_export_materials_pdf(mo_number): from reportlab.lib import colors from reportlab.lib.pagesizes import A4, landscape from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet from reportlab.lib.units import cm conn = db.get_db() rows = conn.execute( "SELECT * FROM mo_materials WHERE mo_number=? ORDER BY material_name", (mo_number,) ).fetchall() conn.close() buf = io.BytesIO() doc = SimpleDocTemplate( buf, pagesize=landscape(A4), topMargin=1.2 * cm, bottomMargin=1.2 * cm, leftMargin=1.2 * cm, rightMargin=1.2 * cm, ) styles = getSampleStyleSheet() elements = [ Paragraph(f"Material untuk MO {mo_number}", styles["Title"]), Spacer(1, 0.3 * cm), ] header_row = [label for _, label in _MATERIAL_COLUMNS] table_data = [header_row] for r in rows: table_data.append([str(r[col]) if r[col] is not None else "" for col, _ in _MATERIAL_COLUMNS]) if len(table_data) == 1: table_data.append(["-"] * len(header_row)) table = Table(table_data, repeatRows=1) table.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#122A44")), ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), ("FONTSIZE", (0, 0), (-1, -1), 7.5), ("GRID", (0, 0), (-1, -1), 0.5, colors.HexColor("#D7DEE4")), ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#F4F6F8")]), ("VALIGN", (0, 0), (-1, -1), "MIDDLE"), ("LEFTPADDING", (0, 0), (-1, -1), 4), ("RIGHTPADDING", (0, 0), (-1, -1), 4), ])) elements.append(table) doc.build(elements) buf.seek(0) return send_file(buf, as_attachment=True, download_name=f"Material_{mo_number}.pdf", mimetype="application/pdf") # -------------------------------------------------------------------------- # Realtime dashboard across every routing/workstation: WO list + status + # cumulative worked time, refreshed by polling from the browser. # -------------------------------------------------------------------------- @app.route("/api/dashboard") def api_dashboard(): conn = db.get_db() db.sweep_stale_sessions(conn) stations = conn.execute( f"""SELECT ws.*, t.name as team_name FROM workstations ws LEFT JOIN teams t ON t.id = ws.team_id WHERE ws.active=1 ORDER BY {db.WORKSTATION_ORDER_SQL}""" ).fetchall() now = datetime.now() result = [] for st in stations: mos = conn.execute( "SELECT * FROM manufacturing_orders WHERE workstation_id=? ORDER BY (status='Completed'), created_at", (st["id"],), ).fetchall() mo_list = [] for mo in mos: worked_sec = db.worked_seconds_for_mo(conn, mo["mo_number"]) active = conn.execute( """SELECT e.name FROM work_sessions ws JOIN employees e ON e.nik = ws.nik WHERE ws.is_open=1 AND ws.mo_number=? AND ws.status='On Duty'""", (mo["mo_number"],), ).fetchall() ideal_sec = (mo["ideal_time_minutes"] or 0) * 60 mo_list.append({ "mo_number": mo["mo_number"], "description": mo["description"], "status": mo["status"], "ideal_time_minutes": mo["ideal_time_minutes"], "worked_seconds": worked_sec, "progress_pct": round(min(worked_sec / ideal_sec * 100, 100), 1) if ideal_sec else None, "active_operators": [a["name"] for a in active], }) open_sessions = conn.execute( "SELECT status, login_time, is_overtime FROM work_sessions WHERE is_open=1 AND workstation_id=?", (st["id"],), ).fetchall() on_duty_count = sum(1 for s in open_sessions if s["status"] == "On Duty") away_count = sum(1 for s in open_sessions if s["status"] != "On Duty") overdue_count = sum( 1 for s in open_sessions if not s["is_overtime"] and db.session_overdue_info(conn, s["login_time"], now)[0] ) result.append({ "id": st["id"], "name": st["name"], "group_name": st["group_name"], "team_name": st["team_name"], "shift_start": st["shift_start"], "shift_end": st["shift_end"], "on_duty_count": on_duty_count, "away_count": away_count, "overdue_count": overdue_count, "mos": mo_list, }) conn.close() return jsonify({"ok": True, "server_time": now.strftime("%H:%M:%S"), "stations": result}) # -------------------------------------------------------------------------- # Reports: per-session log + simple daily summary (worked vs idle minutes) # -------------------------------------------------------------------------- @app.route("/api/report/summary") def api_report_summary(): date_str = request.args.get("date", datetime.now().strftime("%Y-%m-%d")) station = request.args.get("station", "").strip() conn = db.get_db() q = """SELECT a.activity, a.start_time, a.end_time, ws.mo_number, ws.workstation_id, ws.nik, e.name FROM activity_log a JOIN work_sessions ws ON ws.id = a.session_id JOIN employees e ON e.nik = ws.nik WHERE substr(a.start_time,1,10)=?""" params = [date_str] if station: q += " AND ws.workstation_id=?" params.append(station) rows = conn.execute(q, params).fetchall() mos = {m["mo_number"]: dict(m) for m in conn.execute("SELECT * FROM manufacturing_orders").fetchall()} now = datetime.now() mo_agg, op_agg = {}, {} for r in rows: start = datetime.strptime(r["start_time"], "%Y-%m-%d %H:%M:%S") end = datetime.strptime(r["end_time"], "%Y-%m-%d %H:%M:%S") if r["end_time"] else now secs = (end - start).total_seconds() act, mo, nik = r["activity"], r["mo_number"], r["nik"] me = mo_agg.setdefault(mo, { "description": mos.get(mo, {}).get("description", ""), "status": mos.get(mo, {}).get("status", ""), "ideal_time_minutes": mos.get(mo, {}).get("ideal_time_minutes", 0), "worked_sec": 0.0, "activities": {}, }) oe = op_agg.setdefault(nik, {"name": r["name"], "worked_sec": 0.0, "away_sec": 0.0, "activities": {}}) if act == "On Duty": me["worked_sec"] += secs oe["worked_sec"] += secs else: me["activities"][act] = me["activities"].get(act, 0.0) + secs oe["away_sec"] += secs oe["activities"][act] = oe["activities"].get(act, 0.0) + secs def mins(sec): return round(sec / 60, 1) mo_summary = [] for mo, d in mo_agg.items(): worked_min = mins(d["worked_sec"]) ideal = d["ideal_time_minutes"] or 0 eff = round(ideal / worked_min * 100, 1) if worked_min >= 1 and ideal > 0 else None mo_summary.append({ "mo_number": mo, "description": d["description"], "status": d["status"], "ideal_time_minutes": ideal, "worked_minutes": worked_min, "efficiency_pct": eff, "activities_minutes": {k: mins(v) for k, v in d["activities"].items()}, }) op_summary = [ { "nik": nik, "name": d["name"], "worked_minutes": mins(d["worked_sec"]), "away_minutes": mins(d["away_sec"]), "activities_minutes": {k: mins(v) for k, v in d["activities"].items()}, } for nik, d in op_agg.items() ] conn.close() return jsonify({ "ok": True, "date": date_str, "station": station or None, "mo_summary": mo_summary, "operator_summary": op_summary, }) # -------------------------------------------------------------------------- # Report: cross-team "helper" history - operators who worked at a routing # other than the one their own Team is assigned to, and how much time they # actually spent working there. Time is always attributed to the routing # where the work_session was opened (i.e. the routing being helped), never # to the operator's home routing - that's how work_sessions are recorded in # the first place, so this report just needs to identify + aggregate it. # -------------------------------------------------------------------------- @app.route("/api/report/helpers") def api_report_helpers(): date_str = request.args.get("date", datetime.now().strftime("%Y-%m-%d")) station = request.args.get("station", "").strip() conn = db.get_db() q = """SELECT a.start_time, a.end_time, ws.nik, ws.workstation_id, ws.mo_number, e.name as emp_name, e.team_id as emp_team_id, wst.name as ws_name, wst.team_id as ws_team_id, t_emp.name as emp_team_name, t_ws.name as ws_team_name FROM activity_log a JOIN work_sessions ws ON ws.id = a.session_id JOIN employees e ON e.nik = ws.nik JOIN workstations wst ON wst.id = ws.workstation_id LEFT JOIN teams t_emp ON t_emp.id = e.team_id LEFT JOIN teams t_ws ON t_ws.id = wst.team_id WHERE a.activity='On Duty' AND substr(a.start_time,1,10)=?""" params = [date_str] if station: q += " AND ws.workstation_id=?" params.append(station) rows = conn.execute(q, params).fetchall() conn.close() now = datetime.now() agg = {} for r in rows: # Only counts as "helping" when the routing has an assigned Team and # this operator isn't part of it (same rule as the live board badge). if not r["ws_team_id"]: continue if r["emp_team_id"] == r["ws_team_id"]: continue start = datetime.strptime(r["start_time"], "%Y-%m-%d %H:%M:%S") end = datetime.strptime(r["end_time"], "%Y-%m-%d %H:%M:%S") if r["end_time"] else now secs = (end - start).total_seconds() key = (r["nik"], r["workstation_id"]) e = agg.setdefault(key, { "nik": r["nik"], "name": r["emp_name"], "home_team": r["emp_team_name"] or "(Tanpa Team)", "workstation_id": r["workstation_id"], "workstation_name": r["ws_name"], "host_team": r["ws_team_name"], "worked_sec": 0.0, "mo_numbers": set(), }) e["worked_sec"] += secs e["mo_numbers"].add(r["mo_number"]) helpers = [] for e in agg.values(): helpers.append({ "nik": e["nik"], "name": e["name"], "home_team": e["home_team"], "workstation_id": e["workstation_id"], "workstation_name": e["workstation_name"], "host_team": e["host_team"], "worked_minutes": round(e["worked_sec"] / 60, 1), "mo_numbers": sorted(e["mo_numbers"]), }) helpers.sort(key=lambda x: -x["worked_minutes"]) return jsonify({"ok": True, "date": date_str, "station": station or None, "helpers": helpers}) # -------------------------------------------------------------------------- # Active sessions across every routing - lets a Leader/Supervisor spot and # fix a forgotten clock-out from one screen, instead of walking to the # tablet or waiting for the automatic sweep to eventually close it. # -------------------------------------------------------------------------- @app.route("/api/admin/active-sessions") def api_admin_active_sessions(): conn = db.get_db() db.sweep_stale_sessions(conn) now = datetime.now() rows = conn.execute( """SELECT ws.*, e.name as emp_name, w.name as workstation_name, r.name as routing_name FROM work_sessions ws JOIN employees e ON e.nik = ws.nik JOIN workstations w ON w.id = ws.workstation_id LEFT JOIN routings r ON r.id = ws.routing_id WHERE ws.is_open=1 ORDER BY ws.login_time""" ).fetchall() result = [] for r in rows: login_dt = datetime.strptime(r["login_time"], "%Y-%m-%d %H:%M:%S") is_overdue, overdue_minutes = db.session_overdue_info(conn, r["login_time"], now) if r["is_overtime"]: is_overdue = False shift = db.find_shift_for_datetime(conn, login_dt) result.append({ "session_id": r["id"], "nik": r["nik"], "name": r["emp_name"], "workstation_id": r["workstation_id"], "workstation_name": r["workstation_name"], "routing_id": r["routing_id"], "routing_name": r["routing_name"], "mo_number": r["mo_number"], "status": r["status"], "login_time": r["login_time"], "duration_minutes": round((now - login_dt).total_seconds() / 60, 1), "is_overdue": is_overdue, "overdue_minutes": overdue_minutes, "is_overtime": bool(r["is_overtime"]), "shift_name": shift["name"] if shift else None, }) conn.close() return jsonify({"ok": True, "sessions": result}) @app.route("/api/admin/active-sessions//toggle-overtime", methods=["POST"]) def api_admin_toggle_overtime(session_id): data = request.get_json(force=True) conn = db.get_db() ok, msg, actor = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) overtime = bool(data.get("overtime", True)) done = db.set_overtime(conn, session_id, actor, overtime) conn.close() if not done: return _err("Sesi tidak ditemukan atau sudah selesai", 404) return jsonify({"ok": True}) @app.route("/api/admin/active-sessions//force-end", methods=["POST"]) def api_admin_force_end_session(session_id): data = request.get_json(force=True) conn = db.get_db() ok, msg, actor = db.check_role(conn, data.get("actor_nik"), ["leader", "supervisor"]) if not ok: conn.close() return _err(msg, 403) done = db.force_end_session(conn, session_id, actor) conn.close() if not done: return _err("Sesi tidak ditemukan atau sudah selesai", 404) return jsonify({"ok": True}) @app.route("/api/report/sessions") def api_report_sessions(): date_str = request.args.get("date", datetime.now().strftime("%Y-%m-%d")) conn = db.get_db() rows = conn.execute( """SELECT ws.id as session_id, e.nik, e.name, ws.mo_number, ws.login_time, ws.logout_time, ws.status FROM work_sessions ws JOIN employees e ON e.nik = ws.nik WHERE substr(ws.login_time,1,10)=? ORDER BY ws.login_time""", (date_str,), ).fetchall() result = [] for r in rows: acts = conn.execute( "SELECT activity, start_time, end_time FROM activity_log WHERE session_id=? ORDER BY id", (r["session_id"],), ).fetchall() result.append({**dict(r), "activities": [dict(a) for a in acts]}) conn.close() return jsonify({"ok": True, "date": date_str, "sessions": result}) if __name__ == "__main__": # Local development only. In production (cPanel's "Setup Python App" / # any Passenger-based WSGI host) this block never runs - Passenger # imports this module directly and calls the `application` object below. app.run(host="0.0.0.0", port=5000, debug=True) # WSGI entry point for Passenger (cPanel "Setup Python App", Application # Entry point = "application"). Must be a module-level name Passenger can # import; it's the exact same Flask app object as `app` above, so every # route already registered on `app` is served through it. application = app