ir/aars
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

If late/early/OT > 30min → mark as abnormal status (override status_code+text, zero out minutes)

Browse Source
This commit is contained in:
IT狗
2026-07-21 17:52:36 +08:00
parent f41dad6127
commit ef57d7cd59
1 changed files with 119 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/main.py
+119 -105
View File
@@ -267,13 +267,13 @@ def read_excel_file(section: str):
path = get_excel_path(section) path = get_excel_path(section)
if not os.path.exists(path): if not os.path.exists(path):
return None, [] return None, []
wb = openpyxl.load_workbook(path, data_only=True) wb = openpyxl.load_workbook(path, data_only=True)
ws = wb.active ws = wb.active
headers = [cell.value for cell in ws[1]] headers = [cell.value for cell in ws[1]]
rows = list(ws.iter_rows(min_row=2, values_only=True)) rows = list(ws.iter_rows(min_row=2, values_only=True))
return headers, rows return headers, rows
# ============ Upload Excel (Replace entire file) ============ # ============ Upload Excel (Replace entire file) ============
@@ -285,11 +285,11 @@ async def delete_uploaded_file(
"""Delete the stored Excel file for a section""" """Delete the stored Excel file for a section"""
if section not in ["attendance", "accident"]: if section not in ["attendance", "accident"]:
raise HTTPException(status_code=400, detail="Invalid section") raise HTTPException(status_code=400, detail="Invalid section")
path = get_excel_path(section) path = get_excel_path(section)
if os.path.exists(path): if os.path.exists(path):
os.remove(path) os.remove(path)
return {"message": f"{section} file deleted"} return {"message": f"{section} file deleted"}
@app.post("/api/upload/{section}") @app.post("/api/upload/{section}")
@@ -299,7 +299,7 @@ async def upload_excel(
db: Session = Depends(get_db), db: Session = Depends(get_db),
current_user: User = Depends(get_current_user) current_user: User = Depends(get_current_user)
): ):
"""Upload Excel file — saves file AND imports rows into SQL table.""" """Upload Excel file - saves file AND imports rows into SQL table."""
if section not in ["attendance", "accident"]: if section not in ["attendance", "accident"]:
raise HTTPException(status_code=400, detail="Invalid section") raise HTTPException(status_code=400, detail="Invalid section")
@@ -538,13 +538,13 @@ def get_shift_schedule_full(shift_code: str, weekday: str) -> Optional[tuple]:
roster_path = get_excel_path("roster") roster_path = get_excel_path("roster")
if not os.path.exists(roster_path): if not os.path.exists(roster_path):
return None return None
import openpyxl import openpyxl
wb = openpyxl.load_workbook(roster_path, data_only=True) wb = openpyxl.load_workbook(roster_path, data_only=True)
ws = wb.active ws = wb.active
headers = [cell.value for cell in ws[1]] headers = [cell.value for cell in ws[1]]
# Find column indices # Find column indices
shift_col = None shift_col = None
ph_col = None ph_col = None
@@ -553,10 +553,10 @@ def get_shift_schedule_full(shift_code: str, weekday: str) -> Optional[tuple]:
shift_col = i shift_col = i
elif h == 'public holiday': elif h == 'public holiday':
ph_col = i ph_col = i
if shift_col is None: if shift_col is None:
return None return None
# Weekday mapping # Weekday mapping
weekday_map = { weekday_map = {
'Monday': '星期一', 'Tuesday': '星期二', 'Wednesday': '星期三', 'Monday': '星期一', 'Tuesday': '星期二', 'Wednesday': '星期三',
@@ -565,18 +565,18 @@ def get_shift_schedule_full(shift_code: str, weekday: str) -> Optional[tuple]:
'星期三': '星期三', '星期四': '星期四', '星期五': '星期五', '星期三': '星期三', '星期四': '星期四', '星期五': '星期五',
'星期六': '星期六', '星期日': '星期日' '星期六': '星期六', '星期日': '星期日'
} }
day_col_map = { day_col_map = {
'星期一': None, '星期二': None, '星期三': None, '星期一': None, '星期二': None, '星期三': None,
'星期四': None, '星期五': None, '星期六': None, '星期日': None '星期四': None, '星期五': None, '星期六': None, '星期日': None
} }
for i, h in enumerate(headers): for i, h in enumerate(headers):
if h in day_col_map: if h in day_col_map:
day_col_map[h] = i day_col_map[h] = i
target_day = weekday_map.get(weekday, weekday) target_day = weekday_map.get(weekday, weekday)
# Find the shift row # Find the shift row
for row in ws.iter_rows(min_row=2, values_only=True): for row in ws.iter_rows(min_row=2, values_only=True):
if row[shift_col] == shift_code: if row[shift_col] == shift_code:
@@ -596,7 +596,7 @@ def get_shift_schedule_full(shift_code: str, weekday: str) -> Optional[tuple]:
return (start_time, end_time, public_holiday) return (start_time, end_time, public_holiday)
except: except:
pass pass
return None return None
def parse_datetime(time_val) -> Optional[datetime]: def parse_datetime(time_val) -> Optional[datetime]:
@@ -615,7 +615,7 @@ def parse_datetime(time_val) -> Optional[datetime]:
def calculate_attendance_status(check_in, check_out, shift_start, shift_end): def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
"""Calculate attendance status: late_minutes, early_minutes, ot_minutes, is_missing """Calculate attendance status: late_minutes, early_minutes, ot_minutes, is_missing
Returns: { Returns: {
"is_missing": bool, "is_missing": bool,
"late_minutes": int, "late_minutes": int,
@@ -641,7 +641,7 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
"actual_in": "-", "actual_in": "-",
"actual_out": "-" "actual_out": "-"
} }
# Check for missing punch # Check for missing punch
if not check_in or not check_out: if not check_in or not check_out:
result["is_missing"] = True result["is_missing"] = True
@@ -652,20 +652,20 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
if check_out: if check_out:
result["actual_out"] = check_out.strftime("%H:%M") if isinstance(check_out, datetime) else str(check_out) result["actual_out"] = check_out.strftime("%H:%M") if isinstance(check_out, datetime) else str(check_out)
return result return result
# Parse check-in/out times # Parse check-in/out times
check_in_dt = parse_datetime(check_in) check_in_dt = parse_datetime(check_in)
check_out_dt = parse_datetime(check_out) check_out_dt = parse_datetime(check_out)
if not check_in_dt or not check_out_dt: if not check_in_dt or not check_out_dt:
result["is_missing"] = True result["is_missing"] = True
result["status_code"] = "missing" result["status_code"] = "missing"
result["status_text"] = "缺勤" result["status_text"] = "缺勤"
return result return result
result["actual_in"] = check_in_dt.strftime("%H:%M") result["actual_in"] = check_in_dt.strftime("%H:%M")
result["actual_out"] = check_out_dt.strftime("%H:%M") result["actual_out"] = check_out_dt.strftime("%H:%M")
# Skip if invalid times (Excel default "0" = 18:00:00 exactly) # Skip if invalid times (Excel default "0" = 18:00:00 exactly)
# Only treat second-precise 18:00:00 as the default, not 18:00:03 (real swipe) # Only treat second-precise 18:00:00 as the default, not 18:00:03 (real swipe)
if (check_in_dt.hour == 18 and check_in_dt.minute == 0 and check_in_dt.second == 0 if (check_in_dt.hour == 18 and check_in_dt.minute == 0 and check_in_dt.second == 0
@@ -674,7 +674,7 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
result["status_code"] = "missing" result["status_code"] = "missing"
result["status_text"] = "缺勤" result["status_text"] = "缺勤"
return result return result
# Detect same check-in/out (exact same second = forgotten swipe) # Detect same check-in/out (exact same second = forgotten swipe)
if check_in_dt and check_out_dt: if check_in_dt and check_out_dt:
if check_in_dt == check_out_dt: if check_in_dt == check_out_dt:
@@ -686,14 +686,14 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
if not shift_start or not shift_end: if not shift_start or not shift_end:
# No roster data, can't calculate # No roster data, can't calculate
return result return result
# Calculate late minutes # Calculate late minutes
check_in_time = check_in_dt.time() check_in_time = check_in_dt.time()
if check_in_time > shift_start: if check_in_time > shift_start:
diff = (datetime.combine(check_in_dt.date(), check_in_time) - diff = (datetime.combine(check_in_dt.date(), check_in_time) -
datetime.combine(check_in_dt.date(), shift_start)).total_seconds() / 60 datetime.combine(check_in_dt.date(), shift_start)).total_seconds() / 60
result["late_minutes"] = round(diff) result["late_minutes"] = round(diff)
# Calculate early minutes (leaving before scheduled end) # Calculate early minutes (leaving before scheduled end)
check_out_time = check_out_dt.time() check_out_time = check_out_dt.time()
if shift_end > shift_start: if shift_end > shift_start:
@@ -703,11 +703,11 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
expected_end_dt = datetime.combine(check_out_dt.date() + timedelta(days=1), shift_end) expected_end_dt = datetime.combine(check_out_dt.date() + timedelta(days=1), shift_end)
else: else:
expected_end_dt = datetime.combine(check_out_dt.date(), shift_end) expected_end_dt = datetime.combine(check_out_dt.date(), shift_end)
if check_out_dt < expected_end_dt: if check_out_dt < expected_end_dt:
diff_seconds = (expected_end_dt - check_out_dt).total_seconds() diff_seconds = (expected_end_dt - check_out_dt).total_seconds()
result["early_minutes"] = round(diff_seconds / 60) if diff_seconds >= 30 else 0 result["early_minutes"] = round(diff_seconds / 60) if diff_seconds >= 30 else 0
# Calculate OT minutes (leaving after scheduled end) # Calculate OT minutes (leaving after scheduled end)
if check_out_dt > expected_end_dt: if check_out_dt > expected_end_dt:
diff_seconds = (check_out_dt - expected_end_dt).total_seconds() diff_seconds = (check_out_dt - expected_end_dt).total_seconds()
@@ -716,7 +716,7 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
result["ot_minutes"] = round(diff_seconds / 60) result["ot_minutes"] = round(diff_seconds / 60)
else: else:
result["ot_minutes"] = 0 result["ot_minutes"] = 0
# Determine status code and text # Determine status code and text
status_parts = [] status_parts = []
if result["late_minutes"] > 0: if result["late_minutes"] > 0:
@@ -725,7 +725,7 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
status_parts.append(f"早退{result['early_minutes']}分") status_parts.append(f"早退{result['early_minutes']}分")
if result["ot_minutes"] > 0: if result["ot_minutes"] > 0:
status_parts.append(f"OT{result['ot_minutes']}分") status_parts.append(f"OT{result['ot_minutes']}分")
if result["late_minutes"] > 0 and result["early_minutes"] > 0: if result["late_minutes"] > 0 and result["early_minutes"] > 0:
result["status_code"] = "late_early" result["status_code"] = "late_early"
elif result["late_minutes"] > 0 and result["ot_minutes"] > 0: elif result["late_minutes"] > 0 and result["ot_minutes"] > 0:
@@ -740,8 +740,22 @@ def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
result["status_code"] = "ot" result["status_code"] = "ot"
else: else:
result["status_code"] = "normal" result["status_code"] = "normal"
result["status_text"] = " / ".join(status_parts) if status_parts else "正常" result["status_text"] = " / ".join(status_parts) if status_parts else "正常"
# If late / early / OT > 30 min → mark as abnormal (override after status_text is set)
if result["late_minutes"] > 30 or result["early_minutes"] > 30 or result["ot_minutes"] > 30:
result["status_code"] = "abnormal"
if result["late_minutes"] > 30:
result["status_text"] = f"異常-遲到{result['late_minutes']}分"
elif result["early_minutes"] > 30:
result["status_text"] = f"異常-早退{result['early_minutes']}分"
else:
result["status_text"] = f"異常-OT{result['ot_minutes']}分"
result["late_minutes"] = 0
result["early_minutes"] = 0
result["ot_minutes"] = 0
return result return result
@@ -751,9 +765,9 @@ async def get_lateness_stats(current_user: User = Depends(get_current_user)):
headers, rows = read_excel_file("attendance") headers, rows = read_excel_file("attendance")
if headers is None or not rows: if headers is None or not rows:
return {"stats": [], "total_late": 0, "total_early": 0, "total_ot": 0, "total_missing": 0} return {"stats": [], "total_late": 0, "total_early": 0, "total_ot": 0, "total_missing": 0}
staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None
for i, h in enumerate(headers): for i, h in enumerate(headers):
h_lower = str(h).lower() if h else '' h_lower = str(h).lower() if h else ''
if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower: if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower:
@@ -768,31 +782,31 @@ async def get_lateness_stats(current_user: User = Depends(get_current_user)):
checkout_col = i checkout_col = i
elif '班次' in h_lower: elif '班次' in h_lower:
shift_col = i shift_col = i
if staff_col is None or checkin_col is None or shift_col is None: if staff_col is None or checkin_col is None or shift_col is None:
return {"error": "Missing required columns", "headers": headers} return {"error": "Missing required columns", "headers": headers}
employee_stats = {} employee_stats = {}
totals = {"late": 0, "early": 0, "ot": 0, "missing": 0} totals = {"late": 0, "early": 0, "ot": 0, "missing": 0}
for row in rows: for row in rows:
staff = row[staff_col]; check_in = row[checkin_col] staff = row[staff_col]; check_in = row[checkin_col]
check_out = row[checkout_col] if checkout_col is not None else None check_out = row[checkout_col] if checkout_col is not None else None
shift_code = row[shift_col]; week_day = row[week_col] if week_col is not None else None shift_code = row[shift_col]; week_day = row[week_col] if week_col is not None else None
if not staff or not shift_code: if not staff or not shift_code:
continue continue
shift_times = get_shift_schedule_full(shift_code, week_day or '') shift_times = get_shift_schedule_full(shift_code, week_day or '')
shift_start = shift_times[0] if shift_times else None shift_start = shift_times[0] if shift_times else None
shift_end = shift_times[1] if shift_times else None shift_end = shift_times[1] if shift_times else None
status = calculate_attendance_status(check_in, check_out, shift_start, shift_end) status = calculate_attendance_status(check_in, check_out, shift_start, shift_end)
staff_key = str(staff) staff_key = str(staff)
if staff_key not in employee_stats: if staff_key not in employee_stats:
employee_stats[staff_key] = {"name": staff, "shift": shift_code, "late_count": 0, "late_minutes": 0, "early_count": 0, "early_minutes": 0, "ot_count": 0, "ot_minutes": 0, "missing_count": 0} employee_stats[staff_key] = {"name": staff, "shift": shift_code, "late_count": 0, "late_minutes": 0, "early_count": 0, "early_minutes": 0, "ot_count": 0, "ot_minutes": 0, "missing_count": 0}
if status["is_missing"]: if status["is_missing"]:
employee_stats[staff_key]["missing_count"] += 1 employee_stats[staff_key]["missing_count"] += 1
totals["missing"] += 1 totals["missing"] += 1
@@ -809,7 +823,7 @@ async def get_lateness_stats(current_user: User = Depends(get_current_user)):
employee_stats[staff_key]["ot_count"] += 1 employee_stats[staff_key]["ot_count"] += 1
employee_stats[staff_key]["ot_minutes"] += status["ot_minutes"] employee_stats[staff_key]["ot_minutes"] += status["ot_minutes"]
totals["ot"] += status["ot_minutes"] totals["ot"] += status["ot_minutes"]
stats = sorted(employee_stats.values(), key=lambda x: x["late_minutes"], reverse=True) stats = sorted(employee_stats.values(), key=lambda x: x["late_minutes"], reverse=True)
return { return {
"stats": stats, "stats": stats,
@@ -831,9 +845,9 @@ async def get_lateness_records(
headers, rows = read_excel_file("attendance") headers, rows = read_excel_file("attendance")
if headers is None or not rows: if headers is None or not rows:
return [] return []
staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None
for i, h in enumerate(headers): for i, h in enumerate(headers):
h_lower = str(h).lower() if h else '' h_lower = str(h).lower() if h else ''
if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower: if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower:
@@ -848,7 +862,7 @@ async def get_lateness_records(
checkout_col = i checkout_col = i
elif '班次' in h_lower: elif '班次' in h_lower:
shift_col = i shift_col = i
records = [] records = []
for row in rows: for row in rows:
staff_val = row[staff_col] if staff_col is not None else None staff_val = row[staff_col] if staff_col is not None else None
@@ -857,14 +871,14 @@ async def get_lateness_records(
shift_code = row[shift_col] if shift_col is not None else None shift_code = row[shift_col] if shift_col is not None else None
week_day = row[week_col] if week_col is not None else None week_day = row[week_col] if week_col is not None else None
date_val = row[date_col] if date_col is not None else None date_val = row[date_col] if date_col is not None else None
if staff_val and shift_code: if staff_val and shift_code:
shift_times = get_shift_schedule_full(shift_code, week_day or '') shift_times = get_shift_schedule_full(shift_code, week_day or '')
shift_start = shift_times[0] if shift_times else None shift_start = shift_times[0] if shift_times else None
shift_end = shift_times[1] if shift_times else None shift_end = shift_times[1] if shift_times else None
status = calculate_attendance_status(check_in, check_out, shift_start, shift_end) status = calculate_attendance_status(check_in, check_out, shift_start, shift_end)
record = { record = {
"staff": staff_val, "date": str(date_val)[:10] if date_val else None, "staff": staff_val, "date": str(date_val)[:10] if date_val else None,
"weekday": week_day, "shift": shift_code, "weekday": week_day, "shift": shift_code,
@@ -873,10 +887,10 @@ async def get_lateness_records(
"actual_in": status["actual_in"], "actual_out": status["actual_out"], "actual_in": status["actual_in"], "actual_out": status["actual_out"],
"late_minutes": status["late_minutes"], "early_minutes": status["early_minutes"], "ot_minutes": status["ot_minutes"] "late_minutes": status["late_minutes"], "early_minutes": status["early_minutes"], "ot_minutes": status["ot_minutes"]
} }
if staff is None or record["staff"] == staff: if staff is None or record["staff"] == staff:
records.append(record) records.append(record)
return records return records
@app.delete("/api/roster") @app.delete("/api/roster")
@@ -893,7 +907,7 @@ async def get_roster_info(current_user: User = Depends(get_current_user)):
path = get_excel_path("roster") path = get_excel_path("roster")
if not os.path.exists(path): if not os.path.exists(path):
return {"uploaded": False, "rows": 0, "columns": 0, "headers": []} return {"uploaded": False, "rows": 0, "columns": 0, "headers": []}
headers, rows = read_excel_file("roster") headers, rows = read_excel_file("roster")
return { return {
"uploaded": True, "uploaded": True,
@@ -910,15 +924,15 @@ async def upload_roster(
"""Upload Roster Excel file""" """Upload Roster Excel file"""
if not file.filename: if not file.filename:
raise HTTPException(status_code=400, detail="No filename provided") raise HTTPException(status_code=400, detail="No filename provided")
path = get_excel_path("roster") path = get_excel_path("roster")
contents = await file.read() contents = await file.read()
with open(path, "wb") as f: with open(path, "wb") as f:
f.write(contents) f.write(contents)
headers, rows = read_excel_file("roster") headers, rows = read_excel_file("roster")
return { return {
"message": "Roster file uploaded successfully", "message": "Roster file uploaded successfully",
"filename": file.filename, "filename": file.filename,
@@ -941,7 +955,7 @@ async def get_shifts(current_user: User = Depends(get_current_user)):
headers, rows = read_excel_file("roster") headers, rows = read_excel_file("roster")
if headers is None: if headers is None:
return [] return []
# Find columns # Find columns
shift_col = None shift_col = None
desc_col = None desc_col = None
@@ -950,7 +964,7 @@ async def get_shifts(current_user: User = Depends(get_current_user)):
shift_col = i shift_col = i
elif h and str(h).lower() in ['描述', 'description', 'desc']: elif h and str(h).lower() in ['描述', 'description', 'desc']:
desc_col = i desc_col = i
shifts = [] shifts = []
for row in rows: for row in rows:
shift_code = row[shift_col] if shift_col is not None else None shift_code = row[shift_col] if shift_col is not None else None
@@ -959,7 +973,7 @@ async def get_shifts(current_user: User = Depends(get_current_user)):
"code": shift_code, "code": shift_code,
"description": row[desc_col] if desc_col is not None else "" "description": row[desc_col] if desc_col is not None else ""
}) })
return shifts return shifts
@app.get("/api/roster/assignments") @app.get("/api/roster/assignments")
@@ -969,7 +983,7 @@ async def get_assignments(current_user: User = Depends(get_current_user)):
headers, rows = read_excel_file("attendance") headers, rows = read_excel_file("attendance")
if headers is None: if headers is None:
return {} return {}
# Find 班次 column # Find 班次 column
shift_col = None shift_col = None
emp_col = None emp_col = None
@@ -978,7 +992,7 @@ async def get_assignments(current_user: User = Depends(get_current_user)):
shift_col = i shift_col = i
elif h and any(kw in str(h).lower() for kw in ['員工', 'employee', 'name', '員工名稱']): elif h and any(kw in str(h).lower() for kw in ['員工', 'employee', 'name', '員工名稱']):
emp_col = i emp_col = i
assignments = {} assignments = {}
if shift_col is not None: if shift_col is not None:
for row in rows: for row in rows:
@@ -986,7 +1000,7 @@ async def get_assignments(current_user: User = Depends(get_current_user)):
shift = row[shift_col] if shift_col is not None and shift_col < len(row) else None shift = row[shift_col] if shift_col is not None and shift_col < len(row) else None
if emp and shift: if emp and shift:
assignments[str(emp)] = str(shift) assignments[str(emp)] = str(shift)
return assignments return assignments
@app.post("/api/roster/assign") @app.post("/api/roster/assign")
@@ -997,27 +1011,27 @@ async def assign_shift(
): ):
"""Assign employee to shift""" """Assign employee to shift"""
assignments_path = os.path.join(DATA_DIR, "shift_assignments.json") assignments_path = os.path.join(DATA_DIR, "shift_assignments.json")
# Load existing # Load existing
if os.path.exists(assignments_path): if os.path.exists(assignments_path):
with open(assignments_path, "r") as f: with open(assignments_path, "r") as f:
assignments = json.load(f) assignments = json.load(f)
else: else:
assignments = {} assignments = {}
assignments[employee] = shift assignments[employee] = shift
return RedirectResponse(url="/") return RedirectResponse(url="/")
@app.get("/api/{section}/info") @app.get("/api/{section}/info")
async def get_section_info(section: str): async def get_section_info(section: str):
"""Get info about stored Excel file""" """Get info about stored Excel file"""
if section not in ["attendance", "accident"]: if section not in ["attendance", "accident"]:
raise HTTPException(status_code=400, detail="Invalid section") raise HTTPException(status_code=400, detail="Invalid section")
headers, rows = read_excel_file(section) headers, rows = read_excel_file(section)
if headers is None: if headers is None:
return {"uploaded": False, "rows": 0, "columns": 0, "headers": []} return {"uploaded": False, "rows": 0, "columns": 0, "headers": []}
return { return {
"uploaded": True, "uploaded": True,
"rows": len(rows), "rows": len(rows),
@@ -1595,14 +1609,14 @@ async def refresh_holidays_from_gov_hk(
current_user: User = Depends(get_current_admin), current_user: User = Depends(get_current_admin),
): ):
"""Refresh holidays from internal static list (sourced from gov.hk). """Refresh holidays from internal static list (sourced from gov.hk).
Admin endpoint. Adds new dates, updates existing ones (only if source=gov_hk). Admin endpoint. Adds new dates, updates existing ones (only if source=gov_hk).
Manually edited holidays are NOT overwritten. Manually edited holidays are NOT overwritten.
""" """
from parsers import get_all_holidays from parsers import get_all_holidays
target_years = [int(y) for y in years.split(",")] if years else [2025, 2026, 2027] target_years = [int(y) for y in years.split(",")] if years else [2025, 2026, 2027]
all_holidays = [h for h in get_all_holidays() if h[0].year in target_years] all_holidays = [h for h in get_all_holidays() if h[0].year in target_years]
added = 0 added = 0
updated = 0 updated = 0
skipped = 0 skipped = 0
@@ -1626,7 +1640,7 @@ async def refresh_holidays_from_gov_hk(
region="HK", is_mandatory=is_mandatory, source="gov_hk", region="HK", is_mandatory=is_mandatory, source="gov_hk",
)) ))
added += 1 added += 1
db.commit() db.commit()
return { return {
"added": added, "updated": updated, "skipped": skipped, "added": added, "updated": updated, "skipped": skipped,
@@ -1677,7 +1691,7 @@ async def create_leave(
raise HTTPException(status_code=400, detail="leave_type must be SL / CL / AL") raise HTTPException(status_code=400, detail="leave_type must be SL / CL / AL")
if body.hours < 0 or body.hours > 24: if body.hours < 0 or body.hours > 24:
raise HTTPException(status_code=400, detail="hours must be 0-24") raise HTTPException(status_code=400, detail="hours must be 0-24")
lv = LeaveRecord( lv = LeaveRecord(
employee_name=body.employee_name, employee_name=body.employee_name,
employee_id=body.employee_id, employee_id=body.employee_id,
@@ -1749,14 +1763,14 @@ async def upload_leaves(
current_user: User = Depends(get_current_user), current_user: User = Depends(get_current_user),
): ):
"""Bulk upload leave records from CSV or Excel. """Bulk upload leave records from CSV or Excel.
CSV columns: employee_name,date,leave_type,hours,reason,approved_by CSV columns: employee_name,date,leave_type,hours,reason,approved_by
Excel: same columns (row 1 = header) Excel: same columns (row 1 = header)
""" """
from io import BytesIO from io import BytesIO
content = await file.read() content = await file.read()
fname = file.filename.lower() fname = file.filename.lower()
rows = [] rows = []
if fname.endswith(".csv"): if fname.endswith(".csv"):
import csv, io as _io import csv, io as _io
@@ -1792,7 +1806,7 @@ async def upload_leaves(
}) })
else: else:
raise HTTPException(status_code=400, detail="File must be CSV or Excel") raise HTTPException(status_code=400, detail="File must be CSV or Excel")
added = 0 added = 0
updated = 0 updated = 0
skipped = 0 skipped = 0
@@ -1807,19 +1821,19 @@ async def upload_leaves(
errors.append(f"Row {i+2}: invalid leave_type") errors.append(f"Row {i+2}: invalid leave_type")
skipped += 1 skipped += 1
continue continue
from datetime import datetime as _dt from datetime import datetime as _dt
if isinstance(r["date"], str): if isinstance(r["date"], str):
d = _dt.strptime(r["date"], "%Y-%m-%d").date() d = _dt.strptime(r["date"], "%Y-%m-%d").date()
else: else:
d = r["date"] d = r["date"]
existing = db.query(LeaveRecord).filter( existing = db.query(LeaveRecord).filter(
LeaveRecord.employee_name == r["employee_name"], LeaveRecord.employee_name == r["employee_name"],
LeaveRecord.date == d, LeaveRecord.date == d,
LeaveRecord.leave_type == r["leave_type"], LeaveRecord.leave_type == r["leave_type"],
).first() ).first()
if existing: if existing:
if existing.last_edited_at is not None: if existing.last_edited_at is not None:
# Was manually edited - protect from overwrite # Was manually edited - protect from overwrite
@@ -1833,7 +1847,7 @@ async def upload_leaves(
existing.source = "csv_upload" existing.source = "csv_upload"
updated += 1 updated += 1
continue continue
db.add(LeaveRecord( db.add(LeaveRecord(
employee_name=r["employee_name"], employee_name=r["employee_name"],
date=d, date=d,
@@ -1848,7 +1862,7 @@ async def upload_leaves(
except Exception as e: except Exception as e:
errors.append(f"Row {i+2}: {str(e)}") errors.append(f"Row {i+2}: {str(e)}")
skipped += 1 skipped += 1
db.commit() db.commit()
return { return {
"added": added, "updated": updated, "skipped": skipped, "added": added, "updated": updated, "skipped": skipped,
@@ -2013,16 +2027,16 @@ async def list_records(
"""List records from stored Excel file""" """List records from stored Excel file"""
if section not in ["attendance", "accident"]: if section not in ["attendance", "accident"]:
raise HTTPException(status_code=400, detail="Invalid section") raise HTTPException(status_code=400, detail="Invalid section")
headers, rows = read_excel_file(section) headers, rows = read_excel_file(section)
if headers is None: if headers is None:
return {"total": 0, "page": page, "page_size": page_size, "data": []} return {"total": 0, "page": page, "page_size": page_size, "data": []}
# Pagination # Pagination
start = (page - 1) * page_size start = (page - 1) * page_size
end = start + page_size end = start + page_size
page_rows = rows[start:end] page_rows = rows[start:end]
# Build response with row number as id # Build response with row number as id
data = [] data = []
for idx, row in enumerate(page_rows, start=start + 1): for idx, row in enumerate(page_rows, start=start + 1):
@@ -2031,7 +2045,7 @@ async def list_records(
if header: if header:
record[str(header)] = row[i] if i < len(row) else None record[str(header)] = row[i] if i < len(row) else None
data.append(record) data.append(record)
return { return {
"total": len(rows), "total": len(rows),
"page": page, "page": page,
@@ -2045,33 +2059,33 @@ async def dashboard(section: str, current_user: User = Depends(get_current_user)
"""Get dashboard stats from stored Excel file""" """Get dashboard stats from stored Excel file"""
if section not in ["attendance", "accident"]: if section not in ["attendance", "accident"]:
raise HTTPException(status_code=400, detail="Invalid section") raise HTTPException(status_code=400, detail="Invalid section")
headers, rows = read_excel_file(section) headers, rows = read_excel_file(section)
if headers is None or not rows: if headers is None or not rows:
return {"total": 0, "this_month": 0, "by_status": {}, "by_department": {}, "trend": []} return {"total": 0, "this_month": 0, "by_status": {}, "by_department": {}, "trend": []}
today = date.today() today = date.today()
month_start = date(today.year, today.month, 1) month_start = date(today.year, today.month, 1)
# Find date column (look for common date column names) # Find date column (look for common date column names)
date_col_idx = None date_col_idx = None
for i, h in enumerate(headers): for i, h in enumerate(headers):
if h and any(kw in str(h).lower() for kw in ['date', '日期', '時間']): if h and any(kw in str(h).lower() for kw in ['date', '日期', '時間']):
date_col_idx = i date_col_idx = i
break break
# Find department column # Find department column
dept_col_idx = None dept_col_idx = None
for i, h in enumerate(headers): for i, h in enumerate(headers):
if h and any(kw in str(h).lower() for kw in ['dept', '部門', '部門名稱']): if h and any(kw in str(h).lower() for kw in ['dept', '部門', '部門名稱']):
dept_col_idx = i dept_col_idx = i
break break
total = len(rows) total = len(rows)
this_month = 0 this_month = 0
by_department = {} by_department = {}
trend = [] trend = []
for row in rows: for row in rows:
# Count this month # Count this month
if date_col_idx is not None and date_col_idx < len(row): if date_col_idx is not None and date_col_idx < len(row):
@@ -2091,12 +2105,12 @@ async def dashboard(section: str, current_user: User = Depends(get_current_user)
pass pass
if row_date and row_date >= month_start: if row_date and row_date >= month_start:
this_month += 1 this_month += 1
# Count by department # Count by department
if dept_col_idx is not None and dept_col_idx < len(row): if dept_col_idx is not None and dept_col_idx < len(row):
dept = str(row[dept_col_idx] or 'Unknown') dept = str(row[dept_col_idx] or 'Unknown')
by_department[dept] = by_department.get(dept, 0) + 1 by_department[dept] = by_department.get(dept, 0) + 1
return { return {
"total": total, "total": total,
"today": 0, "today": 0,
@@ -2121,7 +2135,7 @@ async def _import_accident_to_db(db: Session, headers, rows, user_id: int) -> di
Col 8 涉及的患者或相關人員 -> patient_or_staff Col 8 涉及的患者或相關人員 -> patient_or_staff
Col 9 事件描述 -> long_description Col 9 事件描述 -> long_description
Col 10 如有事件相關照片可提供-> incident_photos Col 10 如有事件相關照片可提供-> incident_photos
Col 11 處理情況／已採取的行動-> action_taken Col 11 處理情況/已採取的行動-> action_taken
Col 12 需要管理層介入的事項 -> needs_escalation Col 12 需要管理層介入的事項 -> needs_escalation
Col 13 第 12 欄 -> (skipped) Col 13 第 12 欄 -> (skipped)
Col 14 分數 -> incident_score Col 14 分數 -> incident_score
@@ -2839,7 +2853,7 @@ async def accident_export_excel(
# User types case_no in C5 of 個案查詢 sheet. # User types case_no in C5 of 個案查詢 sheet.
# xlsxwriter requires single quotes around sheet names containing non-ASCII. # xlsxwriter requires single quotes around sheet names containing non-ASCII.
# Use proper xlsxwriter hyperlink element (write_url with internal:Sheet!Cell). # Use proper xlsxwriter hyperlink element (write_url with internal:Sheet!Cell).
# HYPERLINK formula would cache value 0 — write_url sets display text directly. # HYPERLINK formula would cache value 0 - write_url sets display text directly.
overview.write_url(row_n, 6, overview.write_url(row_n, 6,
f"internal:個案查詢!A1", f"internal:個案查詢!A1",
link_fmt, link_fmt,
@@ -2847,7 +2861,7 @@ async def accident_export_excel(
tip=f"跳到「個案查詢」輸入個案 ID {r.id}") tip=f"跳到「個案查詢」輸入個案 ID {r.id}")
overview.set_row(row_n, 18) overview.set_row(row_n, 18)
# ============ Charts block — 統計圖表 (Sheet 1 總覽 右側 columns I-N) ============ # ============ Charts block - 統計圖表 (Sheet 1 總覽 右側 columns I-N) ============
# 4 charts: Severity Pie, Severity Column, Department Bar TOP 10, Monthly Trend Line # 4 charts: Severity Pie, Severity Column, Department Bar TOP 10, Monthly Trend Line
# Position: I column (right of KPI + 個案索引 blocks), so no overlap with case index. # Position: I column (right of KPI + 個案索引 blocks), so no overlap with case index.
@@ -2957,7 +2971,7 @@ async def accident_export_excel(
# ============ Sheet 3 (built FIRST so detail-VLOOKUP can reference it): 明細 ============ # ============ Sheet 3 (built FIRST so detail-VLOOKUP can reference it): 明細 ============
# Build it now so that 個案查詢 formulas can resolve its range properly. # Build it now so that 個案查詢 formulas can resolve its range properly.
# However, xlsxwriter's defined_name allows forward reference; formula text is # However, xlsxwriter's defined_name allows forward reference; formula text is
# what matters at runtime — even if the sheet is added later, the formula string # what matters at runtime - even if the sheet is added later, the formula string
# '明細!A:O' will be valid when Excel opens the file. # '明細!A:O' will be valid when Excel opens the file.
# We define 明細 here so that the formula range 明細!A2:O{max_row} resolves. # We define 明細 here so that the formula range 明細!A2:O{max_row} resolves.
# But xlsxwriter requires sheets to be added in the order they're defined. # But xlsxwriter requires sheets to be added in the order they're defined.
@@ -3017,12 +3031,12 @@ async def accident_export_excel(
"align": "center", "num_format": "@"}) "align": "center", "num_format": "@"})
# B5 = label only (NOT merged with C5). Otherwise user-typed text goes to # B5 = label only (NOT merged with C5). Otherwise user-typed text goes to
# B5 (top-left of merge) instead of C5 where the formula references. # B5 (top-left of merge) instead of C5 where the formula references.
detail.write("B5", "輸入個案 ID：", input_label_fmt) detail.write("B5", "輸入個案 ID:", input_label_fmt)
# C5 = case_no input cell (NOT merged) # C5 = case_no input cell (NOT merged)
detail.write_string("C5", "", text_input_fmt) detail.write_string("C5", "", text_input_fmt)
# Hint # Hint
detail.merge_range("B6:C6", detail.merge_range("B6:C6",
"提示：個案 ID 喺「總覽」個案索引或「明細」表嘅 id 欄。輸入後下面自動顯示。", "提示:個案 ID 喺「總覽」個案索引或「明細」表嘅 id 欄。輸入後下面自動顯示。",
subtitle_fmt) subtitle_fmt)
# Detail rows: each pulls from 明細!<col><row> via VLOOKUP keyed on id (col A) # Detail rows: each pulls from 明細!<col><row> via VLOOKUP keyed on id (col A)
@@ -3061,7 +3075,7 @@ async def accident_export_excel(
col_sev = col_index_lookup.get("severity", 6) col_sev = col_index_lookup.get("severity", 6)
col_emp = col_index_lookup.get("employee_name", 3) col_emp = col_index_lookup.get("employee_name", 3)
col_desc = col_index_lookup.get("description", 5) col_desc = col_index_lookup.get("description", 5)
# Use INDIRECT or simply use cell reference for ID — but ID varies per row. # Use INDIRECT or simply use cell reference for ID - but ID varies per row.
# Easier: each row's ID is in B<row_n>; build formula with that. # Easier: each row's ID is in B<row_n>; build formula with that.
# VLOOKUP needs lookup_value, table, col_index, FALSE. # VLOOKUP needs lookup_value, table, col_index, FALSE.
# table 明細!$A:$O. # table 明細!$A:$O.
@@ -3108,13 +3122,13 @@ async def accident_export_excel(
compare_start_row = list_header_row + n_list_rows + 3 compare_start_row = list_header_row + n_list_rows + 3
detail.write(f"B{compare_start_row}", "兩個個案 並列比較 Side-by-Side", section_fmt) detail.write(f"B{compare_start_row}", "兩個個案 並列比較 Side-by-Side", section_fmt)
detail.write(f"B{compare_start_row + 1}", detail.write(f"B{compare_start_row + 1}",
"輸入兩個個案 ID (左個案 A, 右個案 B)，對比每個欄位。空白表示無資料。", "輸入兩個個案 ID (左個案 A, 右個案 B),對比每個欄位。空白表示無資料。",
subtitle_fmt) subtitle_fmt)
cmp_label_row = compare_start_row + 2 cmp_label_row = compare_start_row + 2
detail.write(cmp_label_row, 1, "個案 A ID：", input_label_fmt) detail.write(cmp_label_row, 1, "個案 A ID:", input_label_fmt)
detail.write_string(cmp_label_row, 2, "", text_input_fmt) detail.write_string(cmp_label_row, 2, "", text_input_fmt)
detail.write(cmp_label_row, 4, "個案 B ID：", input_label_fmt) detail.write(cmp_label_row, 4, "個案 B ID:", input_label_fmt)
detail.write_string(cmp_label_row, 5, "", text_input_fmt) detail.write_string(cmp_label_row, 5, "", text_input_fmt)
detail.set_column("F:F", 50) detail.set_column("F:F", 50)
@@ -3146,7 +3160,7 @@ async def accident_export_excel(
# Hint at top # Hint at top
detail.merge_range(f"B{cmp_data_start + len(detail_fields) + 2}:C{cmp_data_start + len(detail_fields) + 2}", detail.merge_range(f"B{cmp_data_start + len(detail_fields) + 2}:C{cmp_data_start + len(detail_fields) + 2}",
"💡 全部公式 (VLOOKUP)，無需啟用巨集。Excel/Numbers 都正常運作。", "💡 全部公式 (VLOOKUP),無需啟用巨集。Excel/Numbers 都正常運作。",
subtitle_fmt) subtitle_fmt)
# ============ Sheet 3: 明細 ============ # ============ Sheet 3: 明細 ============
@@ -3266,21 +3280,21 @@ async def get_record(
"""Get a specific record by row number""" """Get a specific record by row number"""
if section not in ["attendance", "accident"]: if section not in ["attendance", "accident"]:
raise HTTPException(status_code=400, detail="Invalid section") raise HTTPException(status_code=400, detail="Invalid section")
headers, rows = read_excel_file(section) headers, rows = read_excel_file(section)
if headers is None: if headers is None:
raise HTTPException(status_code=404, detail="No data file found") raise HTTPException(status_code=404, detail="No data file found")
idx = row_id - 1 # Convert to 0-based index idx = row_id - 1 # Convert to 0-based index
if idx < 0 or idx >= len(rows): if idx < 0 or idx >= len(rows):
raise HTTPException(status_code=404, detail="Record not found") raise HTTPException(status_code=404, detail="Record not found")
row = rows[idx] row = rows[idx]
record = {"_row": row_id} record = {"_row": row_id}
for i, header in enumerate(headers): for i, header in enumerate(headers):
if header: if header:
record[str(header)] = row[i] if i < len(row) else None record[str(header)] = row[i] if i < len(row) else None
return record return record
# ============ Export ============ # ============ Export ============
@@ -3411,7 +3425,7 @@ async def export_excel(
earliest = min((r.date for r in rows if r.date), default=None) earliest = min((r.date for r in rows if r.date), default=None)
overview.merge_range("F7:G7", earliest.isoformat() if earliest else "-", kpi_value_fmt) overview.merge_range("F7:G7", earliest.isoformat() if earliest else "-", kpi_value_fmt)
# Status distribution table — simplified 6-group (B10:D) # Status distribution table - simplified 6-group (B10:D)
overview.write("B10", "出勤狀態分佈", section_fmt) overview.write("B10", "出勤狀態分佈", section_fmt)
overview.write("B11", "Status", header_fmt_ov) overview.write("B11", "Status", header_fmt_ov)
overview.write("C11", "數量", header_fmt_ov) overview.write("C11", "數量", header_fmt_ov)