""" CRUD & Rekapitulasi Engine untuk Aplikasi Absen Karyawan Online & Integrasi HRIS Pengupahan. """ from datetime import date, datetime from typing import Optional from app.database import ( get_connection, generate_employee_token, SHIFT_DEFINITIONS, get_schedule_for_shift, evaluate_lateness, calculate_work_duration, haversine_distance_meters, ) def list_locations(kategori: Optional[str] = None): conn = get_connection() cur = conn.cursor() if kategori and kategori != "ALL": cur.execute( "SELECT * FROM work_locations WHERE kategori = ? ORDER BY id ASC", (kategori,), ) else: cur.execute("SELECT * FROM work_locations ORDER BY id ASC") rows = [dict(r) for r in cur.fetchall()] conn.close() return rows def save_location(data: dict): conn = get_connection() cur = conn.cursor() loc_id = data.get("id") kode_lokasi = (data.get("kode_lokasi") or "").strip() nama_lokasi = (data.get("nama_lokasi") or "").strip() kategori = data.get("kategori") or "Office" mode_radius = data.get("mode_radius") or ("Bebas Radius" if kategori == "Proyek" else "Ditentukan") latitude = float(data.get("latitude") or -0.95185) longitude = float(data.get("longitude") or 122.78752) radius_meter = int(data.get("radius_meter") or 150) shift_default = data.get("shift_default") or ("PROYEK_06" if kategori == "Proyek" else "OFFICE") if kategori == "Office": jam_in_norm = "08:00" jam_out_norm = "16:00" jam_in_sab = "08:00" jam_out_sab = "13:00" total_jam = 8.0 else: s_def = SHIFT_DEFINITIONS.get(shift_default, SHIFT_DEFINITIONS["PROYEK_06"]) jam_in_norm = s_def["jam_masuk"] jam_out_norm = s_def["jam_pulang_normal"] jam_in_sab = s_def["jam_masuk"] jam_out_sab = s_def["jam_pulang_sabtu"] total_jam = 12.0 keterangan = data.get("keterangan") or "" if loc_id: cur.execute( """ UPDATE work_locations SET kode_lokasi = ?, nama_lokasi = ?, kategori = ?, mode_radius = ?, latitude = ?, longitude = ?, radius_meter = ?, shift_default = ?, jam_masuk_senin_jumat = ?, jam_pulang_senin_jumat = ?, jam_masuk_sabtu = ?, jam_pulang_sabtu = ?, total_jam_kerja = ?, keterangan = ? WHERE id = ? """, ( kode_lokasi, nama_lokasi, kategori, mode_radius, latitude, longitude, radius_meter, shift_default, jam_in_norm, jam_out_norm, jam_in_sab, jam_out_sab, total_jam, keterangan, int(loc_id), ), ) else: if not kode_lokasi: cur.execute("SELECT COUNT(*) as cnt FROM work_locations") cnt = cur.fetchone()["cnt"] + 1 kode_lokasi = f"LOC-{cnt:02d}" cur.execute( """ INSERT INTO work_locations ( kode_lokasi, nama_lokasi, kategori, mode_radius, latitude, longitude, radius_meter, shift_default, jam_masuk_senin_jumat, jam_pulang_senin_jumat, jam_masuk_sabtu, jam_pulang_sabtu, total_jam_kerja, keterangan ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( kode_lokasi, nama_lokasi, kategori, mode_radius, latitude, longitude, radius_meter, shift_default, jam_in_norm, jam_out_norm, jam_in_sab, jam_out_sab, total_jam, keterangan, ), ) loc_id = cur.lastrowid conn.commit() conn.close() return loc_id def delete_location(loc_id: int): conn = get_connection() cur = conn.cursor() cur.execute("DELETE FROM work_locations WHERE id = ?", (loc_id,)) conn.commit() conn.close() def save_employee_face(emp_id: int, face_base64: str): """Menyimpan atau memperbarui foto wajah acuan master karyawan.""" conn = get_connection() cur = conn.cursor() cur.execute("UPDATE employees SET face_template = ? WHERE id = ?", (face_base64, emp_id)) conn.commit() return True def get_employee_face(emp_id: int): """Mengambil foto wajah acuan master karyawan.""" conn = get_connection() cur = conn.cursor() cur.execute("SELECT id, nip, nama, perusahaan, jabatan, face_template FROM employees WHERE id = ?", (emp_id,)) row = cur.fetchone() if row: return dict(row) return None def list_employees(search: str = "", perusahaan: str = "", departemen: str = "", kategori_kerja: str = ""): conn = get_connection() cur = conn.cursor() query = "SELECT * FROM employees WHERE 1=1" params = [] if search: query += " AND (LOWER(nama) LIKE ? OR LOWER(nip) LIKE ? OR LOWER(jabatan) LIKE ?)" s = f"%{search.lower().strip()}%" params.extend([s, s, s]) if perusahaan and perusahaan != "ALL": query += " AND perusahaan = ?" params.append(perusahaan) if departemen and departemen != "ALL": query += " AND departemen = ?" params.append(departemen) if kategori_kerja and kategori_kerja != "ALL": query += " AND kategori_kerja = ?" params.append(kategori_kerja) query += " ORDER BY nama ASC" cur.execute(query, params) rows = [dict(r) for r in cur.fetchall()] conn.close() return rows def save_employee(data: dict): conn = get_connection() cur = conn.cursor() emp_id = data.get("id") nip = (data.get("nip") or "").strip() nama = (data.get("nama") or "").strip() perusahaan = (data.get("perusahaan") or "PT. BSS").strip() jabatan = (data.get("jabatan") or "Staf").strip() departemen = (data.get("departemen") or "General Service").strip() kategori_kerja = data.get("kategori_kerja") or "Office" lokasi_default_id = int(data.get("lokasi_default_id") or 1) shift_default = data.get("shift_default") or ("PROYEK_06" if kategori_kerja == "Proyek" else "OFFICE") gaji_pokok_hris = float(data.get("gaji_pokok_hris") or 4500000) potongan_alfa_per_hari = float(data.get("potongan_alfa_per_hari") or 150000) no_hp = (data.get("no_hp") or "").strip() cur.execute("SELECT nama_lokasi, kategori FROM work_locations WHERE id = ?", (lokasi_default_id,)) loc_row = cur.fetchone() nama_lokasi_default = loc_row["nama_lokasi"] if loc_row else "Kantor Operasional Pasar Tua" if emp_id: cur.execute( """ UPDATE employees SET nip = ?, nama = ?, perusahaan = ?, jabatan = ?, departemen = ?, kategori_kerja = ?, lokasi_default_id = ?, nama_lokasi_default = ?, shift_default = ?, gaji_pokok_hris = ?, potongan_alfa_per_hari = ?, no_hp = ? WHERE id = ? """, ( nip, nama, perusahaan, jabatan, departemen, kategori_kerja, lokasi_default_id, nama_lokasi_default, shift_default, gaji_pokok_hris, potongan_alfa_per_hari, no_hp, int(emp_id), ), ) else: cur.execute( """ INSERT INTO employees ( nip, nama, perusahaan, jabatan, departemen, kategori_kerja, lokasi_default_id, nama_lokasi_default, shift_default, gaji_pokok_hris, potongan_alfa_per_hari, no_hp ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( nip, nama, perusahaan, jabatan, departemen, kategori_kerja, lokasi_default_id, nama_lokasi_default, shift_default, gaji_pokok_hris, potongan_alfa_per_hari, no_hp, ), ) emp_id = cur.lastrowid cur.execute( "UPDATE employees SET token_absen = ? WHERE id = ?", (generate_employee_token(), emp_id), ) conn.commit() conn.close() return emp_id def delete_employee(emp_id: int): conn = get_connection() cur = conn.cursor() cur.execute("DELETE FROM attendance_logs WHERE employee_id = ?", (emp_id,)) cur.execute("DELETE FROM employees WHERE id = ?", (emp_id,)) conn.commit() conn.close() def submit_attendance(data: dict): """ Memproses Absen Masuk (Check-In), Absen Keluar (Check-Out), atau Pencatatan Status (Hadir, Terlambat, Sakit, Alfa, Izin, Cuti) beserta validasi GPS dan perhitungan jam kerja. """ conn = get_connection() cur = conn.cursor() employee_id = int(data.get("employee_id")) tanggal = (data.get("tanggal") or date.today().strftime("%Y-%m-%d")).strip() tipe_aksi = (data.get("tipe_aksi") or "MASUK").upper() # MASUK, KELUAR, STATUS_PENUH kategori_kerja = data.get("kategori_kerja") or "Office" lokasi_id = int(data.get("lokasi_id") or 1) shift_kode = data.get("shift_kode") or ("OFFICE" if kategori_kerja == "Office" else "PROYEK_06") status_input = data.get("status_kehadiran") or "Hadir" jam_masuk_input = (data.get("jam_masuk") or "").strip() jam_keluar_input = (data.get("jam_keluar") or "").strip() lat = data.get("latitude") lng = data.get("longitude") catatan = (data.get("catatan") or "").strip() bebaskan_radius_proyek = bool(data.get("bebaskan_radius_proyek", False)) cur.execute("SELECT * FROM employees WHERE id = ?", (employee_id,)) emp = cur.fetchone() if not emp: conn.close() raise ValueError("Data karyawan tidak ditemukan.") cur.execute("SELECT * FROM work_locations WHERE id = ?", (lokasi_id,)) loc = cur.fetchone() if not loc: cur.execute("SELECT * FROM work_locations ORDER BY id LIMIT 1") loc = cur.fetchone() mode_radius = "Bebas Radius" if (bebaskan_radius_proyek or loc["mode_radius"] == "Bebas Radius") else "Ditentukan" sched = get_schedule_for_shift(shift_kode, tanggal, kategori_kerja) jadwal_masuk = sched["jam_jadwal_masuk"] jadwal_pulang = sched["jam_jadwal_pulang"] target_jam = sched["target_jam_kerja"] # Hitung jarak GPS ke titik lokasi lat_val = float(lat) if lat not in (None, "") else loc["latitude"] lng_val = float(lng) if lng not in (None, "") else loc["longitude"] jarak = haversine_distance_meters(loc["latitude"], loc["longitude"], lat_val, lng_val) if mode_radius == "Bebas Radius": gps_status = f"Bebas Radius Proyek ({jarak}m dari titik referensi)" else: if jarak <= loc["radius_meter"]: gps_status = f"Di Dalam Radius ({jarak}m / Maks {loc['radius_meter']}m)" else: gps_status = f"Di Luar Radius ({jarak}m > {loc['radius_meter']}m)" # Cek apakah sudah ada log absen pada tanggal & karyawan tersebut cur.execute( "SELECT * FROM attendance_logs WHERE tanggal = ? AND employee_id = ?", (tanggal, employee_id), ) existing = cur.fetchone() now_hm = datetime.now().strftime("%H:%M") if tipe_aksi == "KELUAR" and existing: jam_masuk_final = existing["jam_masuk"] or jadwal_masuk jam_keluar_final = jam_keluar_input or now_hm durasi = calculate_work_duration(jam_masuk_final, jam_keluar_final) cur.execute( """ UPDATE attendance_logs SET jam_keluar = ?, durasi_jam_kerja = ?, lat_keluar = ?, lng_keluar = ?, jarak_keluar_meter = ?, status_gps_keluar = ?, catatan = CASE WHEN ? != '' THEN ? ELSE catatan END, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( jam_keluar_final, durasi, lat_val, lng_val, jarak, gps_status, catatan, catatan, existing["id"], ), ) log_id = existing["id"] else: # Aksi Absen Masuk atau Input Status (Hadir, Terlambat, Sakit, Alfa, Izin, Cuti) if status_input in ("Sakit", "Izin", "Cuti", "Alfa"): jam_masuk_final = "" jam_keluar_final = "" is_late = 0 late_mins = 0 durasi = 0.0 status_final = status_input else: jam_masuk_final = jam_masuk_input or now_hm jam_keluar_final = jam_keluar_input or (existing["jam_keluar"] if existing else "") is_late, late_mins = evaluate_lateness(jam_masuk_final, jadwal_masuk) if is_late or status_input == "Terlambat": status_final = "Terlambat" is_late = 1 if late_mins == 0: late_mins = 15 else: status_final = "Hadir" durasi = calculate_work_duration(jam_masuk_final, jam_keluar_final) if jam_keluar_final else 0.0 if existing: cur.execute( """ UPDATE attendance_logs SET kategori_kerja = ?, lokasi_id = ?, nama_lokasi = ?, mode_radius = ?, shift_kode = ?, jam_jadwal_masuk = ?, jam_jadwal_pulang = ?, target_jam_kerja = ?, jam_masuk = ?, jam_keluar = ?, durasi_jam_kerja = ?, status_kehadiran = ?, is_terlambat = ?, menit_terlambat = ?, lat_masuk = ?, lng_masuk = ?, jarak_masuk_meter = ?, status_gps_masuk = ?, catatan = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( kategori_kerja, loc["id"], loc["nama_lokasi"], mode_radius, sched["shift_kode"], jadwal_masuk, jadwal_pulang, target_jam, jam_masuk_final, jam_keluar_final, durasi, status_final, is_late, late_mins, lat_val if jam_masuk_final else None, lng_val if jam_masuk_final else None, jarak if jam_masuk_final else 0.0, gps_status if jam_masuk_final else "-", catatan, existing["id"], ), ) log_id = existing["id"] else: cur.execute( """ INSERT INTO attendance_logs ( tanggal, employee_id, nip, nama_karyawan, perusahaan, departemen, jabatan, kategori_kerja, lokasi_id, nama_lokasi, mode_radius, shift_kode, jam_jadwal_masuk, jam_jadwal_pulang, target_jam_kerja, jam_masuk, jam_keluar, durasi_jam_kerja, status_kehadiran, is_terlambat, menit_terlambat, lat_masuk, lng_masuk, jarak_masuk_meter, status_gps_masuk, catatan ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( tanggal, emp["id"], emp["nip"], emp["nama"], emp["perusahaan"], emp["departemen"], emp["jabatan"], kategori_kerja, loc["id"], loc["nama_lokasi"], mode_radius, sched["shift_kode"], jadwal_masuk, jadwal_pulang, target_jam, jam_masuk_final, jam_keluar_final, durasi, status_final, is_late, late_mins, lat_val if jam_masuk_final else None, lng_val if jam_masuk_final else None, jarak if jam_masuk_final else 0.0, gps_status if jam_masuk_final else "-", catatan, ), ) log_id = cur.lastrowid conn.commit() cur.execute("SELECT * FROM attendance_logs WHERE id = ?", (log_id,)) saved = dict(cur.fetchone()) conn.close() return saved def delete_attendance_log(log_id: int): conn = get_connection() cur = conn.cursor() cur.execute("DELETE FROM attendance_logs WHERE id = ?", (log_id,)) conn.commit() conn.close() def list_attendance_logs( start_date: str = "", end_date: str = "", kategori_kerja: str = "", lokasi_id: str = "", status_kehadiran: str = "", search: str = "", ): conn = get_connection() cur = conn.cursor() query = "SELECT * FROM attendance_logs WHERE 1=1" params = [] if start_date: query += " AND tanggal >= ?" params.append(start_date) if end_date: query += " AND tanggal <= ?" params.append(end_date) if kategori_kerja and kategori_kerja != "ALL": query += " AND kategori_kerja = ?" params.append(kategori_kerja) if lokasi_id and lokasi_id != "ALL": query += " AND lokasi_id = ?" params.append(int(lokasi_id)) if status_kehadiran and status_kehadiran != "ALL": query += " AND status_kehadiran = ?" params.append(status_kehadiran) if search: query += " AND (LOWER(nama_karyawan) LIKE ? OR LOWER(nip) LIKE ? OR LOWER(nama_lokasi) LIKE ?)" s = f"%{search.lower().strip()}%" params.extend([s, s, s]) query += " ORDER BY tanggal DESC, jam_masuk DESC, id DESC" cur.execute(query, params) rows = [dict(r) for r in cur.fetchall()] conn.close() return rows def get_recapitulation_and_hris( start_date: str = "", end_date: str = "", perusahaan: str = "", departemen: str = "", kategori_kerja: str = "", search: str = "", ): """ Menghitung Rekapitulasi Absen per Karyawan beserta Rumus Akumulasi Keterlambatan HRD: - Hadir (Tepat Waktu) - Terlambat (Jumlah hari terlambat & total menit terlambat) - Sakit - Izin - Cuti - Alfa Murni - Akumulasi Alfa dari Terlambat: floor(Jumlah Hari Terlambat / 3) (Setiap 3 hari terlambat terakumulasi menjadi 1 kali Alfa!) - Total Alfa Efektif HRIS = Alfa Murni + floor(Hari Terlambat / 3) - Estimasi Potongan Pengupahan HRIS """ conn = get_connection() cur = conn.cursor() emp_query = "SELECT * FROM employees WHERE 1=1" emp_params = [] if perusahaan and perusahaan != "ALL": emp_query += " AND perusahaan = ?" emp_params.append(perusahaan) if departemen and departemen != "ALL": emp_query += " AND departemen = ?" emp_params.append(departemen) if kategori_kerja and kategori_kerja != "ALL": emp_query += " AND kategori_kerja = ?" emp_params.append(kategori_kerja) if search: emp_query += " AND (LOWER(nama) LIKE ? OR LOWER(nip) LIKE ? OR LOWER(nama_lokasi_default) LIKE ?)" s = f"%{search.lower().strip()}%" emp_params.extend([s, s, s]) emp_query += " ORDER BY id ASC" cur.execute(emp_query, emp_params) employees = [dict(r) for r in cur.fetchall()] log_query = "SELECT * FROM attendance_logs WHERE 1=1" log_params = [] if start_date: log_query += " AND tanggal >= ?" log_params.append(start_date) if end_date: log_query += " AND tanggal <= ?" log_params.append(end_date) cur.execute(log_query, log_params) logs = [dict(r) for r in cur.fetchall()] conn.close() logs_by_emp = {} for lg in logs: logs_by_emp.setdefault(lg["employee_id"], []).append(lg) rekap_rows = [] totals = { "total_karyawan": len(employees), "total_hadir_tepat": 0, "total_terlambat_hari": 0, "total_hadir_gabungan": 0, "total_sakit": 0, "total_izin": 0, "total_cuti": 0, "total_alfa_murni": 0, "total_alfa_akumulasi_terlambat": 0, "total_alfa_final_hris": 0, "jumlah_karyawan_kena_penalti_alfa": 0, } for emp in employees: emp_logs = logs_by_emp.get(emp["id"], []) hadir_tepat = sum(1 for x in emp_logs if x["status_kehadiran"] == "Hadir" and not x["is_terlambat"]) terlambat_hari = sum(1 for x in emp_logs if x["status_kehadiran"] == "Terlambat" or x["is_terlambat"] == 1) total_menit_terlambat = sum(int(x["menit_terlambat"] or 0) for x in emp_logs) sakit = sum(1 for x in emp_logs if x["status_kehadiran"] == "Sakit") izin = sum(1 for x in emp_logs if x["status_kehadiran"] == "Izin") cuti = sum(1 for x in emp_logs if x["status_kehadiran"] == "Cuti") alfa_murni = sum(1 for x in emp_logs if x["status_kehadiran"] == "Alfa") total_jam_kerja = round(sum(float(x["durasi_jam_kerja"] or 0.0) for x in emp_logs), 1) # RUMUS UTAMA HRD: Terlambat dalam 3 hari terakumulasi 1 kali Alfa alfa_akumulasi_terlambat = terlambat_hari // 3 sisa_terlambat = terlambat_hari % 3 total_alfa_hris = alfa_murni + alfa_akumulasi_terlambat # Total kehadiran fisik di tempat kerja (Hadir Tepat Waktu + Hadir Terlambat) total_kehadiran_fisik = hadir_tepat + terlambat_hari potongan_per_alfa = float(emp.get("potongan_alfa_per_hari") or 150000) total_potongan_alfa_rp = total_alfa_hris * potongan_per_alfa if alfa_akumulasi_terlambat > 0: status_disiplin = f"Terakumulasi +{alfa_akumulasi_terlambat} Alfa ({terlambat_hari}x Terlambat)" totals["jumlah_karyawan_kena_penalti_alfa"] += 1 elif sisa_terlambat > 0: status_disiplin = f"Peringatan ({sisa_terlambat}/3 Hari Terlambat)" elif alfa_murni > 0: status_disiplin = f"Alfa Murni ({alfa_murni} Hari)" else: status_disiplin = "Disiplin Baik" row_data = { "employee_id": emp["id"], "nip": emp["nip"], "nama": emp["nama"], "perusahaan": emp["perusahaan"], "jabatan": emp["jabatan"], "departemen": emp["departemen"], "kategori_kerja": emp["kategori_kerja"], "nama_lokasi_default": emp["nama_lokasi_default"], "shift_default": emp["shift_default"], "hadir_tepat": hadir_tepat, "terlambat_hari": terlambat_hari, "total_kehadiran_fisik": total_kehadiran_fisik, "total_menit_terlambat": total_menit_terlambat, "sakit": sakit, "izin": izin, "cuti": cuti, "alfa_murni": alfa_murni, "alfa_akumulasi_terlambat": alfa_akumulasi_terlambat, "sisa_terlambat_menuju_alfa": sisa_terlambat, "total_alfa_hris": total_alfa_hris, "total_jam_kerja": total_jam_kerja, "gaji_pokok_hris": float(emp.get("gaji_pokok_hris") or 4500000), "potongan_per_alfa": potongan_per_alfa, "total_potongan_alfa_rp": total_potongan_alfa_rp, "status_disiplin": status_disiplin, } rekap_rows.append(row_data) totals["total_hadir_tepat"] += hadir_tepat totals["total_terlambat_hari"] += terlambat_hari totals["total_hadir_gabungan"] += total_kehadiran_fisik totals["total_sakit"] += sakit totals["total_izin"] += izin totals["total_cuti"] += cuti totals["total_alfa_murni"] += alfa_murni totals["total_alfa_akumulasi_terlambat"] += alfa_akumulasi_terlambat totals["total_alfa_final_hris"] += total_alfa_hris # Urutkan agar karyawan yang memiliki log kehadiran/keterlambatan tampil di atas rekap_rows.sort( key=lambda r: ( -(r["alfa_akumulasi_terlambat"]), -(r["terlambat_hari"]), -(r["total_kehadiran_fisik"] + r["sakit"] + r["izin"] + r["cuti"] + r["alfa_murni"]), r["nama"], ) ) return { "summary": totals, "rows": rekap_rows, } # ========================================================================== # LINK ABSEN PRIBADI KARYAWAN (diakses dari HP masing-masing) # ========================================================================== GPS_TIDAK_TERSEDIA = "GPS tidak tersedia (tidak terverifikasi)" def get_employee_by_token(token: str): token = (token or "").strip() if not token: return None conn = get_connection() cur = conn.cursor() cur.execute("SELECT * FROM employees WHERE token_absen = ? AND status_aktif = 1", (token,)) row = cur.fetchone() conn.close() return dict(row) if row else None def get_attendance_log(employee_id: int, tanggal: str): conn = get_connection() cur = conn.cursor() cur.execute( "SELECT * FROM attendance_logs WHERE employee_id = ? AND tanggal = ?", (employee_id, tanggal), ) row = cur.fetchone() conn.close() return dict(row) if row else None def get_recent_attendance(employee_id: int, limit: int = 7): conn = get_connection() cur = conn.cursor() cur.execute( """ SELECT tanggal, jam_masuk, jam_keluar, status_kehadiran, menit_terlambat FROM attendance_logs WHERE employee_id = ? ORDER BY tanggal DESC LIMIT ? """, (employee_id, limit), ) rows = [dict(r) for r in cur.fetchall()] conn.close() return rows def get_location_by_id(loc_id: int): conn = get_connection() cur = conn.cursor() cur.execute("SELECT * FROM work_locations WHERE id = ?", (loc_id,)) row = cur.fetchone() conn.close() return dict(row) if row else None def submit_attendance_by_token(token: str, tipe_aksi: str, latitude=None, longitude=None, catatan: str = ""): """ Absen masuk / pulang dari link pribadi karyawan. Tanggal & jam diambil dari server (bukan dari HP) agar tidak bisa dimanipulasi. Lokasi, kategori, dan shift mengikuti data master karyawan. """ emp = get_employee_by_token(token) if not emp: raise ValueError("Link absen tidak valid atau karyawan sudah tidak aktif.") tipe_aksi = (tipe_aksi or "").upper() if tipe_aksi not in ("MASUK", "KELUAR"): raise ValueError("Tipe absen tidak dikenal.") tanggal = date.today().strftime("%Y-%m-%d") existing = get_attendance_log(emp["id"], tanggal) if tipe_aksi == "MASUK": if existing and existing.get("jam_masuk"): raise ValueError(f"Anda sudah absen masuk hari ini pukul {existing['jam_masuk']}.") if existing: raise ValueError( f"Hari ini sudah dicatat HRD dengan status {existing['status_kehadiran']}. " "Hubungi HRD bila ada kesalahan." ) else: if not existing or not existing.get("jam_masuk"): raise ValueError("Anda belum absen masuk hari ini.") if existing.get("jam_keluar"): raise ValueError(f"Anda sudah absen pulang hari ini pukul {existing['jam_keluar']}.") has_gps = latitude not in (None, "") and longitude not in (None, "") payload = { "employee_id": emp["id"], "tanggal": tanggal, "tipe_aksi": tipe_aksi, "kategori_kerja": emp["kategori_kerja"], "lokasi_id": emp["lokasi_default_id"], "shift_kode": emp["shift_default"], "status_kehadiran": "Hadir", "jam_masuk": "", "jam_keluar": "", "latitude": latitude if has_gps else None, "longitude": longitude if has_gps else None, "catatan": (catatan or "").strip() or "Absen via link HP karyawan", } saved = submit_attendance(payload) if not has_gps: conn = get_connection() cur = conn.cursor() if tipe_aksi == "MASUK": cur.execute( "UPDATE attendance_logs SET lat_masuk = NULL, lng_masuk = NULL, " "jarak_masuk_meter = 0, status_gps_masuk = ? WHERE id = ?", (GPS_TIDAK_TERSEDIA, saved["id"]), ) else: cur.execute( "UPDATE attendance_logs SET lat_keluar = NULL, lng_keluar = NULL, " "jarak_keluar_meter = 0, status_gps_keluar = ? WHERE id = ?", (GPS_TIDAK_TERSEDIA, saved["id"]), ) conn.commit() cur.execute("SELECT * FROM attendance_logs WHERE id = ?", (saved["id"],)) saved = dict(cur.fetchone()) conn.close() return saved # ========================================================================== # REKAPITULASI PERIODE (Bulan Lalu | Periode Ini | Akumulasi YTD) & STATISTIK # ========================================================================== from datetime import timedelta KODE_REKAP = ("H", "S", "I", "A", "C", "TL", "X") def _parse_date(value, default: date) -> date: try: return datetime.strptime((value or "").strip(), "%Y-%m-%d").date() except ValueError: return default def _count_workdays(start: date, end: date, cap: date) -> int: """Jumlah hari kerja (Senin-Sabtu) antara start..end, tidak melewati cap (hari ini).""" last = min(end, cap) if start > last: return 0 total = 0 d = start while d <= last: if d.weekday() != 6: # Minggu libur total += 1 d += timedelta(days=1) return total def _empty_counts() -> dict: return {k: 0 for k in KODE_REKAP} def get_rekap_periode( start_date: str = "", end_date: str = "", perusahaan: str = "ALL", departemen: str = "ALL", employee_id: str = "ALL", status_kehadiran: str = "ALL", all_time: bool = False, ): """ Rekap per karyawan untuk tiga periode: - lalu : bulan kalender sebelum bulan 'start_date' - ini : start_date s.d. end_date - ytd : 1 Januari (tahun end_date) s.d. end_date Kode: H=Hadir (termasuk terlambat), S=Sakit, I=Izin, A=Alfa, C=Cuti, TL=Terlambat (bagian dari H), X=hari kerja (Sen-Sab) tanpa catatan absen. """ today = date.today() conn = get_connection() cur = conn.cursor() if all_time: cur.execute("SELECT MIN(tanggal) AS mn FROM attendance_logs") mn = cur.fetchone()["mn"] s = _parse_date(mn, today) e = today else: first_month = date(today.year, today.month, 1) s = _parse_date(start_date, first_month) e = _parse_date(end_date, today) if e < s: s, e = e, s first_of_s = date(s.year, s.month, 1) last_prev = first_of_s - timedelta(days=1) periods = { "lalu": (date(last_prev.year, last_prev.month, 1), last_prev), "ini": (s, e), "ytd": (min(date(e.year, 1, 1), s), e), } workdays = {k: _count_workdays(a, b, today) for k, (a, b) in periods.items()} # Daftar karyawan (untuk filter & tabel) all_emps = list_employees() perusahaan_list = sorted({x["perusahaan"] for x in all_emps if x.get("perusahaan")}) departemen_list = sorted({x["departemen"] for x in all_emps if x.get("departemen")}) employee_options = [ {"id": x["id"], "nama": x["nama"], "nip": x["nip"]} for x in sorted(all_emps, key=lambda z: (z["nama"] or "").lower()) ] emps = list_employees(perusahaan=perusahaan, departemen=departemen) if employee_id and employee_id != "ALL": emps = [x for x in emps if str(x["id"]) == str(employee_id)] lo = min(a for a, _ in periods.values()) hi = max(b for _, b in periods.values()) ids = [x["id"] for x in emps] stats = {i: {k: _empty_counts() for k in periods} for i in ids} logged = {i: {k: 0 for k in periods} for i in ids} if ids: cur.execute( "SELECT employee_id, tanggal, status_kehadiran FROM attendance_logs " "WHERE tanggal >= ? AND tanggal <= ?", (lo.isoformat(), hi.isoformat()), ) for row in cur.fetchall(): eid = row["employee_id"] if eid not in stats: continue try: d = datetime.strptime(row["tanggal"], "%Y-%m-%d").date() except ValueError: continue st = row["status_kehadiran"] for key, (a, b) in periods.items(): if not (a <= d <= b): continue c = stats[eid][key] if st in ("Hadir", "Terlambat"): c["H"] += 1 if st == "Terlambat": c["TL"] += 1 elif st == "Sakit": c["S"] += 1 elif st == "Izin": c["I"] += 1 elif st == "Alfa": c["A"] += 1 elif st == "Cuti": c["C"] += 1 if d.weekday() != 6 and d <= today: logged[eid][key] += 1 conn.close() for i in ids: for key in periods: stats[i][key]["X"] = max(0, workdays[key] - logged[i][key]) rows = [] for x in emps: ini = stats[x["id"]]["ini"] rows.append( { "id": x["id"], "nip": x["nip"], "nama": x["nama"], "perusahaan": x["perusahaan"], "departemen": x["departemen"], "jabatan": x["jabatan"], "lalu": stats[x["id"]]["lalu"], "ini": ini, "ytd": stats[x["id"]]["ytd"], } ) # --- Statistik (dihitung sebelum filter status, berdasarkan periode ini) --- total = len(rows) hadir_orang = sum(1 for r in rows if r["ini"]["H"] > 0) sakit_izin_orang = sum(1 for r in rows if (r["ini"]["S"] + r["ini"]["I"]) > 0) terlambat_orang = sum(1 for r in rows if r["ini"]["TL"] > 0) alfa_orang = sum(1 for r in rows if r["ini"]["A"] > 0) cuti_orang = sum(1 for r in rows if r["ini"]["C"] > 0) tidak_hadir_orang = total - hadir_orang def pct(n, d): return round(n * 100.0 / d, 1) if d else 0.0 hari = {k: sum(r["ini"][k] for r in rows) for k in KODE_REKAP} hari_total = hari["H"] + hari["S"] + hari["I"] + hari["A"] + hari["C"] + hari["X"] summary = { "total_karyawan": total, "hadir_orang": hadir_orang, "sakit_izin_orang": sakit_izin_orang, "sakit_hari": hari["S"], "izin_hari": hari["I"], "tidak_hadir_orang": tidak_hadir_orang, "pct_hadir": pct(hadir_orang, total), "pct_tidak_hadir": pct(tidak_hadir_orang, total), "terlambat_orang": terlambat_orang, "alfa_orang": alfa_orang, "cuti_orang": cuti_orang, "pct_hadir_hari_kerja": pct(hari["H"], hari_total), "hari": hari, } # --- Rincian per perusahaan --- per_perusahaan = [] by_company = {} for r in rows: by_company.setdefault(r["perusahaan"], []).append(r) for nama in sorted(by_company): grp = by_company[nama] n = len(grp) h = sum(1 for r in grp if r["ini"]["H"] > 0) per_perusahaan.append( { "perusahaan": nama, "jumlah_karyawan": n, "hadir_orang": h, "tidak_hadir_orang": n - h, "pct_hadir": pct(h, n), "pct_tidak_hadir": pct(n - h, n), } ) # --- Filter status hanya untuk tabel --- sf = (status_kehadiran or "ALL").strip() if sf and sf != "ALL": def match(r): c = r["ini"] if sf == "Hadir": return c["H"] > 0 if sf == "Terlambat": return c["TL"] > 0 if sf == "Sakit": return c["S"] > 0 if sf == "Izin": return c["I"] > 0 if sf == "Cuti": return c["C"] > 0 if sf == "Alfa": return c["A"] > 0 if sf == "Tidak Absen": return (c["H"] + c["S"] + c["I"] + c["A"] + c["C"]) == 0 return True rows = [r for r in rows if match(r)] return { "periode": {k: {"start": a.isoformat(), "end": b.isoformat(), "hari_kerja": workdays[k]} for k, (a, b) in periods.items()}, "start_date": s.isoformat(), "end_date": e.isoformat(), "summary": summary, "per_perusahaan": per_perusahaan, "rows": rows, "perusahaan_list": perusahaan_list, "departemen_list": departemen_list, "employee_options": employee_options, }