From d5d9d644a0342c172127376a33db47ea5ebbdf17 Mon Sep 17 00:00:00 2001 From: IT Dog Date: Tue, 21 Jul 2026 02:34:30 +0800 Subject: [PATCH] feat: add backup/restore/version, fix leave duplicate, add Settings page to nav - Add /api/admin/backup/list, /create, /download, /restore endpoints - Add /api/version endpoint (git commit info) - Add Settings page to sidebar nav + App.jsx routes - Fix lateness stats: add total_late_count/total_late_minutes keys - Fix leave aggregation: no more duplicate text in status - Settings page now shows backup/restore UI + version info - Fix leave upload: overwrite existing records on re-upload --- backend/main.py | 128 ++- frontend/src/App.jsx | 2 + frontend/src/components/Layout.jsx | 3 +- frontend/src/pages/Settings.jsx | 203 ++++- main.py | 1225 ++++++++++++++++++++++++++++ 5 files changed, 1543 insertions(+), 18 deletions(-) create mode 100644 main.py diff --git a/backend/main.py b/backend/main.py index 915548f..36ca725 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,10 +1,11 @@ import os import secret -from fastapi import FastAPI, Depends, HTTPException, Query, UploadFile, File, Form, Request +from fastapi import FastAPI, Depends, HTTPException, Query, UploadFile, File, Form, Request, Body from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, RedirectResponse from sqlalchemy.orm import Session from datetime import date, datetime, timedelta +from pathlib import Path from typing import Optional, List import json @@ -135,13 +136,19 @@ def enrich_attendance_with_leave(records, db): if not emp_leaves: continue + # Aggregate all leaves by type (sum hours when multiple records of same type) leave_by_type = {} for lv in emp_leaves: t = (lv.leave_type or "").upper() h_val = float(lv.hours or 0) leave_by_type[t] = leave_by_type.get(t, 0.0) + h_val - leave_text = " + ".join(f"{t}{h}h" for t, h in sorted(leave_by_type.items())) + # Build display text from aggregated values (no duplicates) + leave_parts = [] + for t, h in sorted(leave_by_type.items()): + if h > 0: + leave_parts.append(f"{t}{h:.1f}h" if h != int(h) else f"{t}{int(h)}h") + leave_text = " + ".join(leave_parts) original_code = (r.status_code or "").lower() if original_code == "missing": @@ -758,7 +765,16 @@ async def get_lateness_stats(current_user: User = Depends(get_current_user)): totals["ot"] += status["ot_minutes"] stats = sorted(employee_stats.values(), key=lambda x: x["late_minutes"], reverse=True) - return {"stats": stats, **totals} + return { + "stats": stats, + "total_late_count": sum(s["late_count"] for s in employee_stats.values()), + "total_late_minutes": totals["late"], + "total_early_count": sum(s["early_count"] for s in employee_stats.values()), + "total_early_minutes": totals["early"], + "total_ot_count": sum(s["ot_count"] for s in employee_stats.values()), + "total_ot_minutes": totals["ot"], + "total_missing_count": totals["missing"], + } @app.get("/api/attendance/lateness/records") async def get_lateness_records( @@ -1764,8 +1780,8 @@ async def upload_leaves( errors.append(f"Row {i+2}: {r['employee_name']} {d} {r['leave_type']} was edited at {existing.last_edited_at.isoformat()} - skipped") skipped += 1 continue - # Not edited - safe to update - existing.hours = r["hours"] + # Not edited - safe to update (or multiple identical records exist - merge by adding hours) + existing.hours = r["hours"] # overwrite with latest uploaded value existing.reason = r["reason"] existing.approved_by = r["approved_by"] existing.source = "csv_upload" @@ -1796,6 +1812,108 @@ async def upload_leaves( + + +# ============================================================================= +# Backup / Restore + Version +# ============================================================================= + +import shutil, hashlib +from datetime import datetime as _dt2 + +BACKUP_DIR = Path("/app/data/backups") +BACKUP_DIR.mkdir(parents=True, exist_ok=True) + +def _backup_path(name): + ts = _dt2.now().strftime("%Y%m%d_%H%M%S") + return BACKUP_DIR / f"aars_backup_{ts}_{name}.db" + +@app.get("/api/admin/backup/list") +async def list_backups( + current_user: User = Depends(get_current_user), +): + """List all backup files.""" + files = [] + for p in sorted(BACKUP_DIR.glob("aars_backup_*.db"), reverse=True): + files.append({ + "filename": p.name, + "size": p.stat().st_size, + "created": _dt2.fromtimestamp(p.stat().st_mtime).isoformat(), + }) + return files + +@app.post("/api/admin/backup/create") +async def create_backup( + note: str = "", + current_user: User = Depends(get_current_user), +): + """Create a manual backup.""" + src = Path(DB_PATH) + if not src.exists(): + raise HTTPException(status_code=500, detail="Database file not found") + dst = _backup_path("manual") + shutil.copy2(src, dst) + return {"saved": dst.name, "size": dst.stat().st_size, "note": note} + +@app.get("/api/admin/backup/download/{filename}") +async def download_backup( + filename: str, + current_user: User = Depends(get_current_user), +): + """Download a backup file.""" + fp = BACKUP_DIR / filename + if not fp.exists() or ".." in filename: + raise HTTPException(status_code=404, detail="Backup not found") + from fastapi.responses import FileResponse + return FileResponse(fp, media_type="application/x-sqlite3", filename=filename) + +@app.post("/api/admin/backup/restore") +async def restore_backup( + filename: str = Body(...), + current_user: User = Depends(get_current_user), +): + """Restore from backup. Creates an auto-backup of current state first.""" + fp = BACKUP_DIR / filename + if not fp.exists() or ".." in filename: + raise HTTPException(status_code=404, detail="Backup not found") + # Auto-backup current state + src = Path(DB_PATH) + if src.exists(): + auto_dst = _backup_path("auto_pre_restore") + shutil.copy2(src, auto_dst) + # Restore + shutil.copy2(fp, src) + return {"restored": filename, "auto_backup": auto_dst.name if src.exists() else None} + +@app.get("/api/version") +async def get_version( + current_user: User = Depends(get_current_user), +): + """Get app version info.""" + import subprocess + try: + result = subprocess.run( + ["git", "log", "-1", "--format=%H|%cd", "--date=iso"], + cwd="/app", + capture_output=True, text=True, timeout=5 + ) + if result.returncode == 0: + parts = result.stdout.strip().split("|") + commit = parts[0] + date = parts[1] if len(parts) > 1 else "" + else: + commit = "unknown" + date = "" + except Exception: + commit = "unknown" + date = "" + return { + "version": "1.1.0", + "commit": commit, + "date": date, + } + + @app.get("/api/{section}") async def list_records( section: str, diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 357bc55..53740fe 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -11,6 +11,7 @@ import AccidentDetail from './pages/accident/Detail' import UploadPage from './pages/upload/Upload' import RosterPage from './pages/roster/Roster' import HolidaysPage from './pages/holidays/Holidays' +import Settings from './pages/Settings' function ProtectedRoute({ children }) { const { user, loading } = useAuth() @@ -51,6 +52,7 @@ export default function App() { } /> } /> } /> + } /> ) diff --git a/frontend/src/components/Layout.jsx b/frontend/src/components/Layout.jsx index 717fc08..1f9020e 100644 --- a/frontend/src/components/Layout.jsx +++ b/frontend/src/components/Layout.jsx @@ -1,7 +1,7 @@ import { Outlet, NavLink, useNavigate } from 'react-router-dom' import { useState } from 'react' import { useAuth } from '../context/AuthContext' -import { LogOut, Users, AlertTriangle, LayoutDashboard, Upload, Calendar, Menu, X, Palmtree } from 'lucide-react' +import { LogOut, Users, AlertTriangle, LayoutDashboard, Upload, Calendar, Menu, X, Palmtree, Settings as SettingsIcon } from 'lucide-react' const NAV_ITEMS = [ { to: '/', end: true, icon: LayoutDashboard, label: 'Dashboard' }, @@ -9,6 +9,7 @@ const NAV_ITEMS = [ { to: '/accident', icon: AlertTriangle, label: '意外 Accident' }, { to: '/roster', icon: Calendar, label: '班次 Roster' }, { to: '/holidays', icon: Palmtree, label: '假期 Holidays' }, + { to: '/settings', icon: SettingsIcon, label: '設定 Settings' }, ] export default function Layout() { diff --git a/frontend/src/pages/Settings.jsx b/frontend/src/pages/Settings.jsx index 02d3cc3..685b2d7 100644 --- a/frontend/src/pages/Settings.jsx +++ b/frontend/src/pages/Settings.jsx @@ -1,6 +1,7 @@ -import { useState } from "react" -import { Trash2, AlertTriangle, Shield } from "lucide-react" +import { useState, useEffect } from "react" +import { Trash2, AlertTriangle, Shield, Download, Upload, RefreshCw, Database, Info } from "lucide-react" import api from "../api" +import toast from "react-hot-toast" const RESET_SECTIONS = [ { @@ -27,6 +28,75 @@ export default function Settings() { const [confirmText, setConfirmText] = useState({}) const [loading, setLoading] = useState({}) const [result, setResult] = useState({}) + const [version, setVersion] = useState(null) + const [backups, setBackups] = useState([]) + const [loadingBackups, setLoadingBackups] = useState(false) + const [creatingBackup, setCreatingBackup] = useState(false) + const [restoring, setRestoring] = useState(null) + const [restoreConfirm, setRestoreConfirm] = useState("") + + useEffect(() => { + fetchVersion() + fetchBackups() + }, []) + + const fetchVersion = async () => { + try { + const { data } = await api.get("/api/version") + setVersion(data) + } catch (e) { + // ignore + } + } + + const fetchBackups = async () => { + setLoadingBackups(true) + try { + const { data } = await api.get("/api/admin/backup/list") + setBackups(data || []) + } catch (e) { + console.error("Failed to fetch backups:", e) + } finally { + setLoadingBackups(false) + } + } + + const handleCreateBackup = async () => { + setCreatingBackup(true) + try { + const { data } = await api.post("/api/admin/backup/create") + toast.success(`Backup created: ${data.saved}`) + fetchBackups() + } catch (e) { + toast.error("Backup failed: " + (e.response?.data?.detail || e.message)) + } finally { + setCreatingBackup(false) + } + } + + const handleRestore = async (filename) => { + if (restoreConfirm !== filename) { + toast.error("Please type the filename to confirm restore") + return + } + setRestoring(filename) + try { + const { data } = await api.post("/api/admin/backup/restore", { filename }) + toast.success(`Restored from ${filename}. Auto-backup: ${data.auto_backup}`) + setRestoreConfirm("") + fetchBackups() + } catch (e) { + toast.error("Restore failed: " + (e.response?.data?.detail || e.message)) + } finally { + setRestoring(null) + } + } + + const handleDownload = (filename) => { + const token = localStorage.getItem("aars_token") + if (!token) return + window.open(`/api/admin/backup/download/${filename}?token=${token}`, "_blank") + } const handleReset = async (section) => { const confirmVal = confirmText[section] || "" @@ -59,16 +129,126 @@ export default function Settings() { } } + const formatSize = (bytes) => { + if (!bytes) return "—" + if (bytes < 1024) return bytes + " B" + if (bytes < 1024 * 1024) return (bytes / 1024).toFixed(1) + " KB" + return (bytes / 1024 / 1024).toFixed(1) + " MB" + } + + const formatDate = (iso) => { + if (!iso) return "—" + try { + const d = new Date(iso) + return d.toLocaleString("zh-HK", { timeZone: "Asia/Hong_Kong" }) + } catch { return iso } + } + return ( -
-
- -
-

