import gradio as gr import pandas as pd import numpy as np from datetime import datetime, date, time, timedelta from pathlib import Path import os # ===== Policy ===== POLICY_CHANGE = date(2025, 7, 7) # inclusive BASELINE_BEFORE = time(17, 30) # BEFORE 2025-07-07 BASELINE_AFTER = time(17, 0) # FROM 2025-07-07 VIRTUAL_BEFORE = time(18, 0) # Case 4 floor (BEFORE) VIRTUAL_AFTER = time(17, 30) # Case 4 floor (FROM) IN_TOKENS = {"P10","IN","INN","CHECKIN","CHECK-IN"} OUT_TOKENS = {"P20","OUT","CHECKOUT","CHECK-OUT"} SMALL_GAP_MERGE_SEC = 30 * 60 # merge OUT→IN gaps under 30 minutes # ===== Helpers ===== def parse_yyyymmdd(s): s = str(s).strip() for fmt in ("%Y%m%d","%Y-%m-%d","%d/%m/%Y","%m/%d/%Y"): try: return datetime.strptime(s, fmt).date() except Exception: pass return pd.NaT def parse_hms(s): s = str(s).strip() if s.isdigit(): s = s.zfill(6)[:6] hh,mm,ss = s[:2], s[2:4], s[4:6] try: return time(int(hh), int(mm), int(ss)) except Exception: return pd.NaT for fmt in ("%H:%M:%S","%H:%M"): try: return datetime.strptime(s, fmt).time() except Exception: pass return pd.NaT def combine_dt(d, t): if pd.isna(d) or pd.isna(t): return pd.NaT return datetime.combine(d, t) def from_policy(d): return (not pd.isna(d)) and (d >= POLICY_CHANGE) def baseline_cutoff(d): return BASELINE_AFTER if from_policy(d) else BASELINE_BEFORE def virtual_checkin(d): return VIRTUAL_AFTER if from_policy(d) else VIRTUAL_BEFORE def detect_event_token(v): u = str(v).upper() is_in = any(tok == u or tok in u for tok in IN_TOKENS) is_out = any(tok == u or tok in u for tok in OUT_TOKENS) if is_in and not is_out: return "in" if is_out and not is_in: return "out" return np.nan def uniq_times_str(ts): return ", ".join(sorted({t.strftime("%H:%M:%S") for t in ts.dropna()})) def fmt_time(x): if pd.isna(x): return "" if isinstance(x, datetime): return x.strftime("%H:%M:%S") if isinstance(x, time): return x.strftime("%H:%M:%S") return str(x) # Collapse runs of identical events: # - keep earliest time of an IN-run # - keep latest time of an OUT-run def compress_events(events): # events: list[(ts, 'in'|'out')] ascending if not events: return [] out = [] for ts, ev in events: if not out: out.append([ts, ev]) continue last_ts, last_ev = out[-1] if ev == last_ev: if ev == "out": out[-1][0] = ts # extend OUT-run to latest OUT # IN-run: keep earliest else: out.append([ts, ev]) return [(ts, ev) for ts, ev in out] # Merge small OFF gaps: remove OUT→IN pairs with gap < threshold def merge_small_gaps(events, threshold_sec=SMALL_GAP_MERGE_SEC): # events must be compressed already. merged = [] i = 0 n = len(events) while i < n: ts, ev = events[i] if ev == "out" and merged and merged[-1][1] == "in" and (i + 1) < n and events[i+1][1] == "in": gap = (events[i+1][0] - ts).total_seconds() if gap < threshold_sec: i += 2 continue merged.append([ts, ev]) i += 1 return [(ts, ev) for ts, ev in merged] # Summarize OT between start_time and end_time on a clean event stream def summarize_ot(start_time, end_time, events, start_state): # returns: (work_total_sec, list_internal_off_gaps_sec, segments_count) work_total = 0.0 segments = 0 off_gaps = [] cur_state = start_state cur = start_time off_start = None for ts, ev in events: if ts <= start_time: continue if ts > end_time: break if cur_state == "in": work_total += (ts - cur).total_seconds() if ev == "out": segments += 1 off_start = ts else: if ev == "in" and off_start is not None: off_gaps.append((ts - off_start).total_seconds()) off_start = None cur_state = ev cur = ts if cur < end_time and cur_state == "in": work_total += (end_time - cur).total_seconds() return work_total, off_gaps, segments # ===== Core computation ===== def compute_ot(df_raw: pd.DataFrame) -> pd.DataFrame: df = df_raw.copy() df.columns = [str(c).strip().lower() for c in df.columns] # detect 3 required columns heuristically def find_col(df, candidates, fallback=None): for col in df.columns: if any(pat in col for pat in candidates): return col return fallback emp_col = find_col(df, ["emp","employee","staff","id","person"], df.columns[0] if len(df.columns)>0 else None) date_col = find_col(df, ["date","yyyymmdd","ngay"], df.columns[1] if len(df.columns)>1 else None) time_col = find_col(df, ["time","gio","hh","clock"], df.columns[2] if len(df.columns)>2 else None) if any(c is None for c in [emp_col, date_col, time_col]): raise gr.Error(f"Expected 3 columns: employee, yyyymmdd, hh:mm:ss (or hhmmss). Got {list(df.columns)}") # status column (optional) evt_col = None for c in df.columns: vals = df[c].astype(str).str.upper() if vals.isin(["P10","P20"]).any() or vals.str.contains(r"\bP10\b|\bP20\b", regex=True, na=False).any(): evt_col = c break df["__emp"] = df[emp_col].astype(str).str.strip() df["__date"] = df[date_col].apply(parse_yyyymmdd) df["__time"] = df[time_col].apply(parse_hms) df["__ts"] = [combine_dt(d, t) for d, t in zip(df["__date"], df["__time"])] if evt_col: df["__event"] = df[evt_col].apply(detect_event_token) else: def scan_row(row): for c in row.index: ev = detect_event_token(row[c]) if pd.notna(ev): return ev return np.nan df["__event"] = df.apply(scan_row, axis=1) df = df.dropna(subset=["__ts"]).sort_values(["__emp","__date","__ts"]).reset_index(drop=True) df["__baseline_dt"] = [combine_dt(d, baseline_cutoff(d)) for d in df["__date"]] df["__virtual_dt"] = [combine_dt(d, virtual_checkin(d)) for d in df["__date"]] rows = [] for (emp, d), day in df.groupby(["__emp","__date"], sort=True): if pd.isna(d): continue day = day.sort_values("__ts") baseline_dt = day["__baseline_dt"].iloc[0] virtual_dt = day["__virtual_dt"].iloc[0] pre = day[day["__ts"] < baseline_dt] after = day[day["__ts"] >= baseline_dt] if after.empty: continue last_pre_event = pre.iloc[-1]["__event"] if not pre.empty else np.nan ins_after = after.loc[after["__event"].eq("in"), "__ts"] outs_after = after.loc[after["__event"].eq("out"), "__ts"] in_times_list = uniq_times_str(ins_after) out_times_list = uniq_times_str(outs_after) code = "" ot_hours = np.nan virtual_used_str = "" # Need at least one OUT after cutoff if len(outs_after) == 0: if len(ins_after) > 0: code = "alert" continue end_dt = outs_after.max() # Build + clean events after cutoff up to last OUT events_df = after.loc[(after["__ts"] <= end_dt) & (after["__event"].isin(["in","out"])), ["__ts","__event"]] events_raw = list(events_df.itertuples(index=False, name=None)) events = merge_small_gaps(compress_events(events_raw), threshold_sec=SMALL_GAP_MERGE_SEC) if len(ins_after) == 0: # No IN after cutoff if last_pre_event == "in": # Case 4: inside at cutoff; baseline minus OFF gaps, floored by virtual work_secs, _, _ = summarize_ot(baseline_dt, end_dt, events, start_state="in") eff_start = end_dt - timedelta(seconds=work_secs) start_dt = max(eff_start, virtual_dt) ot_hours = round(max(0.0, (end_dt - start_dt).total_seconds())/3600.0, 2) virtual_used_str = fmt_time(start_dt) else: # Case 2: from baseline ot_hours = round(max(0.0, (end_dt - baseline_dt).total_seconds())/3600.0, 2) else: # Have both IN and OUT after cutoff first_after_ev = after.iloc[0]["__event"] inside_at_cutoff = (last_pre_event == "in") or (first_after_ev == "out") if inside_at_cutoff: # Case 4: baseline minus gaps, floored by virtual work_secs, _, _ = summarize_ot(baseline_dt, end_dt, events, start_state="in") eff_start = end_dt - timedelta(seconds=work_secs) start_dt = max(eff_start, virtual_dt) ot_hours = round(max(0.0, (end_dt - start_dt).total_seconds())/3600.0, 2) virtual_used_str = fmt_time(start_dt) else: # Case 1: earliest IN after cutoff, subtract OFF gaps start_dt = ins_after.min() work_secs, _, _ = summarize_ot(start_dt, end_dt, events, start_state="in") ot_hours = round(max(0.0, work_secs)/3600.0, 2) rows.append({ "employee_id": emp, "date": d.strftime("%Y%m%d"), "ot_hours": ot_hours, "code": code, "in_times": in_times_list, "out_times": out_times_list, "virtual_in_time": virtual_used_str }) res = pd.DataFrame(rows, columns=[ "employee_id","date","ot_hours","code","in_times","out_times","virtual_in_time" ]) if not res.empty: res = res.sort_values(["employee_id","date"]).reset_index(drop=True) return res # ===== I/O wrappers ===== def load_dataframe(upload): # Gradio v4 passes a FileData object with .path; also accept str/Path if upload is None: raise gr.Error("Please upload a file.") path = None if isinstance(upload, (str, Path)): path = str(upload) elif hasattr(upload, "path"): path = upload.path elif hasattr(upload, "name") and os.path.exists(upload.name): path = upload.name if not path or not os.path.exists(path): raise gr.Error("Could not resolve uploaded file path.") lower = path.lower() if lower.endswith((".xlsx",".xls")): df = pd.read_excel(path, dtype=str) elif lower.endswith(".csv"): try: df = pd.read_csv(path, dtype=str) except Exception: df = pd.read_csv(path, dtype=str, sep=";") else: raise gr.Error("Unsupported file type. Upload .xlsx/.xls/.csv.") return df def predict(upload): df_in = load_dataframe(upload) res = compute_ot(df_in) out_path = Path("ot_summary.xlsx") try: import xlsxwriter; engine = "xlsxwriter" except Exception: engine = "openpyxl" with pd.ExcelWriter(out_path, engine=engine) as w: res.to_excel(w, index=False, sheet_name="OT") return res, str(out_path) # Tiny example so users can test without uploading example_path = Path("example.csv") if not example_path.exists(): example_path.write_text( "emp,date,time,status\n" "3809273,20250704,125718,P10\n" "3809273,20250704,174514,P20\n" "3809273,20250704,174553,P10\n" "3809273,20250704,174608,P20\n" "3809273,20250704,174631,P10\n" "3809273,20250704,220230,P20\n", encoding="utf-8" ) # ===== Gradio app ===== iface = gr.Interface( fn=predict, inputs=gr.File(label="Upload Excel/CSV", file_types=[".xlsx",".xls",".csv"]), outputs=[ gr.Dataframe(label="Result preview", row_count=(12, "dynamic")), gr.File(label="Download OT Excel (.xlsx)") ], title="OT Calculator (P10/P20)", description=( "Policy: cutoff 17:30 before 2025-07-07, 17:00 from 2025-07-07. " "Case 4 floors start at 18:00 (before) or 17:30 (from). " "Consecutive duplicate events are collapsed; OUT→IN gaps < 30′ are merged." ), flagging_mode="never", examples=[[str(example_path)]], ) if __name__ == "__main__": iface.launch()