- {/* Result message */}
{res && (
str:
+ """Get path to the stored Excel file for a section"""
+ return os.path.join(DATA_DIR, f"{section}.xlsx")
+
+def read_excel_file(section: str):
+ """Read Excel file and return headers and rows"""
+ import openpyxl
+ path = get_excel_path(section)
+ if not os.path.exists(path):
+ return None, []
+
+ wb = openpyxl.load_workbook(path, data_only=True)
+ ws = wb.active
+
+ headers = [cell.value for cell in ws[1]]
+ rows = list(ws.iter_rows(min_row=2, values_only=True))
+
+ return headers, rows
+
+# ============ Upload Excel (Replace entire file) ============
+@app.delete("/api/upload/{section}")
+async def delete_uploaded_file(
+ section: str,
+ current_user: User = Depends(get_current_user)
+):
+ """Delete the stored Excel file for a section"""
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ path = get_excel_path(section)
+ if os.path.exists(path):
+ os.remove(path)
+
+ return {"message": f"{section} file deleted"}
+
+@app.post("/api/upload/{section}")
+async def upload_excel(
+ section: str,
+ file: UploadFile = File(...),
+ current_user: User = Depends(get_current_user)
+):
+ """Upload Excel file - replaces any existing file for this section"""
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ if not file.filename:
+ raise HTTPException(status_code=400, detail="No filename provided")
+
+ # Save the file
+ path = get_excel_path(section)
+ contents = await file.read()
+
+ with open(path, "wb") as f:
+ f.write(contents)
+
+ # Return info about the uploaded file
+ headers, rows = read_excel_file(section)
+
+ return {
+ "message": f"{section} file uploaded successfully",
+ "filename": file.filename,
+ "rows": len(rows),
+ "columns": len(headers),
+ "headers": headers
+ }
+
+
+# ============ Attendance Calculation ============
+def get_shift_schedule_full(shift_code: str, weekday: str) -> Optional[tuple]:
+ """Get the expected start and end time for a shift on a given weekday.
+ Returns (start_time, end_time) tuple.
+ """
+ roster_path = get_excel_path("roster")
+ if not os.path.exists(roster_path):
+ return None
+
+ import openpyxl
+ wb = openpyxl.load_workbook(roster_path, data_only=True)
+ ws = wb.active
+
+ headers = [cell.value for cell in ws[1]]
+
+ # Find column indices
+ shift_col = None
+ for i, h in enumerate(headers):
+ if h == '班次':
+ shift_col = i
+ break
+
+ if shift_col is None:
+ return None
+
+ # Weekday mapping
+ weekday_map = {
+ 'Monday': '星期一', 'Tuesday': '星期二', 'Wednesday': '星期三',
+ 'Thursday': '星期四', 'Friday': '星期五', 'Saturday': '星期六',
+ 'Sunday': '星期日', '星期一': '星期一', '星期二': '星期二',
+ '星期三': '星期三', '星期四': '星期四', '星期五': '星期五',
+ '星期六': '星期六', '星期日': '星期日'
+ }
+
+ day_col_map = {
+ '星期一': None, '星期二': None, '星期三': None,
+ '星期四': None, '星期五': None, '星期六': None, '星期日': None
+ }
+
+ for i, h in enumerate(headers):
+ if h in day_col_map:
+ day_col_map[h] = i
+
+ target_day = weekday_map.get(weekday, weekday)
+
+ # Find the shift row
+ for row in ws.iter_rows(min_row=2, values_only=True):
+ if row[shift_col] == shift_code:
+ day_col = day_col_map.get(target_day)
+ if day_col is not None:
+ schedule = row[day_col]
+ if schedule and schedule != '-' and schedule != '休息':
+ # Parse "HH:MM-HH:MM" format
+ try:
+ parts = str(schedule).split('-')
+ if len(parts) == 2:
+ start_str = parts[0].strip()
+ end_str = parts[1].strip()
+ start_time = datetime.strptime(start_str, '%H:%M').time()
+ end_time = datetime.strptime(end_str, '%H:%M').time()
+ return (start_time, end_time)
+ except:
+ pass
+
+ return None
+
+def parse_datetime(time_val) -> Optional[datetime]:
+ """Parse datetime from various formats"""
+ if not time_val:
+ return None
+ if isinstance(time_val, datetime):
+ return time_val
+ if isinstance(time_val, str):
+ for fmt in ['%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M', '%H:%M:%S', '%H:%M']:
+ try:
+ return datetime.strptime(time_val, fmt)
+ except:
+ continue
+ return None
+
+def calculate_attendance_status(check_in, check_out, shift_start, shift_end):
+ """Calculate attendance status: late_minutes, early_minutes, ot_minutes, is_missing
+
+ Returns: {
+ "is_missing": bool,
+ "late_minutes": int,
+ "early_minutes": int,
+ "ot_minutes": int,
+ "status_code": str, # e.g. "late_early", "early_ot", "normal"
+ "status_text": str,
+ "expected_in": str,
+ "expected_out": str,
+ "actual_in": str,
+ "actual_out": str
+ }
+ """
+ result = {
+ "is_missing": False,
+ "late_minutes": 0,
+ "early_minutes": 0,
+ "ot_minutes": 0,
+ "status_code": "normal",
+ "status_text": "正常",
+ "expected_in": shift_start.strftime("%H:%M") if shift_start else "-",
+ "expected_out": shift_end.strftime("%H:%M") if shift_end else "-",
+ "actual_in": "-",
+ "actual_out": "-"
+ }
+
+ # Check for missing punch
+ if not check_in or not check_out:
+ result["is_missing"] = True
+ result["status_code"] = "missing"
+ result["status_text"] = "缺勤"
+ if check_in:
+ result["actual_in"] = check_in.strftime("%H:%M") if isinstance(check_in, datetime) else str(check_in)
+ if check_out:
+ result["actual_out"] = check_out.strftime("%H:%M") if isinstance(check_out, datetime) else str(check_out)
+ return result
+
+ # Parse check-in/out times
+ check_in_dt = parse_datetime(check_in)
+ check_out_dt = parse_datetime(check_out)
+
+ if not check_in_dt or not check_out_dt:
+ result["is_missing"] = True
+ result["status_code"] = "missing"
+ result["status_text"] = "缺勤"
+ return result
+
+ result["actual_in"] = check_in_dt.strftime("%H:%M")
+ result["actual_out"] = check_out_dt.strftime("%H:%M")
+
+ # Skip if invalid times (Excel default "0" = 18:00)
+ if check_in_dt.hour == 18 and check_in_dt.minute == 0 and check_in_dt.second == 0:
+ result["is_missing"] = True
+ result["status_code"] = "missing"
+ result["status_text"] = "缺勤"
+ return result
+
+ if not shift_start or not shift_end:
+ # No roster data, can't calculate
+ return result
+
+ # Calculate late minutes
+ check_in_time = check_in_dt.time()
+ if check_in_time > shift_start:
+ diff = (datetime.combine(check_in_dt.date(), check_in_time) -
+ datetime.combine(check_in_dt.date(), shift_start)).total_seconds() / 60
+ result["late_minutes"] = int(diff)
+
+ # Calculate early minutes (leaving before scheduled end)
+ check_out_time = check_out_dt.time()
+ if shift_end > shift_start:
+ # Handle overnight shifts
+ if shift_end < shift_start:
+ # Assume next day
+ expected_end_dt = datetime.combine(check_out_dt.date() + timedelta(days=1), shift_end)
+ else:
+ expected_end_dt = datetime.combine(check_out_dt.date(), shift_end)
+
+ if check_out_dt < expected_end_dt:
+ diff = (expected_end_dt - check_out_dt).total_seconds() / 60
+ result["early_minutes"] = int(diff)
+
+ # Calculate OT minutes (leaving after scheduled end)
+ if check_out_dt > expected_end_dt:
+ diff = (check_out_dt - expected_end_dt).total_seconds() / 60
+ result["ot_minutes"] = int(diff)
+
+ # Determine status code and text
+ status_parts = []
+ if result["late_minutes"] > 0:
+ status_parts.append(f"遲到{result['late_minutes']}分")
+ if result["early_minutes"] > 0:
+ status_parts.append(f"早退{result['early_minutes']}分")
+ if result["ot_minutes"] > 0:
+ status_parts.append(f"OT{result['ot_minutes']}分")
+
+ if result["late_minutes"] > 0 and result["early_minutes"] > 0:
+ result["status_code"] = "late_early"
+ elif result["late_minutes"] > 0 and result["ot_minutes"] > 0:
+ result["status_code"] = "late_ot"
+ elif result["early_minutes"] > 0 and result["ot_minutes"] > 0:
+ result["status_code"] = "early_ot"
+ elif result["late_minutes"] > 0:
+ result["status_code"] = "late"
+ elif result["early_minutes"] > 0:
+ result["status_code"] = "early"
+ elif result["ot_minutes"] > 0:
+ result["status_code"] = "ot"
+ else:
+ result["status_code"] = "normal"
+
+ result["status_text"] = " / ".join(status_parts) if status_parts else "正常"
+
+
+@app.get("/api/attendance/lateness")
+async def get_lateness_stats(current_user: User = Depends(get_current_user)):
+ """Calculate full attendance statistics: late, early, OT"""
+ headers, rows = read_excel_file("attendance")
+ if headers is None or not rows:
+ 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
+
+ for i, h in enumerate(headers):
+ h_lower = str(h).lower() if h else ''
+ if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower:
+ staff_col = i
+ elif 'date' in h_lower or '日期' in h_lower:
+ date_col = i
+ elif 'week' in h_lower or '星期' in h_lower:
+ week_col = i
+ elif 'check' in h_lower and 'in' in h_lower or '上班' in h_lower:
+ checkin_col = i
+ elif 'check' in h_lower and 'out' in h_lower or '下班' in h_lower:
+ checkout_col = i
+ elif '班次' in h_lower:
+ shift_col = i
+
+ if staff_col is None or checkin_col is None or shift_col is None:
+ return {"error": "Missing required columns", "headers": headers}
+
+ employee_stats = {}
+ totals = {"late": 0, "early": 0, "ot": 0, "missing": 0}
+
+ for row in rows:
+ staff = row[staff_col]; check_in = row[checkin_col]
+ 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
+
+ if not staff or not shift_code:
+ continue
+
+ shift_times = get_shift_schedule_full(shift_code, week_day or '')
+ shift_start = shift_times[0] 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)
+
+ staff_key = str(staff)
+ 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}
+
+ if status["is_missing"]:
+ employee_stats[staff_key]["missing_count"] += 1
+ totals["missing"] += 1
+ else:
+ if status["late_minutes"] > 0:
+ employee_stats[staff_key]["late_count"] += 1
+ employee_stats[staff_key]["late_minutes"] += status["late_minutes"]
+ totals["late"] += status["late_minutes"]
+ if status["early_minutes"] > 0:
+ employee_stats[staff_key]["early_count"] += 1
+ employee_stats[staff_key]["early_minutes"] += status["early_minutes"]
+ totals["early"] += status["early_minutes"]
+ if status["ot_minutes"] > 0:
+ employee_stats[staff_key]["ot_count"] += 1
+ employee_stats[staff_key]["ot_minutes"] += status["ot_minutes"]
+ totals["ot"] += status["ot_minutes"]
+
+ stats = sorted(employee_stats.values(), key=lambda x: x["late_minutes"], reverse=True)
+ return {"stats": stats, **totals}
+
+@app.get("/api/attendance/lateness/records")
+async def get_lateness_records(
+ staff: Optional[str] = None,
+ current_user: User = Depends(get_current_user)
+):
+ """Get detailed attendance status records"""
+ headers, rows = read_excel_file("attendance")
+ if headers is None or not rows:
+ return []
+
+ staff_col = None; date_col = None; week_col = None; checkin_col = None; checkout_col = None; shift_col = None
+
+ for i, h in enumerate(headers):
+ h_lower = str(h).lower() if h else ''
+ if 'staff' in h_lower or '員工' in h_lower or 'name' in h_lower:
+ staff_col = i
+ elif 'date' in h_lower or '日期' in h_lower:
+ date_col = i
+ elif 'week' in h_lower or '星期' in h_lower:
+ week_col = i
+ elif 'check' in h_lower and 'in' in h_lower or '上班' in h_lower:
+ checkin_col = i
+ elif 'check' in h_lower and 'out' in h_lower or '下班' in h_lower:
+ checkout_col = i
+ elif '班次' in h_lower:
+ shift_col = i
+
+ records = []
+ for row in rows:
+ staff_val = row[staff_col] if staff_col is not None else None
+ check_in = row[checkin_col] if checkin_col is not None else None
+ check_out = row[checkout_col] if checkout_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
+ date_val = row[date_col] if date_col is not None else None
+
+ if staff_val and shift_code:
+ shift_times = get_shift_schedule_full(shift_code, week_day or '')
+ shift_start = shift_times[0] 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)
+
+ record = {
+ "staff": staff_val, "date": str(date_val)[:10] if date_val else None,
+ "weekday": week_day, "shift": shift_code,
+ "status_code": status["status_code"], "status_text": status["status_text"],
+ "expected_in": status["expected_in"], "expected_out": status["expected_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"]
+ }
+
+ if staff is None or record["staff"] == staff:
+ records.append(record)
+
+ return records
+
+@app.delete("/api/roster")
+async def delete_roster(current_user: User = Depends(get_current_user)):
+ """Delete the stored roster file"""
+ path = get_excel_path("roster")
+ if os.path.exists(path):
+ os.remove(path)
+ return {"message": "Roster file deleted"}
+
+@app.get("/api/roster/info")
+async def get_roster_info(current_user: User = Depends(get_current_user)):
+ """Get info about stored roster Excel file"""
+ path = get_excel_path("roster")
+ if not os.path.exists(path):
+ return {"uploaded": False, "rows": 0, "columns": 0, "headers": []}
+
+ headers, rows = read_excel_file("roster")
+ return {
+ "uploaded": True,
+ "rows": len(rows) if rows else 0,
+ "columns": len(headers) if headers else 0,
+ "headers": headers if headers else []
+ }
+
+@app.post("/api/roster/upload")
+async def upload_roster(
+ file: UploadFile = File(...),
+ current_user: User = Depends(get_current_user)
+):
+ """Upload Roster Excel file"""
+ if not file.filename:
+ raise HTTPException(status_code=400, detail="No filename provided")
+
+ path = get_excel_path("roster")
+ contents = await file.read()
+
+ with open(path, "wb") as f:
+ f.write(contents)
+
+ headers, rows = read_excel_file("roster")
+
+ return {
+ "message": "Roster file uploaded successfully",
+ "filename": file.filename,
+ "rows": len(rows) if rows else 0,
+ "columns": len(headers) if headers else 0,
+ "headers": headers if headers else []
+ }
+
+@app.get("/api/roster/data")
+async def get_roster_data(current_user: User = Depends(get_current_user)):
+ """Get full roster Excel data"""
+ headers, rows = read_excel_file("roster")
+ if headers is None:
+ return {"headers": [], "rows": []}
+ return {"headers": headers, "rows": rows}
+
+@app.get("/api/roster/shifts")
+async def get_shifts(current_user: User = Depends(get_current_user)):
+ """Get all shifts from roster file"""
+ headers, rows = read_excel_file("roster")
+ if headers is None:
+ return []
+
+ # Find columns
+ shift_col = None
+ desc_col = None
+ for i, h in enumerate(headers):
+ if h and str(h).lower() in ['班次', 'shift', 'shift code']:
+ shift_col = i
+ elif h and str(h).lower() in ['描述', 'description', 'desc']:
+ desc_col = i
+
+ shifts = []
+ for row in rows:
+ shift_code = row[shift_col] if shift_col is not None else None
+ if shift_code:
+ shifts.append({
+ "code": shift_code,
+ "description": row[desc_col] if desc_col is not None else ""
+ })
+
+ return shifts
+
+@app.get("/api/roster/assignments")
+async def get_assignments(current_user: User = Depends(get_current_user)):
+ """Get employee-shift assignments"""
+ # Read from attendance data
+ headers, rows = read_excel_file("attendance")
+ if headers is None:
+ return {}
+
+ # Find 班次 column
+ shift_col = None
+ emp_col = None
+ for i, h in enumerate(headers):
+ if h and '班次' in str(h):
+ shift_col = i
+ elif h and any(kw in str(h).lower() for kw in ['員工', 'employee', 'name', '員工名稱']):
+ emp_col = i
+
+ assignments = {}
+ if shift_col is not None:
+ for row in rows:
+ emp = row[emp_col] if emp_col is not None and emp_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:
+ assignments[str(emp)] = str(shift)
+
+ return assignments
+
+@app.post("/api/roster/assign")
+async def assign_shift(
+ employee: str = Form(...),
+ shift: str = Form(...),
+ current_user: User = Depends(get_current_user)
+):
+ """Assign employee to shift"""
+ assignments_path = os.path.join(DATA_DIR, "shift_assignments.json")
+
+ # Load existing
+ if os.path.exists(assignments_path):
+ with open(assignments_path, "r") as f:
+ assignments = json.load(f)
+ else:
+ assignments = {}
+
+ assignments[employee] = shift
+
+ return RedirectResponse(url="/")
+@app.get("/api/{section}/info")
+async def get_section_info(section: str):
+ """Get info about stored Excel file"""
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ headers, rows = read_excel_file(section)
+ if headers is None:
+ return {"uploaded": False, "rows": 0, "columns": 0, "headers": []}
+
+ return {
+ "uploaded": True,
+ "rows": len(rows),
+ "columns": len(headers),
+ "headers": headers
+ }
+
+# ============ List Records (from Excel) ============
+@app.get("/api/{section}")
+async def list_records(
+ section: str,
+ page: int = Query(1, ge=1),
+ page_size: int = Query(50, ge=1, le=1000),
+ current_user: User = Depends(get_current_user)
+):
+ """List records from stored Excel file"""
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ headers, rows = read_excel_file(section)
+ if headers is None:
+ return {"total": 0, "page": page, "page_size": page_size, "data": []}
+
+ # Pagination
+ start = (page - 1) * page_size
+ end = start + page_size
+ page_rows = rows[start:end]
+
+ # Build response with row number as id
+ data = []
+ for idx, row in enumerate(page_rows, start=start + 1):
+ record = {"_row": idx} # Row number (1-based, excluding header)
+ for i, header in enumerate(headers):
+ if header:
+ record[str(header)] = row[i] if i < len(row) else None
+ data.append(record)
+
+ return {
+ "total": len(rows),
+ "page": page,
+ "page_size": page_size,
+ "data": data
+ }
+
+# ============ Dashboard Summary (combined) ============
+@app.get("/api/dashboard/summary")
+async def dashboard_summary(current_user: User = Depends(get_current_user)):
+ """Combined dashboard data for frontend Dashboard page."""
+ # Attendance data
+ att_headers, att_rows = read_excel_file("attendance")
+ # Accident data
+ acc_headers, acc_rows = read_excel_file("accident")
+
+ return {
+ "attendance": {
+ "total": len(att_rows) if att_rows else 0,
+ "this_month": 0,
+ },
+ "accident": {
+ "total": len(acc_rows) if acc_rows else 0,
+ "this_month": 0,
+ }
+ }
+
+# ============ Attendance Stats ============
+@app.get("/api/attendance/stats")
+async def attendance_stats(current_user: User = Depends(get_current_user)):
+ """Stats for attendance dashboard."""
+ headers, rows = read_excel_file("attendance")
+ if headers is None:
+ return {"total": 0, "by_department": {}, "by_status": {}}
+
+ dept_col = None
+ for i, h in enumerate(headers):
+ if h and any(k in str(h).lower() for k in ["company", "dept", "部門", "公司"]):
+ dept_col = i; break
+
+ dept_map = {}
+ for row in rows:
+ if dept_col is not None and dept_col < len(row):
+ d = str(row[dept_col] or "Unknown")
+ dept_map[d] = dept_map.get(d, 0) + 1
+
+ return {"total": len(rows), "by_department": dept_map, "by_status": {}}
+
+# ============ Accident Dashboard ============
+@app.get("/api/accident/dashboard/summary")
+async def accident_dashboard_summary(current_user: User = Depends(get_current_user)):
+ """Accident dashboard summary."""
+ headers, rows = read_excel_file("accident")
+ if headers is None:
+ return {"total": 0, "by_severity": {}, "by_location": {}, "recent": []}
+
+ sev_col = None; loc_col = None; date_col = None
+ for i, h in enumerate(headers):
+ if h and any(k in str(h) for k in ["緊急", "程度", "severity", "Severity"]):
+ sev_col = i
+ elif h and any(k in str(h) for k in ["地點", "部門", "location", "Location"]):
+ loc_col = i
+ elif h and any(k in str(h) for k in ["日期", "date", "Date"]):
+ date_col = i
+
+ sev_map = {}; loc_map = {}; recent = []
+ for row in rows:
+ if sev_col is not None and sev_col < len(row) and row[sev_col]:
+ sev_map[str(row[sev_col])] = sev_map.get(str(row[sev_col]), 0) + 1
+ if loc_col is not None and loc_col < len(row) and row[loc_col]:
+ loc_map[str(row[loc_col])] = loc_map.get(str(row[loc_col]), 0) + 1
+ if len(recent) < 10:
+ recent.append({"date": str(row[date_col] if date_col and date_col < len(row) else "")})
+
+ return {"total": len(rows), "by_severity": sev_map, "by_location": loc_map, "recent": recent}
+
+@app.get("/api/accident/dashboard/stats")
+async def accident_dashboard_stats(current_user: User = Depends(get_current_user)):
+ """Accident dashboard stats."""
+ headers, rows = read_excel_file("accident")
+ if headers is None:
+ return {"total": 0, "by_severity": {}, "by_location": {}, "recent": []}
+
+ sev_col = None; loc_col = None; date_col = None
+ for i, h in enumerate(headers):
+ if h and any(k in str(h) for k in ["緊急", "程度", "severity", "Severity"]):
+ sev_col = i
+ elif h and any(k in str(h) for k in ["地點", "部門", "location", "Location"]):
+ loc_col = i
+ elif h and any(k in str(h) for k in ["日期", "date", "Date"]):
+ date_col = i
+
+ sev_map = {}; loc_map = {}; recent = []
+ for row in rows:
+ if sev_col is not None and sev_col < len(row) and row[sev_col]:
+ sev_map[str(row[sev_col])] = sev_map.get(str(row[sev_col]), 0) + 1
+ if loc_col is not None and loc_col < len(row) and row[loc_col]:
+ loc_map[str(row[loc_col])] = loc_map.get(str(row[loc_col]), 0) + 1
+ if len(recent) < 10:
+ recent.append({"date": str(row[date_col] if date_col and date_col < len(row) else "")})
+
+ return {"total": len(rows), "by_severity": sev_map, "by_location": loc_map, "recent": recent}
+
+# ============ Dashboard (from Excel) ============
+@app.get("/api/dashboard/{section}")
+async def dashboard(section: str, current_user: User = Depends(get_current_user)):
+ """Get dashboard stats from stored Excel file"""
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ headers, rows = read_excel_file(section)
+ if headers is None or not rows:
+ return {"total": 0, "this_month": 0, "by_status": {}, "by_department": {}, "trend": []}
+
+ today = date.today()
+ month_start = date(today.year, today.month, 1)
+
+ # Find date column (look for common date column names)
+ date_col_idx = None
+ for i, h in enumerate(headers):
+ if h and any(kw in str(h).lower() for kw in ['date', '日期', '時間']):
+ date_col_idx = i
+ break
+
+ # Find department column
+ dept_col_idx = None
+ for i, h in enumerate(headers):
+ if h and any(kw in str(h).lower() for kw in ['dept', '部門', '部門名稱']):
+ dept_col_idx = i
+ break
+
+ total = len(rows)
+ this_month = 0
+ by_department = {}
+ trend = []
+
+ for row in rows:
+ # Count this month
+ if date_col_idx is not None and date_col_idx < len(row):
+ val = row[date_col_idx]
+ if val:
+ row_date = None
+ if isinstance(val, datetime):
+ row_date = val.date()
+ elif isinstance(val, date):
+ row_date = val
+ elif isinstance(val, str):
+ for fmt in ['%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y']:
+ try:
+ row_date = datetime.strptime(val, fmt).date()
+ break
+ except:
+ pass
+ if row_date and row_date >= month_start:
+ this_month += 1
+
+ # Count by department
+ if dept_col_idx is not None and dept_col_idx < len(row):
+ dept = str(row[dept_col_idx] or 'Unknown')
+ by_department[dept] = by_department.get(dept, 0) + 1
+
+ return {
+ "total": total,
+ "today": 0,
+ "this_month": this_month,
+ "by_status": {},
+ "by_department": by_department,
+ "trend": []
+ }
+
+# ============ Record Detail (from Excel) ============
+@app.get("/api/{section}/{row_id}")
+async def get_record(
+ section: str,
+ row_id: int,
+ current_user: User = Depends(get_current_user)
+):
+ """Get a specific record by row number"""
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ headers, rows = read_excel_file(section)
+ if headers is None:
+ raise HTTPException(status_code=404, detail="No data file found")
+
+ idx = row_id - 1 # Convert to 0-based index
+ if idx < 0 or idx >= len(rows):
+ raise HTTPException(status_code=404, detail="Record not found")
+
+ row = rows[idx]
+ record = {"_row": row_id}
+ for i, header in enumerate(headers):
+ if header:
+ record[str(header)] = row[i] if i < len(row) else None
+
+ return record
+
+# ============ Export ============
+@app.get("/api/export/{section}/excel")
+async def export_excel(
+ request: Request,
+ section: str,
+ token: Optional[str] = Query(None),
+ db: Session = Depends(get_db)
+):
+ """Export the stored Excel file"""
+ from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+ from jose import JWTError, jwt
+ from database import SECRET_KEY, ALGORITHM
+
+ credentials_exception = HTTPException(
+ status_code=401,
+ detail="Invalid or expired token",
+ headers={"WWW-Authenticate": "Bearer"},
+ )
+
+ current_user = None
+
+ # If token in query, use it
+ if token:
+ try:
+ payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+ user_id = int(payload.get("sub"))
+ current_user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
+ except (JWTError, ValueError, TypeError):
+ pass
+
+ # If no query token, try Authorization header
+ if not current_user:
+ auth_header = request.headers.get("Authorization", "")
+ if auth_header.startswith("Bearer "):
+ bearer_token = auth_header[7:]
+ try:
+ payload = jwt.decode(bearer_token, SECRET_KEY, algorithms=[ALGORITHM])
+ user_id = int(payload.get("sub"))
+ current_user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
+ except (JWTError, ValueError, TypeError):
+ pass
+
+ if not current_user:
+ raise credentials_exception
+
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ path = get_excel_path(section)
+ if not os.path.exists(path):
+ raise HTTPException(status_code=404, detail="No file found")
+
+ return FileResponse(
+ path,
+ filename=f"{section}_export.xlsx",
+ media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+ headers={"Content-Disposition": f"attachment; filename={section}_export.xlsx"}
+ )
+
+# ============ HTML5 Report Export ============
+import re
+
+def _make_chart_config(data, rows, section):
+ """Build chart config + stats from raw Excel rows."""
+ headers = data["headers"]
+ records = data["rows"]
+
+ # Build dicts
+ recs = []
+ for row in records:
+ rec = {}
+ for i, h in enumerate(headers):
+ if h:
+ rec[str(h)] = row[i] if i < len(row) else None
+ recs.append(rec)
+
+ # ---- Stats ----
+ stats = []
+
+ # Total records
+ stats.append({"label": "總記錄", "value": len(recs), "sub": "筆"})
+
+ if section == "attendance":
+ # Find relevant columns
+ staff_col = next((h for h in headers if h and any(k in str(h) for k in ["員工", "員工名稱", "employee", "name", "Name"])), None)
+ shift_col = next((h for h in headers if h and "班次" in str(h)), None)
+
+ # Count unique staff
+ unique_staff = set()
+ late_count = 0
+ early_count = 0
+ ot_count = 0
+ missing_count = 0
+
+ for r in recs:
+ if staff_col and r.get(staff_col):
+ unique_staff.add(r[staff_col])
+ # Status-like fields
+ for k, v in r.items():
+ if not v: continue
+ sv = str(v).lower()
+ if any(t in sv for t in ["missing", "缺勤"]): missing_count += 1
+ elif "遲" in sv or "late" in sv: late_count += 1
+ elif "早退" in sv or "early" in sv: early_count += 1
+ elif "ot" in sv or "超時" in sv: ot_count += 1
+
+ stats.append({"label": "員工人數", "value": len(unique_staff), "sub": "人"})
+ stats.append({"label": "遲到人次", "value": late_count, "sub": "次"})
+ stats.append({"label": "早退人次", "value": early_count, "sub": "次"})
+ stats.append({"label": "OT 人次", "value": ot_count, "sub": "次"})
+ stats.append({"label": "缺勤", "value": missing_count, "sub": "次"})
+
+ # Department count
+ dept_col = next((h for h in headers if h and any(k in str(h) for k in ["dept", "部門", "公司"])), None)
+ if dept_col:
+ dept_count = len(set(r.get(dept_col) for r in recs if r.get(dept_col)))
+ stats.append({"label": "部門數目", "value": dept_count, "sub": "個"})
+
+ elif section == "accident":
+ # Severity breakdown
+ sev_col = next((h for h in headers if h and any(k in str(h) for k in ["severity", "緊急", "程度", "Severity"])), None)
+ loc_col = next((h for h in headers if h and any(k in str(h) for k in ["location", "地點", "部門", "地點"])), None)
+ date_col = next((h for h in headers if h and any(k in str(h) for k in ["date", "日期", "時間", "Date"])), None)
+
+ sev_count = {}
+ loc_count = {}
+
+ for r in recs:
+ if sev_col and r.get(sev_col):
+ sev_count[r[sev_col]] = sev_count.get(r[sev_col], 0) + 1
+ if loc_col and r.get(loc_col):
+ loc_count[r[loc_col]] = loc_count.get(r[loc_col], 0) + 1
+
+ if sev_count:
+ total_sev = sum(sev_count.values())
+ stats.append({"label": "緊急/嚴重個案", "value": total_sev, "sub": "宗"})
+ if loc_count:
+ stats.append({"label": "出事地點", "value": len(loc_count), "sub": "個地點"})
+
+ # ---- Charts ----
+ charts = []
+
+ if section == "attendance":
+ # Dept bar chart
+ dept_col = next((h for h in headers if h and any(k in str(h) for k in ["dept", "部門", "公司", "company"])), None)
+ if dept_col:
+ dept_map = {}
+ for r in recs:
+ d = str(r.get(dept_col) or "Unknown")
+ dept_map[d] = dept_map.get(d, 0) + 1
+ sorted_depts = sorted(dept_map.items(), key=lambda x: x[1], reverse=True)[:8]
+ charts.append({
+ "type": "bar",
+ "data": {
+ "labels": [d[0][:12] for d in sorted_depts],
+ "datasets": [{"label": "人次", "data": [d[1] for d in sorted_depts], "backgroundColor": ["#3b82f6","#06b6d4","#10b981","#f59e0b","#ef4444","#8b5cf6","#ec4899","#64748b"]}]
+ },
+ "filterColumn": dept_col,
+ "tooltipCallbacks": {}
+ })
+
+ # Status pie chart
+ status_map = {"正常": 0, "遲到": 0, "早退": 0, "缺勤": 0, "OT": 0}
+ for r in recs:
+ for v in r.values():
+ if not v: continue
+ sv = str(v).lower()
+ if "missing" in sv or "缺勤" in sv: status_map["缺勤"] = status_map.get("缺勤", 0) + 1
+ elif "遲" in sv or "late" in sv: status_map["遲到"] = status_map.get("遲到", 0) + 1
+ elif "早退" in sv or "early" in sv: status_map["早退"] = status_map.get("早退", 0) + 1
+ elif "ot" in sv or "超時" in sv: status_map["OT"] = status_map.get("OT", 0) + 1
+ else: status_map["正常"] = status_map.get("正常", 0) + 1
+ pie_data = [v for v in status_map.values() if v > 0]
+ pie_labels = [k for k, v in status_map.items() if v > 0]
+ if pie_data:
+ charts.append({
+ "type": "doughnut",
+ "data": {
+ "labels": pie_labels,
+ "datasets": [{"data": pie_data, "backgroundColor": ["#10b981","#f59e0b","#f97316","#ef4444","#3b82f6"]}]
+ },
+ "filterColumn": None,
+ "tooltipCallbacks": {}
+ })
+
+ # Monthly/date trend
+ date_col = next((h for h in headers if h and any(k in str(h) for k in ["date", "日期", "Date"])), None)
+ if date_col:
+ date_map = {}
+ for r in recs:
+ d = r.get(date_col)
+ if d:
+ key = str(d)[:7] # YYYY-MM
+ date_map[key] = date_map.get(key, 0) + 1
+ sorted_dates = sorted(date_map.items())
+ charts.append({
+ "type": "line",
+ "data": {
+ "labels": [d[0] for d in sorted_dates],
+ "datasets": [{"label": "出勤人次", "data": [d[1] for d in sorted_dates], "borderColor": "#3b82f6", "backgroundColor": "rgba(59,130,246,0.1)", "fill": True, "tension": 0.3}]
+ },
+ "filterColumn": None,
+ "tooltipCallbacks": {}
+ })
+
+ elif section == "accident":
+ # Severity bar chart
+ sev_col = next((h for h in headers if h and any(k in str(h) for k in ["severity", "緊急", "程度"])), None)
+ if sev_col:
+ sev_map = {}
+ for r in recs:
+ s = str(r.get(sev_col) or "Unknown")
+ sev_map[s] = sev_map.get(s, 0) + 1
+ sorted_sev = sorted(sev_map.items(), key=lambda x: x[1], reverse=True)[:6]
+ charts.append({
+ "type": "bar",
+ "data": {
+ "labels": [s[0] for s in sorted_sev],
+ "datasets": [{"label": "個案", "data": [s[1] for s in sorted_sev], "backgroundColor": ["#ef4444","#f59e0b","#3b82f6","#10b981","#8b5cf6","#64748b"]}]
+ },
+ "filterColumn": sev_col,
+ "tooltipCallbacks": {}
+ })
+
+ # Location bar chart
+ loc_col = next((h for h in headers if h and any(k in str(h) for k in ["location", "地點", "部門", "地點"])), None)
+ if loc_col:
+ loc_map = {}
+ for r in recs:
+ l = str(r.get(loc_col) or "Unknown")
+ loc_map[l] = loc_map.get(l, 0) + 1
+ sorted_locs = sorted(loc_map.items(), key=lambda x: x[1], reverse=True)[:8]
+ charts.append({
+ "type": "bar",
+ "data": {
+ "labels": [l[0][:12] for l in sorted_locs],
+ "datasets": [{"label": "個案", "data": [l[1] for l in sorted_locs], "backgroundColor": ["#ef4444","#f59e0b","#3b82f6","#10b981","#8b5cf6","#ec4899","#06b6d4","#64748b"]}]
+ },
+ "filterColumn": loc_col,
+ "tooltipCallbacks": {}
+ })
+
+ # Monthly trend
+ date_col = next((h for h in headers if h and any(k in str(h) for k in ["date", "日期", "時間", "Date"])), None)
+ if date_col:
+ date_map = {}
+ for r in recs:
+ d = r.get(date_col)
+ if d:
+ key = str(d)[:7]
+ date_map[key] = date_map.get(key, 0) + 1
+ sorted_dates = sorted(date_map.items())
+ charts.append({
+ "type": "line",
+ "data": {
+ "labels": [d[0] for d in sorted_dates],
+ "datasets": [{"label": "意外個案", "data": [d[1] for d in sorted_dates], "borderColor": "#ef4444", "backgroundColor": "rgba(239,68,68,0.1)", "fill": True, "tension": 0.3}]
+ },
+ "filterColumn": None,
+ "tooltipCallbacks": {}
+ })
+
+ # Fill missing chart slots
+ while len(charts) < 3:
+ charts.append({"type": "bar", "data": {"labels": [], "datasets": [{"label": "無數據", "data": []}]}, "filterColumn": None, "tooltipCallbacks": {}})
+
+ return {"stats": stats, "charts": charts[:3]}
+
+
+@app.get("/api/report/{section}/html")
+async def export_html_report(
+ request: Request,
+ section: str,
+ token: Optional[str] = Query(None),
+):
+ """Generate self-contained HTML5 report with embedded data, charts, sorting, filtering, popup."""
+ from fastapi.responses import HTMLResponse
+ import json
+ from datetime import datetime
+ from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+ from jose import JWTError, jwt
+ from database import SECRET_KEY, ALGORITHM
+ from database import get_db
+
+ credentials_exception = HTTPException(
+ status_code=401,
+ detail="Invalid or expired token",
+ headers={"WWW-Authenticate": "Bearer"},
+ )
+
+ current_user = None
+
+ # If token in query, use it
+ if token:
+ try:
+ payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+ user_id = int(payload.get("sub"))
+ db = next(get_db())
+ current_user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
+ db.close()
+ except (JWTError, ValueError, TypeError):
+ pass
+
+ # If no query token, try Authorization header
+ if not current_user:
+ auth_header = request.headers.get("Authorization", "")
+ if auth_header.startswith("Bearer "):
+ bearer_token = auth_header[7:]
+ try:
+ payload = jwt.decode(bearer_token, SECRET_KEY, algorithms=[ALGORITHM])
+ user_id = int(payload.get("sub"))
+ db = next(get_db())
+ current_user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
+ db.close()
+ except (JWTError, ValueError, TypeError):
+ pass
+
+ if not current_user:
+ raise credentials_exception
+
+ if section not in ["attendance", "accident"]:
+ raise HTTPException(status_code=400, detail="Invalid section")
+
+ # Read data
+ headers, rows = read_excel_file(section)
+ if headers is None:
+ return HTMLResponse(
+ content=""
+ "
暫無數據
請先上傳 Excel 檔案再生成報告。
"
+ "
返回首頁",
+ media_type="text/html"
+ )
+
+ # Build records
+ records = []
+ for row_idx, row in enumerate(rows, 1):
+ rec = {"_row": row_idx}
+ for i, h in enumerate(headers):
+ if h:
+ rec[str(h)] = row[i] if i < len(row) else None
+ records.append(rec)
+
+ # Chart + stats config
+ chart_cfg = _make_chart_config({"headers": headers, "rows": rows}, rows, section)
+
+ # Column defs
+ col_defs = []
+ hide_keys = {"_row"}
+ for h in headers:
+ if not h or h in hide_keys:
+ continue
+ label = str(h)
+ col_defs.append({"key": str(h), "label": label, "hide": False})
+
+ # Template config
+ if section == "attendance":
+ title = "出勤 Attendance"
+ header_color = "#2563EB"
+ header_dark = "#1D4ED8"
+ accent = "#2563EB"
+ else:
+ title = "意外 Accident"
+ header_color = "#DC2626"
+ header_dark = "#B91C1C"
+ accent = "#DC2626"
+
+ generated_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+ total_records = len(records)
+
+ # Column options for filter
+ col_options = "".join(f'
' for c in col_defs[:8])
+
+ # Serialize for embedding
+ def safe_json(obj):
+ return json.dumps(obj, default=lambda x: None, ensure_ascii=False)
+
+ embedded_data = safe_json(records)
+ column_defs = safe_json(col_defs)
+ chart_config = safe_json(chart_cfg)
+
+ # Read template
+ import os
+ template_path = os.path.join(os.path.dirname(__file__), "report_template.html")
+ if os.path.exists(template_path):
+ with open(template_path, "r", encoding="utf-8") as f:
+ template = f.read()
+ else:
+ return HTMLResponse(content="
Template not found
", status_code=500)
+
+ # Replace placeholders
+ html = template
+ html = html.replace("{{TITLE}}", title)
+ html = html.replace("{{REPORT_TITLE}}", title)
+ html = html.replace("{{GENERATED_DATE}}", generated_date)
+ html = html.replace("{{TOTAL_RECORDS}}", str(total_records))
+ html = html.replace("{{HEADER_COLOR}}", header_color)
+ html = html.replace("{{HEADER_DARK}}", header_dark)
+ html = html.replace("{{ACCENT_COLOR}}", accent)
+ html = html.replace("{{COLUMN_OPTIONS}}", col_options)
+ html = html.replace("{{EMBEDDED_DATA}}", embedded_data)
+ html = html.replace("{{COLUMN_DEFS}}", column_defs)
+ html = html.replace("{{CHART_CONFIG}}", chart_config)
+
+ filename = f"{section}_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
+ headers_dict = {"Content-Disposition": f"attachment; filename={filename}"}
+ return HTMLResponse(content=html, media_type="text/html", headers=headers_dict)
+
+# ============ Roster / Shift Management ============
+
+@app.get("/")
+async def root():
+ return RedirectResponse(url="/index.html")
+
+if __name__ == "__main__":
+ import uvicorn
+ uvicorn.run(app, host="0.0.0.0", port=8000)