IT狗
535d10a885
- Backend: /api/report/{section}/html endpoint generates self-contained HTML5 report
- Frontend: Export buttons added to Dashboard, Attendance List, Accident List
- Report features: embedded data, 3 interactive Chart.js charts, stats cards,
sortable/filterable table with color-coded rows, popup detail on click,
click chart segment to auto-filter table, print support
- Template: report_template.html with responsive design, Chinese labels
1089 lines
40 KiB
Python
1089 lines
40 KiB
Python
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 (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(
|
|
section: str,
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Generate self-contained HTML5 report with embedded data, charts, sorting, filtering, popup."""
|
|
from fastapi.responses import HTMLResponse
|
|
import json
|
|
from datetime import datetime
|
|
|
|
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="<html><body style='font-family:sans-serif;padding:40px;text-align:center;color:#666;'>"
|
|
"<h2>暫無數據</h2><p>請先上傳 Excel 檔案再生成報告。</p>"
|
|
"<a href='/'>返回首頁</a></body></html>",
|
|
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'<option value="{c["key"]}">{c["label"]}</option>' 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="<html><body><h2>Template not found</h2></body></html>", 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)
|