import os
import secret
from fastapi import FastAPI, Depends, HTTPException, Query, UploadFile, File, Form, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, RedirectResponse
from sqlalchemy.orm import Session
from datetime import date, datetime, timedelta
from typing import Optional, List
import json
from database import get_db, create_access_token, verify_password, init_db, Base, engine
from models import User
from schemas import *
from auth import get_current_user, get_current_admin, get_current_user_from_query
app = FastAPI(title="AARS API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Data directory for storing uploaded Excel files
DATA_DIR = "/app/data"
os.makedirs(DATA_DIR, exist_ok=True)
@app.on_event("startup")
async def startup():
init_db()
# ============ Auth ============
@app.post("/api/auth/login", response_model=TokenResponse)
async def login(form_data: LoginRequest, db: Session = Depends(get_db)):
user = db.query(User).filter(User.email == form_data.email).first()
if not user or not verify_password(form_data.password, user.password_hash):
raise HTTPException(status_code=401, detail="Invalid email or password")
access_token = create_access_token({"sub": str(user.id)})
return TokenResponse(access_token=access_token)
@app.get("/api/auth/me", response_model=UserResponse)
async def get_me(current_user: User = Depends(get_current_user)):
return current_user
# ============ Excel File Storage ============
def get_excel_path(section: str) -> 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)