samurice commited on
Commit
fa92c08
·
verified ·
1 Parent(s): a9b8d7d

Upload folder using huggingface_hub

Browse files
Files changed (2) hide show
  1. app.py +1 -226
  2. requirements.txt +5 -5
app.py CHANGED
@@ -1,226 +1 @@
1
- import pandas as pd
2
- import numpy as np
3
- from datetime import datetime, date, time
4
- from pathlib import Path
5
-
6
- # ===================== USER CONFIG =====================
7
- INPUT_XLSX = "data.xlsx" # your input file
8
- OUTPUT_XLSX = "ot_summary.xlsx" # output path
9
-
10
- # If your headers differ, you can set them here (case-insensitive). Otherwise auto-detect.
11
- EMPLOYEE_COL = None # e.g., "employee_id"
12
- DATE_COL = None # e.g., "date"
13
- TIME_COL = None # e.g., "time"
14
- EVENT_COL = None # e.g., "status" (values like P10/P20), or leave None to auto-detect
15
-
16
- # Policy:
17
- # - Baseline cutoff: 17:00 before 2025-07-07; 17:30 from 2025-07-07
18
- # - Case 4 (virtual check-in): 17:30 before 2025-07-07; 18:00 from 2025-07-07
19
- POLICY_CHANGE = date(2025, 7, 7)
20
- BASELINE_BEFORE = time(17, 0)
21
- BASELINE_AFTER = time(17, 30)
22
- VIRTUAL_BEFORE = time(17, 30)
23
- VIRTUAL_AFTER = time(18, 0)
24
- # =======================================================
25
-
26
- # ----------------- helpers -----------------
27
- def parse_yyyymmdd(s):
28
- s = str(s).strip()
29
- for fmt in ("%Y%m%d", "%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"):
30
- try:
31
- return datetime.strptime(s, fmt).date()
32
- except Exception:
33
- pass
34
- return pd.NaT
35
-
36
- def parse_hms(s):
37
- """
38
- Accept 'HH:MM[:SS]', 'HHMM', 'HHMMSS', or 5-digit like '82547' -> 08:25:47.
39
- """
40
- s = str(s).strip()
41
- if s.isdigit():
42
- s = s.zfill(6)[:6] # left-pad to 6 digits and cap at 6
43
- hh, mm, ss = s[:2], s[2:4], s[4:6]
44
- try:
45
- return time(int(hh), int(mm), int(ss))
46
- except Exception:
47
- return pd.NaT
48
- for fmt in ("%H:%M:%S", "%H:%M"):
49
- try:
50
- return datetime.strptime(s, fmt).time()
51
- except Exception:
52
- pass
53
- return pd.NaT
54
-
55
- def combine_dt(d, t):
56
- if pd.isna(d) or pd.isna(t):
57
- return pd.NaT
58
- return datetime.combine(d, t)
59
-
60
- def is_from_policy(d):
61
- return (not pd.isna(d)) and (d >= POLICY_CHANGE)
62
-
63
- def baseline_cutoff_time(d):
64
- return BASELINE_AFTER if is_from_policy(d) else BASELINE_BEFORE
65
-
66
- def virtual_checkin_time(d):
67
- return VIRTUAL_AFTER if is_from_policy(d) else VIRTUAL_BEFORE
68
-
69
- IN_TOKENS = {"P10", "IN", "INN", "CHECKIN", "CHECK-IN"}
70
- OUT_TOKENS = {"P20", "OUT", "CHECKOUT", "CHECK-OUT"}
71
-
72
- def detect_event_token(v):
73
- u = str(v).upper()
74
- is_in = any(tok == u or tok in u for tok in IN_TOKENS)
75
- is_out = any(tok == u or tok in u for tok in OUT_TOKENS)
76
- if is_in and not is_out:
77
- return "in"
78
- if is_out and not is_in:
79
- return "out"
80
- return np.nan
81
-
82
- def format_unique_times(ts_series):
83
- uniq = sorted({t.strftime("%H:%M:%S") for t in ts_series.dropna()})
84
- return ", ".join(uniq)
85
-
86
- # ----------------- load & normalize -----------------
87
- src = pd.read_excel(INPUT_XLSX, dtype=str)
88
- src.columns = [str(c).strip().lower() for c in src.columns]
89
-
90
- def find_col(candidates, fallback=None):
91
- for col in src.columns:
92
- for pat in candidates:
93
- if pat in col:
94
- return col
95
- return fallback
96
-
97
- emp_col = (EMPLOYEE_COL.lower() if EMPLOYEE_COL else None) or find_col(["emp", "employee", "staff", "id", "person"], src.columns[0] if len(src.columns)>0 else None)
98
- date_col = (DATE_COL.lower() if DATE_COL else None) or find_col(["date", "yyyymmdd", "ngay"], src.columns[1] if len(src.columns)>1 else None)
99
- time_col = (TIME_COL.lower() if TIME_COL else None) or find_col(["time", "gio", "hh", "clock"], src.columns[2] if len(src.columns)>2 else None)
100
-
101
- # try to guess a P10/P20 column if not specified
102
- if EVENT_COL and EVENT_COL.lower() in src.columns:
103
- evt_col = EVENT_COL.lower()
104
- else:
105
- evt_col = None
106
- for c in src.columns:
107
- vals = src[c].astype(str).str.upper()
108
- if vals.isin(["P10","P20"]).any() or vals.str.contains(r"\bP10\b|\bP20\b", regex=True).any():
109
- evt_col = c
110
- break
111
-
112
- if emp_col not in src.columns or date_col not in src.columns or time_col not in src.columns:
113
- raise ValueError(f"Could not find expected columns. Found: {list(src.columns)}")
114
-
115
- work = src.copy()
116
- work["__emp"] = work[emp_col].astype(str).str.strip()
117
- work["__date"] = work[date_col].apply(parse_yyyymmdd)
118
- work["__time"] = work[time_col].apply(parse_hms)
119
- work["__ts"] = [combine_dt(d, t) for d, t in zip(work["__date"], work["__time"])]
120
-
121
- # detect in/out (prefer dedicated event column; otherwise scan)
122
- if evt_col:
123
- work["__event"] = work[evt_col].apply(detect_event_token)
124
- else:
125
- def scan_row(row):
126
- for c in row.index:
127
- ev = detect_event_token(row[c])
128
- if pd.notna(ev):
129
- return ev
130
- return np.nan
131
- work["__event"] = work.apply(scan_row, axis=1)
132
-
133
- # build baseline & virtual cutoffs
134
- work["__baseline_t"] = work["__date"].apply(baseline_cutoff_time)
135
- work["__baseline_dt"] = [combine_dt(d, t) for d, t in zip(work["__date"], work["__baseline_t"])]
136
- work["__virtual_t"] = work["__date"].apply(virtual_checkin_time)
137
- work["__virtual_dt"] = [combine_dt(d, t) for d, t in zip(work["__date"], work["__virtual_t"])]
138
-
139
- # ----------------- compute OT per employee/date -----------------
140
- rows = []
141
- for (emp, d), full_day in work.groupby(["__emp", "__date"], sort=True):
142
- if pd.isna(d):
143
- continue
144
-
145
- # Sort all events of the day
146
- full_day = full_day.sort_values("__ts")
147
-
148
- baseline_dt = full_day["__baseline_dt"].iloc[0]
149
- virtual_dt = full_day["__virtual_dt"].iloc[0]
150
-
151
- # Determine if the person was already inside at the cutoff:
152
- pre = full_day[full_day["__ts"] < baseline_dt]
153
- if not pre.empty:
154
- last_pre_event = pre.iloc[-1]["__event"]
155
- inside_at_cutoff = (last_pre_event == "in")
156
- else:
157
- inside_at_cutoff = False
158
-
159
- # Consider only events at/after the baseline cutoff for outputs
160
- after = full_day[full_day["__ts"] >= baseline_dt].copy()
161
-
162
- # Unique in/out time strings (AFTER cutoff) for the report
163
- ins_after = after.loc[after["__event"].eq("in"), "__ts"]
164
- outs_after = after.loc[after["__event"].eq("out"), "__ts"]
165
- in_times_str = format_unique_times(ins_after)
166
- out_times_str = format_unique_times(outs_after)
167
-
168
- # Compute OT per policy
169
- code = ""
170
- ot_hours = np.nan
171
-
172
- if len(ins_after) > 0 and len(outs_after) > 0:
173
- # Case 1: both in and out after cutoff → earliest IN vs latest OUT
174
- earliest_in = ins_after.min()
175
- latest_out = outs_after.max()
176
- if pd.notna(earliest_in) and pd.notna(latest_out) and latest_out >= earliest_in:
177
- ot_hours = (latest_out - earliest_in).total_seconds() / 3600.0
178
- else:
179
- code = "alert"
180
-
181
- elif len(ins_after) == 0 and len(outs_after) > 0:
182
- # Case 2 or Case 4:
183
- # If they were already inside at the cutoff → Case 4 (use virtual check-in)
184
- # Else → Case 2 (use baseline)
185
- start_dt = virtual_dt if inside_at_cutoff else baseline_dt
186
- latest_out = outs_after.max()
187
- if pd.notna(latest_out) and latest_out >= start_dt:
188
- ot_hours = (latest_out - start_dt).total_seconds() / 3600.0
189
- else:
190
- ot_hours = 0.0 # guarded by baseline filter; keep 0 if anomalous
191
-
192
- elif len(ins_after) > 0 and len(outs_after) == 0:
193
- # Case 3: has IN but no OUT after cutoff
194
- code = "alert"
195
-
196
- else:
197
- # No events after cutoff → nothing to report
198
- continue
199
-
200
- ot_hours = round(float(ot_hours), 2) if pd.notna(ot_hours) else np.nan
201
-
202
- rows.append({
203
- "employee_id": emp,
204
- "date": d.strftime("%Y%m%d"),
205
- "ot_hours": ot_hours,
206
- "code": code,
207
- "in_times": in_times_str,
208
- "out_times": out_times_str
209
- })
210
-
211
- res = pd.DataFrame(rows, columns=["employee_id","date","ot_hours","code","in_times","out_times"])
212
- if not res.empty:
213
- res = res.sort_values(by=["employee_id","date"]).reset_index(drop=True)
214
-
215
- # ----------------- write Excel (Colab-friendly) -----------------
216
- try:
217
- import xlsxwriter # noqa: F401
218
- engine = "xlsxwriter"
219
- except Exception:
220
- engine = "openpyxl" # available by default in Colab
221
-
222
- with pd.ExcelWriter(OUTPUT_XLSX, engine=engine) as w:
223
- res.to_excel(w, index=False, sheet_name="OT")
224
-
225
- print(f"Done. Wrote: {Path(OUTPUT_XLSX).resolve()}")
226
- display(res.head(50))
 
1
+ <PASTE THE WHOLE app.py CONTENT HERE>
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
requirements.txt CHANGED
@@ -1,5 +1,5 @@
1
- gradio>=4.29.0,<5
2
- pandas>=2.0.0
3
- numpy>=1.24.0
4
- openpyxl>=3.1.0
5
- XlsxWriter>=3.1.0
 
1
+ gradio>=4.29.0,<5
2
+ pandas>=2.0.0
3
+ numpy>=1.24.0
4
+ openpyxl>=3.1.0
5
+ XlsxWriter>=3.1.0