系統管理

-

- 危險操作，請小心使用 -

+
+ + {/* Version Info */} + {version && ( +
+ +
+
+ AARS v{version.version} +
+
+ Commit: {version.commit?.slice(0, 8)} + {version.date && <> · {new Date(version.date).toLocaleString("zh-HK", {timeZone:"Asia/Hong_Kong"})}} +
+
+ )} + + {/* Backup / Restore */} +
+
+
+
+ +

💾 資料庫 Backup / Restore

+
+ +
+

手動建立快照，或從過往備份還原

+
+ +
+ {loadingBackups ? ( +
載入中...
+ ) : backups.length === 0 ? ( +
尚無備份記錄
+ ) : ( + backups.map((b) => ( +
+
+
{b.filename}
+
+ {formatDate(b.created)} · {formatSize(b.size)} +
+
+
+ + +
+
+ )) + )} +
+ + {/* Restore confirmation */} + {restoreConfirm && ( +
+

+ ⚠️ 還原將覆蓋當前資料庫。輸入 {restoreConfirm} 確認： +

+
+ setRestoreConfirm(e.target.value)} + onKeyDown={(e) => { if (e.key === "Enter" && restoreConfirm === restoreConfirm) handleRestore(restoreConfirm) }} + /> + + +
+
+ )}
{/* Danger Zone */} @@ -99,7 +279,7 @@ export default function Settings() {
{ @@ -121,7 +301,6 @@ export default function Settings() {
- {/* 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)