import os
import secret
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
from database import get_db, create_access_token, verify_password, init_db, Base, engine
from models import User, AttendanceRecord, Shift, Accident, Holiday, LeaveRecord
from sqlalchemy import func as sqlfunc
from schemas import *
from auth import get_current_user, get_current_admin, get_current_user_from_query
from pydantic import BaseModel, Field
# ============ Pydantic Schemas (Holidays + Leave) ============
class HolidayCreate(BaseModel):
date: date
name: str
name_en: Optional[str] = None
region: Optional[str] = "HK"
is_mandatory: Optional[bool] = True
notes: Optional[str] = None
class HolidayUpdate(BaseModel):
date: Optional[date] = None
name: Optional[str] = None
name_en: Optional[str] = None
region: Optional[str] = None
is_mandatory: Optional[bool] = None
notes: Optional[str] = None
class HolidayOut(BaseModel):
id: int
date: date
name: str
name_en: Optional[str] = None
region: Optional[str] = None
is_mandatory: Optional[bool] = None
source: Optional[str] = None
notes: Optional[str] = None
last_edited_at: Optional[datetime] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class Config:
from_attributes = True
class LeaveCreate(BaseModel):
employee_name: str
employee_id: Optional[str] = None
date: date
leave_type: str
hours: float
reason: Optional[str] = None
approved_by: Optional[str] = None
notes: Optional[str] = None
class LeaveUpdate(BaseModel):
employee_name: Optional[str] = None
employee_id: Optional[str] = None
date: Optional[date] = None
leave_type: Optional[str] = None
hours: Optional[float] = None
reason: Optional[str] = None
approved_by: Optional[str] = None
notes: Optional[str] = None
class LeaveOut(BaseModel):
id: int
employee_name: str
employee_id: Optional[str] = None
date: date
leave_type: str
hours: float
reason: Optional[str] = None
approved_by: Optional[str] = None
source: Optional[str] = None
notes: Optional[str] = None
last_edited_at: Optional[datetime] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class Config:
from_attributes = True
def enrich_attendance_with_leave(records, db):
"""Apply holiday + leave override to attendance records (in-place)."""
from collections import defaultdict as _dd
if not records:
return records
dates = [r.date for r in records if r.date]
if not dates:
return records
min_date, max_date = min(dates), max(dates)
holidays = db.query(Holiday).filter(
Holiday.date >= min_date, Holiday.date <= max_date
).all()
holidays_by_date = {h.date: h for h in holidays}
emp_names = list({r.employee_name for r in records if r.employee_name})
leaves = []
if emp_names:
leaves = db.query(LeaveRecord).filter(
LeaveRecord.date >= min_date, LeaveRecord.date <= max_date,
LeaveRecord.employee_name.in_(emp_names),
).all()
leave_by_emp_date = _dd(lambda: _dd(list))
for lv in leaves:
leave_by_emp_date[lv.employee_name][lv.date].append(lv)
for r in records:
if not r.date:
continue
if r.date in holidays_by_date:
h = holidays_by_date[r.date]
# Check roster: if shift's public_holiday == 'Yes', employee works on holidays
# Don't override with holiday status in that case
shift_holiday = '休息' # default: assume rest on holiday
if r.shift_code:
shift_info = get_shift_schedule_full(r.shift_code, r.weekday or '')
if shift_info and len(shift_info) >= 3:
shift_holiday = shift_info[2]
if shift_holiday == 'Yes':
# Employee works on public holidays - keep attendance status, append holiday note
base = r.status_text or "正常"
r.status_text = f"{base} (🎉{h.name})"
continue
# Employee rests on public holidays - mark as holiday unless already has clock-in
has_clock_in = (
(r.check_in is not None) or
(r.actual_in and r.actual_in not in ("", "-"))
)
if not has_clock_in:
r.status_code = "holiday"
r.status_text = f"\U0001F3D6\uFE0F{h.name}"
continue
# Has clock-in but roster says rest on holiday - still show holiday
r.status_code = "holiday"
r.status_text = f"\U0001F3D6\uFE0F{h.name}"
continue
emp_leaves = leave_by_emp_date.get(r.employee_name, {}).get(r.date, [])
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
# 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":
types = sorted(leave_by_type.keys())
if len(types) == 1:
r.status_code = types[0].lower()
else:
r.status_code = "mixed_leave"
r.status_text = leave_text
elif original_code in ("al", "sl", "cl"):
# Pure leave status: replace with aggregated leave text
if len(leave_by_type) > 1:
r.status_code = "mixed_leave"
r.status_text = leave_text
else:
r.status_text = leave_text
elif original_code not in ("missing", "", "holiday", "al", "sl", "cl"):
# Non-leave attendance status: strip any previously-enriched leave suffix and rebuild
base_text = (r.status_text or "正常").split(" + ")[0] # keep only attendance part
base_text = base_text.split(" (")[0] # strip holiday annotation
if leave_text:
r.status_text = f"{base_text} + {leave_text}"
else:
r.status_text = base_text
else:
# original_code is empty or holiday or already handled above
base = r.status_text or "正常"
r.status_text = f"{base} + {leave_text}"
return records
import logging
from logging_config import setup_logging, set_request_id, set_user_id
from middleware import (
RequestIDMiddleware,
AccessLogMiddleware,
unhandled_exception_handler,
)
setup_logging()
logger = logging.getLogger(__name__)
app = FastAPI(title="AARS API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Structured logging middleware
app.add_middleware(AccessLogMiddleware)
app.add_middleware(RequestIDMiddleware)
# Catch-all exception handler (after middleware)
app.add_exception_handler(Exception, unhandled_exception_handler)
# 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()
logger.info("AARS backend started")
# ============ 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):
logger.warning("login failed", extra={"context": {"email": form_data.email, "reason": "invalid_credentials"}})
raise HTTPException(status_code=401, detail="Invalid email or password")
set_user_id(user.id)
access_token = create_access_token({"sub": str(user.id)})
logger.info("login success", extra={"context": {"user_id": user.id, "email": form_data.email}})
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(...),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Upload Excel file - saves file AND imports rows into SQL table."""
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 (keep for legacy compatibility + export)
path = get_excel_path(section)
contents = await file.read()
with open(path, "wb") as f:
f.write(contents)
# Read rows back
headers, rows = read_excel_file(section)
if headers is None:
return {
"message": f"{section} file uploaded (no rows parsed)",
"filename": file.filename, "rows": 0, "columns": 0, "headers": [],
}
if section == "attendance":
imported = await _import_attendance_to_db(db, headers, rows, current_user.id)
logger.info("attendance import to DB",
extra={"context": {"user_id": current_user.id,
"imported": imported["imported"],
"updated": imported["updated"],
"skipped": imported["skipped"]}})
return {
"message": f"attendance uploaded + imported",
"filename": file.filename,
"rows": len(rows),
"columns": len(headers),
"headers": headers,
"db_import": imported,
}
if section == "accident":
imported = await _import_accident_to_db(db, headers, rows, current_user.id)
logger.info("accident import to DB",
extra={"context": {"user_id": current_user.id,
"imported": imported["imported"],
"updated": imported["updated"],
"errors": imported["errors"]}})
return {
"message": f"accident uploaded + imported",
"filename": file.filename,
"rows": len(rows),
"columns": len(headers),
"headers": headers,
"db_import": imported,
}
return {
"message": f"{section} file uploaded",
"filename": file.filename,
"rows": len(rows),
"columns": len(headers),
"headers": headers,
}
async def _import_attendance_to_db(db: Session, headers, rows, user_id: int) -> dict:
"""Parse Excel rows and upsert into attendance_records table.
Skips records with is_manually_edited=True (manual edits are preserved).
Returns counts: imported (new), updated (existing overwritten), skipped.
"""
# Find column indices from header
col = {}
for i, h in enumerate(headers):
if not h:
continue
hs = str(h).strip().lower()
if hs in ("company",):
col["company"] = i
elif "staff" in hs or "員工" in str(h) or "name" in hs:
col["staff"] = i
elif "dept" in hs or "部門" in str(h):
col["department"] = i
elif hs in ("date", "日期") or "date" in hs:
col["date"] = i
elif "week" in hs or "星期" in str(h):
col["weekday"] = i
elif "check" in hs and "in" in hs or "actual" in hs and "in" in hs or "上班" in str(h) or "實際" in str(h) and "上班" in str(h):
col["check_in"] = i
elif "check" in hs and "out" in hs or "actual" in hs and "out" in hs or "下班" in str(h) or "實際" in str(h) and "下班" in str(h):
col["check_out"] = i
elif "班次" in str(h) or "shift" in hs:
col["shift"] = i
imported = updated = skipped = errors = 0
for row in rows:
try:
def cell(k, default=None):
idx = col.get(k)
if idx is None or idx >= len(row):
return default
return row[idx]
staff = cell("staff")
date_val = cell("date")
if not staff or not date_val:
errors += 1
continue
# Parse date
if hasattr(date_val, "date"):
d = date_val.date() if isinstance(date_val, datetime) else date_val
elif isinstance(date_val, date):
d = date_val
elif isinstance(date_val, str):
d = None
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"):
try:
d = datetime.strptime(date_val, fmt).date()
break
except ValueError:
continue
if d is None:
errors += 1
continue
else:
errors += 1
continue
# Find existing
existing = db.query(AttendanceRecord).filter(
AttendanceRecord.employee_name == staff,
AttendanceRecord.date == d,
).first()
if existing and (existing.last_edited_at is not None or existing.is_manually_edited):
skipped += 1
continue
# Parse times
def parse_time(v, fallback_date):
if not v:
return None
if isinstance(v, datetime):
return v
if isinstance(v, date):
return None
if isinstance(v, str):
for fmt in ("%H:%M:%S", "%H:%M"):
try:
return datetime.strptime(v, fmt).time()
except ValueError:
continue
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M",
"%d/%m/%Y %H:%M:%S", "%d/%m/%Y %H:%M"):
try:
return datetime.strptime(v, fmt)
except ValueError:
continue
return None
ci_raw = cell("check_in")
co_raw = cell("check_out")
ci_dt = parse_time(ci_raw, d)
co_dt = parse_time(co_raw, d)
# Combine date + time
def to_dt(t, base_date):
if t is None:
return None
if isinstance(t, datetime):
return t
if isinstance(t, time):
return datetime.combine(base_date, t)
return None
ci_full = to_dt(ci_dt, d)
co_full = to_dt(co_dt, d)
shift_code = cell("shift")
# Calculate status using existing helper
shift_start = None
shift_end = None
if shift_code:
shift_obj = db.query(Shift).filter(Shift.shift_code == shift_code).first()
if shift_obj:
weekday_en = cell("weekday") or d.strftime("%A")
start_str, end_str = shift_obj.get_schedule(weekday_en)
try:
if start_str and end_str:
shift_start = datetime.strptime(start_str, "%H:%M").time()
shift_end = datetime.strptime(end_str, "%H:%M").time()
except (ValueError, TypeError):
pass
status = calculate_attendance_status(ci_full, co_full, shift_start, shift_end)
record = existing or AttendanceRecord(
employee_name=staff, date=d,
)
record.company = cell("company")
record.department = cell("department")
record.date = d
record.weekday = cell("weekday")
record.check_in = ci_full
record.check_out = co_full
record.shift_code = shift_code
record.status_code = status["status_code"]
record.status_text = status["status_text"]
record.late_minutes = status["late_minutes"]
record.early_minutes = status["early_minutes"]
record.ot_minutes = status["ot_minutes"]
record.expected_in = status["expected_in"]
record.expected_out = status["expected_out"]
record.actual_in = status["actual_in"]
record.actual_out = status["actual_out"]
record.raw_data = {"source_row": list(row)}
if existing:
updated += 1
else:
db.add(record)
imported += 1
except Exception as e:
errors += 1
logger.error("attendance import row failed",
extra={"context": {"error": str(e)[:200], "row": list(row)[:8]}})
db.commit()
return {"imported": imported, "updated": updated, "skipped": skipped, "errors": errors}
# ============ 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, public_holiday) tuple.
public_holiday is 'Yes' if the shift requires working on public holidays, '休息' otherwise.
"""
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
ph_col = None
for i, h in enumerate(headers):
if h == '班次':
shift_col = i
elif h == 'public holiday':
ph_col = i
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:
public_holiday = str(row[ph_col]).strip() if ph_col is not None else '休息'
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, public_holiday)
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:00 exactly)
# Only treat second-precise 18:00:00 as the default, not 18:00:03 (real swipe)
if (check_in_dt.hour == 18 and check_in_dt.minute == 0 and check_in_dt.second == 0
and check_out_dt.hour == 18 and check_out_dt.minute == 0 and check_out_dt.second == 0):
result["is_missing"] = True
result["status_code"] = "missing"
result["status_text"] = "缺勤"
return result
# Detect same check-in/out (exact same second = forgotten swipe)
if check_in_dt and check_out_dt:
if check_in_dt == check_out_dt:
result["is_missing"] = True
result["status_code"] = "abnormal"
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"] = round(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_seconds = (expected_end_dt - check_out_dt).total_seconds()
result["early_minutes"] = round(diff_seconds / 60) if diff_seconds >= 30 else 0
# Calculate OT minutes (leaving after scheduled end)
if check_out_dt > expected_end_dt:
diff_seconds = (check_out_dt - expected_end_dt).total_seconds()
# Ignore tiny differences (< 30 seconds) to avoid floating-point noise
if diff_seconds >= 30:
result["ot_minutes"] = round(diff_seconds / 60)
else:
result["ot_minutes"] = 0
# 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 "正常"
# If late / early / OT > 30 min → mark as abnormal (override after status_text is set)
if result["late_minutes"] > 30 or result["early_minutes"] > 30 or result["ot_minutes"] > 30:
result["status_code"] = "abnormal"
if result["late_minutes"] > 30:
result["status_text"] = f"異常-遲到{result['late_minutes']}分"
elif result["early_minutes"] > 30:
result["status_text"] = f"異常-早退{result['early_minutes']}分"
else:
result["status_text"] = f"異常-OT{result['ot_minutes']}分"
return result
@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,
"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(
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) ============
# ============ Frontend Compat Aliases ============
# Inserted BEFORE wildcard routes to win FastAPI route matching.
@app.get("/api/dashboard/summary")
async def dashboard_summary(
date_from: Optional[str] = None,
date_to: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Read from attendance_records table (SQL). Returns frontend-expected field names."""
from datetime import date as _date, datetime as _dt
q = db.query(AttendanceRecord)
if date_from:
try:
q = q.filter(AttendanceRecord.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(AttendanceRecord.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
rows = q.all()
today = _date.today()
month_start = today.replace(day=1)
# Frontend expects total_records + count-by-status fields
by_status = {}
by_department = {}
staff_set = set()
for r in rows:
sc = r.status_code or "unknown"
by_status[sc] = by_status.get(sc, 0) + 1
if r.department:
by_department[r.department] = by_department.get(r.department, 0) + 1
if r.employee_name:
staff_set.add(r.employee_name)
return {
# Backend / dashboard canonical fields
"total": len(rows),
"today": sum(1 for r in rows if r.date == today),
"this_month": sum(1 for r in rows if r.date and r.date >= month_start),
"by_status": by_status,
"by_department": by_department,
"trend": [],
# Frontend-expected fields (alias for legacy UI)
"total_records": len(rows),
"normal_count": by_status.get("normal", 0),
"late_count": sum(v for k, v in by_status.items() if "late" in k),
"early_count": sum(v for k, v in by_status.items() if "early" in k),
"ot_count": sum(v for k, v in by_status.items() if "ot" in k),
"missing_count": by_status.get("missing", 0),
"abnormal_count": by_status.get("abnormal", 0),
"staff_count": len(staff_set),
}
@app.get("/api/attendance/status_texts")
async def attendance_status_texts(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Return distinct status_text values for filter dropdown."""
rows = db.query(AttendanceRecord.status_text).distinct().all()
return [r[0] for r in rows if r[0]]
@app.get("/api/attendance/stats")
async def attendance_stats(
date_from: Optional[str] = None,
date_to: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Read aggregated stats from attendance_records (SQL).
Returns Option 1 detail per employee:
- late_count, late_minutes
- late_avg_minutes, late_max_minutes ← new
- early_count, early_minutes
- early_avg_minutes, early_max_minutes ← new
- ot_count, ot_minutes
- ot_avg_minutes, ot_max_minutes ← new
- missing_count
- last_attendance_date ← new
- consecutive_missing ← new (current streak of missing)
- attendance_rate ← new (% non-missing)
"""
from datetime import datetime as _dt, date as _date, timedelta as _td
q = db.query(AttendanceRecord)
if date_from:
try:
q = q.filter(AttendanceRecord.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(AttendanceRecord.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
rows = q.order_by(AttendanceRecord.date.asc()).all()
# Group rows by employee (preserve chronological order)
by_emp = {}
for r in rows:
if not r.employee_name:
continue
by_emp.setdefault(r.employee_name, []).append(r)
employee_stats = []
totals = {"late": 0, "early": 0, "ot": 0, "missing": 0}
for name, emp_rows in by_emp.items():
late_count = late_minutes = 0
late_min_list = []
early_count = early_minutes = 0
early_min_list = []
ot_count = ot_minutes = 0
ot_min_list = []
missing_count = 0
last_date = None
for r in emp_rows:
sc = (r.status_code or "").lower()
if "late" in sc:
late_count += 1
lm = r.late_minutes or 0
late_minutes += lm
late_min_list.append(lm)
totals["late"] += lm
if "early" in sc:
early_count += 1
em = r.early_minutes or 0
early_minutes += em
early_min_list.append(em)
totals["early"] += em
if "ot" in sc:
ot_count += 1
om = r.ot_minutes or 0
ot_minutes += om
ot_min_list.append(om)
totals["ot"] += om
if sc == "missing":
missing_count += 1
totals["missing"] += 1
if r.date and (last_date is None or r.date > last_date):
last_date = r.date
# Consecutive missing streak (current run ending at last row)
consecutive_missing = 0
for r in reversed(emp_rows):
if (r.status_code or "").lower() == "missing":
consecutive_missing += 1
else:
break
total_records = len(emp_rows)
attendance_rate = (
round((total_records - missing_count) / total_records * 100)
if total_records > 0 else 0
)
employee_stats.append({
"name": name,
"shift": emp_rows[-1].shift_code or "-",
# Late
"late_count": late_count,
"late_minutes": late_minutes,
"late_avg_minutes": round(late_minutes / late_count) if late_count > 0 else 0,
"late_max_minutes": max(late_min_list) if late_min_list else 0,
# Early
"early_count": early_count,
"early_minutes": early_minutes,
"early_avg_minutes": round(early_minutes / early_count) if early_count > 0 else 0,
"early_max_minutes": max(early_min_list) if early_min_list else 0,
# OT
"ot_count": ot_count,
"ot_minutes": ot_minutes,
"ot_avg_minutes": round(ot_minutes / ot_count) if ot_count > 0 else 0,
"ot_max_minutes": max(ot_min_list) if ot_min_list else 0,
# Missing
"missing_count": missing_count,
"consecutive_missing": consecutive_missing,
"last_attendance_date": last_date.isoformat() if last_date else None,
# Overall
"total_records": total_records,
"attendance_rate": attendance_rate,
})
employee_stats.sort(key=lambda x: x["late_minutes"], reverse=True)
return {"stats": employee_stats, **totals}
@app.get("/api/attendance/records")
async def attendance_records(
page: int = 1,
per_page: int = 50,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "desc",
search: Optional[str] = None,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
staff: Optional[str] = None,
status: Optional[str] = None,
status_text: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Paginated records from attendance_records (SQL)."""
from datetime import datetime as _dt
from sqlalchemy import or_, and_
q = db.query(AttendanceRecord)
if date_from:
try:
q = q.filter(AttendanceRecord.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(AttendanceRecord.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
if staff:
q = q.filter(AttendanceRecord.employee_name == staff)
if status:
q = q.filter(AttendanceRecord.status_code == status)
if status_text:
# Map status_text filter to actual conditions
if status_text == 'abnormal':
q = q.filter(
(AttendanceRecord.late_minutes > 30) |
(AttendanceRecord.early_minutes > 30) |
(AttendanceRecord.ot_minutes > 30)
)
elif status_text == 'late':
q = q.filter(
(AttendanceRecord.late_minutes > 0) &
(AttendanceRecord.late_minutes <= 30)
)
elif status_text == 'early':
q = q.filter(
(AttendanceRecord.early_minutes > 0) &
(AttendanceRecord.early_minutes <= 30)
)
elif status_text == 'ot':
q = q.filter(
(AttendanceRecord.ot_minutes > 0) &
(AttendanceRecord.ot_minutes <= 30)
)
elif status_text == 'missing':
q = q.filter(AttendanceRecord.status_code == 'missing')
elif status_text in ('holiday', 'al', 'sl', 'cl', 'mixed_leave'):
q = q.filter(AttendanceRecord.status_code == status_text)
elif status_text == 'leave':
# Any leave type — group filter
q = q.filter(AttendanceRecord.status_code.in_(['al', 'sl', 'cl', 'mixed_leave']))
elif status_text == 'normal':
q = q.filter(
(AttendanceRecord.late_minutes == 0) &
(AttendanceRecord.early_minutes == 0) &
(AttendanceRecord.ot_minutes == 0)
)
else:
# Fallback to text search
q = q.filter(AttendanceRecord.status_text.like(f"%{status_text}%"))
if search:
like = f"%{search}%"
q = q.filter(or_(
AttendanceRecord.employee_name.like(like),
AttendanceRecord.department.like(like),
AttendanceRecord.shift_code.like(like),
))
# sort
sortable = {
"date": AttendanceRecord.date,
"employee_name": AttendanceRecord.employee_name,
"weekday": AttendanceRecord.weekday,
"shift_code": AttendanceRecord.shift_code,
"status_code": AttendanceRecord.status_code,
}
col = sortable.get(sort_by, AttendanceRecord.date)
q = q.order_by(col.desc() if sort_order == "desc" else col.asc())
total = q.count()
rows = q.offset((page - 1) * per_page).limit(per_page).all()
enrich_attendance_with_leave(rows, db)
def to_dict(r):
return {
"id": r.id,
"staff": r.employee_name,
"employee_name": r.employee_name,
"company": r.company,
"department": r.department,
"date": r.date.isoformat() if r.date else None,
"weekday": r.weekday,
"shift": r.shift_code,
"shift_code": r.shift_code,
"status_code": r.status_code,
"status_text": r.status_text,
"expected_in": r.expected_in,
"expected_out": r.expected_out,
"actual_in": r.actual_in,
"actual_out": r.actual_out,
"late_minutes": r.late_minutes or 0,
"early_minutes": r.early_minutes or 0,
"ot_minutes": r.ot_minutes or 0,
"is_manually_edited": bool(r.is_manually_edited),
}
return {
"records": [to_dict(r) for r in rows],
"total": total,
"page": page,
"per_page": per_page,
}
@app.get("/api/attendance/{record_id}")
async def get_attendance_record(
record_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Get single attendance record by id from DB."""
r = db.query(AttendanceRecord).filter(AttendanceRecord.id == record_id).first()
if not r:
raise HTTPException(status_code=404, detail="Record not found")
return {
"id": r.id,
"employee_name": r.employee_name,
"company": r.company,
"department": r.department,
"date": r.date.isoformat() if r.date else None,
"weekday": r.weekday,
"shift_code": r.shift_code,
"check_in": r.check_in.isoformat() if r.check_in else None,
"check_out": r.check_out.isoformat() if r.check_out else None,
"status_code": r.status_code,
"status_text": r.status_text,
"late_minutes": r.late_minutes or 0,
"early_minutes": r.early_minutes or 0,
"ot_minutes": r.ot_minutes or 0,
"expected_in": r.expected_in,
"expected_out": r.expected_out,
"actual_in": r.actual_in,
"actual_out": r.actual_out,
"is_manually_edited": bool(r.is_manually_edited),
}
@app.put("/api/attendance/{record_id}")
async def update_attendance_record(
record_id: int,
payload: dict,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update attendance record. Marks is_manually_edited = True so re-import won't overwrite."""
from datetime import datetime as _dt
r = db.query(AttendanceRecord).filter(AttendanceRecord.id == record_id).first()
if not r:
raise HTTPException(status_code=404, detail="Record not found")
editable = {"status_code", "status_text", "late_minutes", "early_minutes",
"ot_minutes", "actual_in", "actual_out", "shift_code", "department"}
for k, v in payload.items():
if k in editable:
setattr(r, k, v)
r.is_manually_edited = True
r.last_edited_at = datetime.utcnow()
r.last_edited_by = current_user.id
db.commit()
logger.info("attendance record updated", extra={"context": {"record_id": record_id, "user_id": current_user.id}})
return {"message": "ok", "id": record_id}
# ============ Employees List (combined source) ============
@app.get("/api/employees/list")
async def list_employees(
source: Optional[str] = "all", # "all" / "attendance" / "leave"
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""List distinct employee names (real staff only).
Sources: attendance_records + leave_records (NOT accident - that table stores
submitter emails, not staff names).
Filter: names containing "@" (email format) are excluded automatically.
Used by frontend dropdowns to show ALL staff, even those who haven't taken leave yet.
"""
import re
EMAIL_RE = re.compile(r"@")
names = set()
if source in ("all", "attendance"):
for n in db.query(AttendanceRecord.employee_name).distinct().all():
if n[0] and not EMAIL_RE.search(n[0]):
names.add(n[0])
if source in ("all", "leave"):
for n in db.query(LeaveRecord.employee_name).distinct().all():
if n[0] and not EMAIL_RE.search(n[0]):
names.add(n[0])
# NOTE: Accident.submitter_email contains submitter EMAIL (e.g. info@scmedical.hk),
# NOT the injured staff name. We exclude accident from employee list to avoid
# polluting staff dropdown with email addresses.
sorted_names = sorted(names, key=lambda x: x.lower())
return {"employees": sorted_names, "count": len(sorted_names), "source": source, "excluded": "accident (email column)"}
# ============ Holidays CRUD ============# ============ Holidays CRUD ============
@app.get("/api/holidays", response_model=List[HolidayOut])
async def list_holidays(
year: Optional[int] = None,
region: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""List all holidays, optionally filtered by year / region."""
q = db.query(Holiday)
if year:
q = q.filter(sqlfunc.extract("year", Holiday.date) == year)
if region:
q = q.filter(Holiday.region == region)
return q.order_by(Holiday.date.asc()).all()
@app.get("/api/holidays/{holiday_id}", response_model=HolidayOut)
async def get_holiday(
holiday_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Get single holiday by id."""
h = db.query(Holiday).filter(Holiday.id == holiday_id).first()
if not h:
raise HTTPException(status_code=404, detail="Holiday not found")
return h
@app.post("/api/holidays", response_model=HolidayOut)
async def create_holiday(
body: HolidayCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin),
):
"""Create a new holiday (admin only)."""
existing = db.query(Holiday).filter(Holiday.date == body.date).first()
if existing:
raise HTTPException(status_code=409, detail=f"Holiday already exists for {body.date}")
h = Holiday(
date=body.date,
name=body.name,
name_en=body.name_en,
region=body.region or "HK",
is_mandatory=body.is_mandatory if body.is_mandatory is not None else True,
source="manual",
notes=body.notes,
)
db.add(h)
db.commit()
db.refresh(h)
logger.info("holiday created", extra={"context": {"id": h.id, "date": str(h.date)}})
return h
@app.put("/api/holidays/{holiday_id}", response_model=HolidayOut)
async def update_holiday(
holiday_id: int,
body: HolidayUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin),
):
"""Update an existing holiday (admin only)."""
h = db.query(Holiday).filter(Holiday.id == holiday_id).first()
if not h:
raise HTTPException(status_code=404, detail="Holiday not found")
update_data = body.dict(exclude_unset=True)
for k, v in update_data.items():
setattr(h, k, v)
h.source = "manual_edit"
h.last_edited_at = datetime.utcnow()
h.last_edited_by = current_user.id
db.commit()
db.refresh(h)
return h
@app.delete("/api/holidays/{holiday_id}")
async def delete_holiday(
holiday_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin),
):
"""Delete a holiday (admin only)."""
h = db.query(Holiday).filter(Holiday.id == holiday_id).first()
if not h:
raise HTTPException(status_code=404, detail="Holiday not found")
db.delete(h)
db.commit()
return {"deleted": holiday_id}
@app.post("/api/holidays/upload")
async def upload_holidays(
file: UploadFile = File(...),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin),
):
"""Bulk upload holidays from CSV or Excel.
CSV columns: date,name,name_en,is_mandatory,notes
Excel: same columns (row 1 = header)
Logic:
- Date exists + never edited (last_edited_at IS NULL) -> UPDATE
- Date exists + manually edited (last_edited_at NOT NULL) -> SKIP
- New date -> INSERT
"""
from io import BytesIO
content = await file.read()
fname = file.filename.lower()
rows = []
if fname.endswith(".csv"):
import csv, io as _io
reader = csv.DictReader(_io.StringIO(content.decode("utf-8-sig")))
for row in reader:
rows.append({
"date": row.get("date", "").strip(),
"name": row.get("name", "").strip(),
"name_en": row.get("name_en", "").strip() or None,
"is_mandatory": (row.get("is_mandatory", "true").strip().lower() in ("true", "1", "yes", "y")),
"notes": row.get("notes", "").strip() or None,
})
elif fname.endswith((".xlsx", ".xls")):
import openpyxl
wb = openpyxl.load_workbook(BytesIO(content), data_only=True)
ws = wb.active
headers = [str(c.value or "").strip() for c in ws[1]]
col_idx = {h: i for i, h in enumerate(headers)}
for row in ws.iter_rows(min_row=2, values_only=True):
if not row or all(v is None for v in row):
continue
def getv(h, default=None):
v = row[col_idx[h]] if h in col_idx and col_idx[h] < len(row) else None
return v if v is not None else default
rows.append({
"date": str(getv("date") or "").strip(),
"name": str(getv("name") or "").strip(),
"name_en": str(getv("name_en") or "").strip() or None,
"is_mandatory": bool(getv("is_mandatory", True)),
"notes": str(getv("notes") or "").strip() or None,
})
else:
raise HTTPException(status_code=400, detail="File must be CSV or Excel")
added = 0
updated = 0
skipped = 0
errors = []
for i, r in enumerate(rows):
try:
if not r["date"] or not r["name"]:
errors.append(f"Row {i+2}: missing date or name")
skipped += 1
continue
from datetime import datetime as _dt
if isinstance(r["date"], str):
d = _dt.strptime(r["date"], "%Y-%m-%d").date()
else:
d = r["date"]
existing = db.query(Holiday).filter(Holiday.date == d).first()
if existing:
if existing.last_edited_at is not None:
errors.append(f"Row {i+2}: {d} was edited at {existing.last_edited_at.isoformat()} - skipped")
skipped += 1
continue
# Not edited - safe to update
existing.name = r["name"]
existing.name_en = r["name_en"]
existing.is_mandatory = r["is_mandatory"]
existing.notes = r["notes"]
existing.source = "csv_upload"
updated += 1
continue
db.add(Holiday(
date=d,
name=r["name"],
name_en=r["name_en"],
region="HK",
is_mandatory=r["is_mandatory"],
source="csv_upload",
notes=r["notes"],
))
added += 1
except Exception as e:
errors.append(f"Row {i+2}: {str(e)}")
skipped += 1
db.commit()
return {
"added": added, "updated": updated, "skipped": skipped,
"errors": errors[:20],
"total_in_db": db.query(Holiday).count(),
}
@app.post("/api/holidays/refresh")
async def refresh_holidays_from_gov_hk(
years: Optional[str] = None, # CSV: "2025,2026,2027"
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin),
):
"""Refresh holidays from internal static list (sourced from gov.hk).
Admin endpoint. Adds new dates, updates existing ones (only if source=gov_hk).
Manually edited holidays are NOT overwritten.
"""
from parsers import get_all_holidays
target_years = [int(y) for y in years.split(",")] if years else [2025, 2026, 2027]
all_holidays = [h for h in get_all_holidays() if h[0].year in target_years]
added = 0
updated = 0
skipped = 0
for d, name_zh, name_en, is_mandatory in all_holidays:
existing = db.query(Holiday).filter(Holiday.date == d).first()
if existing:
# Don't overwrite manual edits
if existing.source and existing.source.startswith("manual"):
skipped += 1
continue
if (existing.name != name_zh or existing.name_en != name_en
or existing.is_mandatory != is_mandatory):
existing.name = name_zh
existing.name_en = name_en
existing.is_mandatory = is_mandatory
existing.source = "gov_hk"
updated += 1
else:
db.add(Holiday(
date=d, name=name_zh, name_en=name_en,
region="HK", is_mandatory=is_mandatory, source="gov_hk",
))
added += 1
db.commit()
return {
"added": added, "updated": updated, "skipped": skipped,
"total_in_db": db.query(Holiday).count(),
"years": target_years,
}
# ============ Leave Records CRUD ============
@app.get("/api/leave", response_model=List[LeaveOut])
async def list_leaves(
employee_name: Optional[str] = None,
leave_type: Optional[str] = None,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""List leave records with optional filters."""
from datetime import datetime as _dt
q = db.query(LeaveRecord)
if employee_name:
q = q.filter(LeaveRecord.employee_name == employee_name)
if leave_type:
q = q.filter(LeaveRecord.leave_type == leave_type.upper())
if date_from:
try:
q = q.filter(LeaveRecord.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(LeaveRecord.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
return q.order_by(LeaveRecord.date.desc(), LeaveRecord.employee_name.asc()).all()
@app.post("/api/leave", response_model=LeaveOut)
async def create_leave(
body: LeaveCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Create a leave record."""
body.leave_type = body.leave_type.upper()
if body.leave_type not in ("SL", "CL", "AL"):
raise HTTPException(status_code=400, detail="leave_type must be SL / CL / AL")
if body.hours < 0 or body.hours > 24:
raise HTTPException(status_code=400, detail="hours must be 0-24")
lv = LeaveRecord(
employee_name=body.employee_name,
employee_id=body.employee_id,
date=body.date,
leave_type=body.leave_type,
hours=body.hours,
reason=body.reason,
approved_by=body.approved_by,
source="manual",
notes=body.notes,
created_by=current_user.id,
)
db.add(lv)
try:
db.commit()
except Exception as e:
db.rollback()
if "UNIQUE" in str(e).upper() or "Duplicate" in str(e):
raise HTTPException(status_code=409, detail=f"Leave record exists for {body.employee_name} {body.date} {body.leave_type}")
raise
db.refresh(lv)
return lv
@app.put("/api/leave/{leave_id}", response_model=LeaveOut)
async def update_leave(
leave_id: int,
body: LeaveUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update a leave record."""
lv = db.query(LeaveRecord).filter(LeaveRecord.id == leave_id).first()
if not lv:
raise HTTPException(status_code=404, detail="Leave record not found")
update_data = body.dict(exclude_unset=True)
if "leave_type" in update_data and update_data["leave_type"]:
update_data["leave_type"] = update_data["leave_type"].upper()
if update_data["leave_type"] not in ("SL", "CL", "AL"):
raise HTTPException(status_code=400, detail="leave_type must be SL / CL / AL")
for k, v in update_data.items():
setattr(lv, k, v)
lv.last_edited_at = datetime.utcnow()
lv.last_edited_by = current_user.id
db.commit()
db.refresh(lv)
return lv
@app.delete("/api/leave/{leave_id}")
async def delete_leave(
leave_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Delete a leave record."""
lv = db.query(LeaveRecord).filter(LeaveRecord.id == leave_id).first()
if not lv:
raise HTTPException(status_code=404, detail="Leave record not found")
db.delete(lv)
db.commit()
return {"deleted": leave_id}
@app.post("/api/leave/upload")
async def upload_leaves(
file: UploadFile = File(...),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Bulk upload leave records from CSV or Excel.
CSV columns: employee_name,date,leave_type,hours,reason,approved_by
Excel: same columns (row 1 = header)
"""
from io import BytesIO
content = await file.read()
fname = file.filename.lower()
rows = []
if fname.endswith(".csv"):
import csv, io as _io
reader = csv.DictReader(_io.StringIO(content.decode("utf-8-sig")))
for row in reader:
rows.append({
"employee_name": row.get("employee_name", "").strip(),
"date": row.get("date", "").strip(),
"leave_type": row.get("leave_type", "").strip().upper(),
"hours": float(row.get("hours", 0) or 0),
"reason": row.get("reason", "").strip() or None,
"approved_by": row.get("approved_by", "").strip() or None,
})
elif fname.endswith((".xlsx", ".xls")):
import openpyxl
wb = openpyxl.load_workbook(BytesIO(content), data_only=True)
ws = wb.active
headers = [str(c.value or "").strip() for c in ws[1]]
col_idx = {h: i for i, h in enumerate(headers)}
for row in ws.iter_rows(min_row=2, values_only=True):
if not row or all(v is None for v in row):
continue
def getv(h):
v = row[col_idx[h]] if h in col_idx and col_idx[h] < len(row) else None
return v
rows.append({
"employee_name": str(getv("employee_name") or "").strip(),
"date": str(getv("date") or "").strip(),
"leave_type": str(getv("leave_type") or "").strip().upper(),
"hours": float(getv("hours") or 0),
"reason": str(getv("reason") or "").strip() or None,
"approved_by": str(getv("approved_by") or "").strip() or None,
})
else:
raise HTTPException(status_code=400, detail="File must be CSV or Excel")
added = 0
updated = 0
skipped = 0
errors = []
for i, r in enumerate(rows):
try:
if not r["employee_name"] or not r["date"] or not r["leave_type"]:
errors.append(f"Row {i+2}: missing required field")
skipped += 1
continue
if r["leave_type"] not in ("SL", "CL", "AL"):
errors.append(f"Row {i+2}: invalid leave_type")
skipped += 1
continue
from datetime import datetime as _dt
if isinstance(r["date"], str):
d = _dt.strptime(r["date"], "%Y-%m-%d").date()
else:
d = r["date"]
existing = db.query(LeaveRecord).filter(
LeaveRecord.employee_name == r["employee_name"],
LeaveRecord.date == d,
LeaveRecord.leave_type == r["leave_type"],
).first()
if existing:
if existing.last_edited_at is not None:
# Was manually edited - protect from overwrite
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 (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"
updated += 1
continue
db.add(LeaveRecord(
employee_name=r["employee_name"],
date=d,
leave_type=r["leave_type"],
hours=r["hours"],
reason=r["reason"],
approved_by=r["approved_by"],
source="csv_upload",
created_by=current_user.id,
))
added += 1
except Exception as e:
errors.append(f"Row {i+2}: {str(e)}")
skipped += 1
db.commit()
return {
"added": added, "updated": updated, "skipped": skipped,
"errors": errors[:20], # First 20 errors only
"total_in_db": db.query(LeaveRecord).count(),
}
# =============================================================================
# Backup / Restore + Version
# =============================================================================
import shutil, hashlib
from datetime import datetime as _dt2
BACKUP_DIR = Path("/app/data/backups")
DB_PATH = "/app/data/aars.db"
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.delete("/api/admin/backup/{filename}")
async def delete_backup(
filename: str,
current_user: User = Depends(get_current_admin),
):
"""Delete a backup file."""
fp = BACKUP_DIR / filename
if not fp.exists() or ".." in filename:
raise HTTPException(status_code=404, detail="Backup not found")
fp.unlink()
return {"deleted": filename}
@app.post("/api/admin/reset/{section}")
async def reset_section(
section: str,
current_user: User = Depends(get_current_admin),
db: Session = Depends(get_db),
):
"""Reset (truncate) a specific data section. Creates auto-backup first."""
single_models = {"attendance": AttendanceRecord, "accident": Accident}
if section not in single_models and section not in ("holidays", "leaves"):
raise HTTPException(status_code=400, detail=f"Unknown section: {section}")
# Auto-backup before reset
src = Path(DB_PATH)
auto_dst = None
if src.exists():
auto_dst = _backup_path(f"auto_pre_{section}_reset")
shutil.copy2(src, auto_dst)
# Truncate table(s)
if section == "holidays":
db.execute(Holiday.__table__.delete())
db.execute(LeaveRecord.__table__.delete())
elif section == "leaves":
db.execute(LeaveRecord.__table__.delete())
else:
db.execute(single_models[section].__table__.delete())
db.commit()
return {"reset": section, "auto_backup": auto_dst.name if auto_dst 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.2.0",
"commit": commit,
"date": date,
}
@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": []
}
async def _import_accident_to_db(db: Session, headers, rows, user_id: int) -> dict:
"""Parse Excel rows and upsert into accidents table.
Excel (Google Form) → SQL column map (15 columns):
Col 1 時間戳記 -> submitted_at
Col 2 電子郵件地址 -> employee_name
Col 3 部門名稱 -> department
Col 4 當日事件摘要 -> description
Col 5 緊急程度 -> severity
Col 6 事件發生時間 -> time
Col 7 涉及部門 -> location
Col 8 涉及的患者或相關人員 -> patient_or_staff
Col 9 事件描述 -> long_description
Col 10 如有事件相關照片可提供-> incident_photos
Col 11 處理情況/已採取的行動-> action_taken
Col 12 需要管理層介入的事項 -> needs_escalation
Col 13 第 12 欄 -> (skipped)
Col 14 分數 -> incident_score
Col 15 事件發生日期 -> date
"""
from datetime import datetime, date, time as _time
def _parse_dt(v):
if v is None:
return None
if isinstance(v, datetime):
return v
if isinstance(v, str):
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y/%m/%d %H:%M:%S"):
try:
return datetime.strptime(v, fmt)
except ValueError:
continue
return None
def _parse_date(v):
if v is None:
return None
if isinstance(v, datetime):
return v.date()
if isinstance(v, date):
return v
if isinstance(v, str):
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%d/%m/%Y", "%m/%d/%Y"):
try:
return datetime.strptime(v, fmt).date()
except ValueError:
continue
return None
def _parse_time(v):
if v is None:
return None
if isinstance(v, _time):
return v.strftime("%H:%M:%S")
if isinstance(v, datetime):
return v.strftime("%H:%M:%S")
if isinstance(v, str):
for fmt in ("%H:%M:%S", "%H:%M"):
try:
return datetime.strptime(v, fmt).strftime("%H:%M:%S")
except ValueError:
continue
return str(v) if v else None
def _cell(row, idx):
if idx is None or idx >= len(row):
return None
return row[idx]
imported = updated = skipped = errors = 0
for row_idx, row in enumerate(rows, start=2): # row_idx is the Excel 1-based row number
try:
submitted_at = _parse_dt(_cell(row, 0))
submitter_email = _cell(row, 1) or ""
department = _cell(row, 2)
description = _cell(row, 3) or ""
severity = str(_cell(row, 4) or "")
time_str = _parse_time(_cell(row, 5))
location = _cell(row, 6) or ""
patient_or_staff = _cell(row, 7)
long_description = _cell(row, 8)
incident_photos = _cell(row, 9)
action_taken = _cell(row, 10)
needs_escalation = _cell(row, 11)
incident_score = _cell(row, 13)
date_val = _parse_date(_cell(row, 14))
if not date_val and submitted_at:
# Fallback: derive date from submitted timestamp (col 1)
date_val = submitted_at.date()
# Skip rows missing critical identifiers
if not date_val or not submitter_email:
errors += 1
continue
if incident_score is not None and incident_score != "":
try:
incident_score = int(incident_score)
except (TypeError, ValueError):
incident_score = None
existing = db.query(Accident).filter(Accident.excel_row == row_idx).first()
if existing:
if existing.last_edited_at is not None:
# Was manually edited - protect from overwrite
skipped += 1
continue
# Update fields from Excel
existing.submitted_at = submitted_at
existing.submitter_email = submitter_email
existing.department = department
existing.description = description
existing.severity = severity
existing.time = time_str
existing.location = location
existing.patient_or_staff = patient_or_staff
existing.long_description = long_description
existing.incident_photos = incident_photos
existing.action_taken = action_taken
existing.needs_escalation = str(needs_escalation) if needs_escalation is not None else None
existing.incident_score = incident_score
existing.date = date_val
updated += 1
else:
rec = Accident(
submitted_at=submitted_at,
submitter_email=submitter_email,
department=department,
description=description,
severity=severity,
time=time_str,
location=location,
patient_or_staff=patient_or_staff,
long_description=long_description,
incident_photos=incident_photos,
action_taken=action_taken,
needs_escalation=str(needs_escalation) if needs_escalation is not None else None,
incident_score=incident_score,
date=date_val,
excel_row=row_idx,
created_by=user_id,
)
db.add(rec)
imported += 1
except Exception as e:
errors += 1
logger.warning("accident row import error",
extra={"context": {"row": row_idx, "error": str(e)}})
db.commit()
return {"imported": imported, "updated": updated, "skipped": skipped, "errors": errors}
def _get_excel_headers_accident() -> dict:
"""Read /app/data/accident.xlsx row 1 and return SQL col → Excel header map.
Cached after first call.
"""
if hasattr(_get_excel_headers_accident, "_cache"):
return _get_excel_headers_accident._cache
try:
import openpyxl
path = os.path.join(DATA_DIR, "accident.xlsx")
if not os.path.exists(path):
_get_excel_headers_accident._cache = {}
return {}
wb = openpyxl.load_workbook(path, data_only=True)
ws = wb.active
headers = [ws.cell(row=1, column=c).value for c in range(1, ws.max_column + 1)]
# Hardcoded column index → SQL field mapping (matches _import_accident_to_db)
idx_to_sql = {
0: "submitted_at",
1: "submitter_email", # was employee_name, renamed 2026-07-20
2: "department",
3: "description",
4: "severity",
5: "time",
6: "location",
7: "patient_or_staff",
8: "long_description",
9: "incident_photos",
10: "action_taken",
11: "needs_escalation",
13: "incident_score",
14: "date",
}
result = {}
for i, h in enumerate(headers):
if i in idx_to_sql and h is not None:
result[idx_to_sql[i]] = str(h)
_get_excel_headers_accident._cache = result
return result
except Exception as e:
logger.warning("read accident excel headers failed", extra={"context": {"error": str(e)}})
_get_excel_headers_accident._cache = {}
return {}
def _get_excel_column_order_accident() -> list:
"""Return SQL columns in Excel order (Col 1..15 mapped to SQL fields).
Plus extra DB-only columns (responsible_person, medical_report) at the end
so they can be VLOOKUP'd from 個案查詢 sheet even though they are not in
the original Excel.
"""
return [
"id", # virtual
"submitted_at",
"submitter_email", # was employee_name, renamed 2026-07-20
"department",
"description",
"severity",
"time",
"incident_datetime", # combined "YYYY-MM-DD HH:MM:SS" (computed)
"location",
"patient_or_staff",
"long_description",
"incident_photos",
"action_taken",
"needs_escalation",
"incident_score",
"date",
# DB-only (not in Excel)
"injured_person", # was responsible_person, renamed 2026-07-19
"medical_report",
]
@app.get("/api/accident/dashboard/summary")
async def accident_dashboard_summary(
date_from: Optional[str] = None,
date_to: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Accident dashboard summary stats."""
from datetime import date as _date, datetime as _dt
q = db.query(Accident).filter(Accident.deleted_at.is_(None))
if date_from:
try:
q = q.filter(Accident.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(Accident.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
rows = q.all()
today = _date.today()
month_start = today.replace(day=1)
total = len(rows)
by_severity = {}
by_department = {}
by_location = {}
by_injured = {}
this_month = 0
earliest_date = None
latest_date = None
for r in rows:
sc = r.severity or "unknown"
by_severity[sc] = by_severity.get(sc, 0) + 1
if r.department:
by_department[r.department] = by_department.get(r.department, 0) + 1
if r.location:
by_location[r.location] = by_location.get(r.location, 0) + 1
if r.injured_person:
by_injured[r.injured_person] = by_injured.get(r.injured_person, 0) + 1
if r.date and r.date >= month_start:
this_month += 1
if r.date:
if earliest_date is None or r.date < earliest_date:
earliest_date = r.date
if latest_date is None or r.date > latest_date:
latest_date = r.date
# Avg per day (over date span)
days_span = 1
if earliest_date and latest_date:
days_span = max((latest_date - earliest_date).days + 1, 1)
avg_per_day = round(total / days_span, 2) if total > 0 else 0
return {
"total_records": total,
"this_month": this_month,
"today": sum(1 for r in rows if r.date == today),
"by_severity": by_severity,
"by_department": by_department,
"by_location": by_location,
"by_injured": by_injured,
"severity_count": { # frontend-expected
"1": by_severity.get("1", 0),
"2": by_severity.get("2", 0),
"3": by_severity.get("3", 0),
"4": by_severity.get("4", 0),
"5": by_severity.get("5", 0),
},
"unique_locations": len(by_location),
"unique_departments": len(by_department),
"unique_responsible": len(by_injured),
"avg_per_day": avg_per_day,
"earliest_date": earliest_date.isoformat() if earliest_date else None,
"latest_date": latest_date.isoformat() if latest_date else None,
}
@app.get("/api/accident/dashboard/stats")
async def accident_dashboard_stats(
date_from: Optional[str] = None,
date_to: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Accident per-entity leaderboards (location, employee, department, responsible)."""
from datetime import datetime as _dt
q = db.query(Accident).filter(Accident.deleted_at.is_(None))
if date_from:
try:
q = q.filter(Accident.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(Accident.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
rows = q.all()
# Bucket by entity
by_location = {}
by_employee = {}
by_department = {}
by_injured = {}
for r in rows:
# Severity-weighted score (severity 5 = most critical)
try:
sev_weight = int(r.severity or 1)
except (ValueError, TypeError):
sev_weight = 1
sev_weight = max(sev_weight, 1)
loc = r.location or "未分類"
by_location.setdefault(loc, {"name": loc, "count": 0, "severity_sum": 0, "avg_severity": 0})
by_location[loc]["count"] += 1
by_location[loc]["severity_sum"] += sev_weight
emp = r.submitter_email or "未分類"
by_employee.setdefault(emp, {"name": emp, "count": 0, "severity_sum": 0, "avg_severity": 0})
by_employee[emp]["count"] += 1
by_employee[emp]["severity_sum"] += sev_weight
dept = r.department or "未分類"
by_department.setdefault(dept, {"name": dept, "count": 0, "severity_sum": 0, "avg_severity": 0})
by_department[dept]["count"] += 1
by_department[dept]["severity_sum"] += sev_weight
rp = r.injured_person or "未分類"
by_injured.setdefault(rp, {"name": rp, "count": 0, "severity_sum": 0, "avg_severity": 0})
by_injured[rp]["count"] += 1
by_injured[rp]["severity_sum"] += sev_weight
def finalize(d):
out = []
for v in d.values():
v["avg_severity"] = round(v["severity_sum"] / v["count"], 2) if v["count"] > 0 else 0
out.append(v)
return out
return {
"by_location": sorted(finalize(by_location), key=lambda x: x["count"], reverse=True),
"by_employee": sorted(finalize(by_employee), key=lambda x: x["count"], reverse=True),
"by_department": sorted(finalize(by_department), key=lambda x: x["count"], reverse=True),
"by_injured": sorted(finalize(by_injured), key=lambda x: x["count"], reverse=True),
}
@app.get("/api/accident/records")
async def accident_records(
page: int = 1,
per_page: int = 50,
sort_by: Optional[str] = None,
sort_order: Optional[str] = "desc",
search: Optional[str] = None,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
severity: Optional[str] = None,
location: Optional[str] = None,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Paginated accident records from SQL."""
from datetime import datetime as _dt
from sqlalchemy import or_
q = db.query(Accident).filter(Accident.deleted_at.is_(None))
if date_from:
try:
q = q.filter(Accident.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(Accident.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
if severity:
q = q.filter(Accident.severity == severity)
if location:
q = q.filter(Accident.location == location)
if search:
like = f"%{search}%"
q = q.filter(or_(
Accident.submitter_email.like(like),
Accident.location.like(like),
Accident.description.like(like),
Accident.department.like(like),
))
sortable = {
"id": Accident.id,
"submitted_at": Accident.submitted_at,
"date": Accident.date,
"severity": Accident.severity,
"time": Accident.time,
"location": Accident.location,
"employee_name": Accident.submitter_email,
"department": Accident.department,
"description": Accident.description,
"action_taken": Accident.action_taken,
"patient_or_staff": Accident.patient_or_staff,
"incident_score": Accident.incident_score,
"responsible_person": Accident.injured_person,
}
col = sortable.get(sort_by, Accident.date)
q = q.order_by(col.desc() if sort_order == "desc" else col.asc())
total = q.count()
rows = q.offset((page - 1) * per_page).limit(per_page).all()
def to_dict(r):
# Compose incident_datetime from date + time (display as "YYYY-MM-DD HH:MM:SS")
incident_dt = None
if r.date:
date_part = r.date.isoformat()
time_part = r.time or "00:00:00"
# Ensure time has HH:MM:SS format (may be HH:MM from Excel import)
if len(time_part) == 5: # "HH:MM"
time_part += ":00"
incident_dt = f"{date_part} {time_part}"
return {
"id": r.id,
# Submitted time formatted as "YYYY-MM-DD HH:MM:SS" (display-friendly)
"submitted_at": r.submitted_at.strftime("%Y-%m-%d %H:%M:%S") if r.submitted_at else None,
"date": r.date.isoformat() if r.date else None,
"time": r.time,
"incident_datetime": incident_dt, # YYYY-MM-DD HH:MM:SS combined
"location": r.location,
# Backend alias: frontend column_order uses submitter_email as key
# but legacy UI expected employee_name
"submitter_email": r.submitter_email,
"employee_name": r.submitter_email, # alias for backward compat
"injured_person_id": r.injured_person_id,
"employee_id": r.injured_person_id, # alias
"department": r.department,
"description": r.description,
"severity": r.severity,
"medical_report": r.medical_report,
"action_taken": r.action_taken,
"responsible_person": r.injured_person,
"patient_or_staff": r.patient_or_staff,
"long_description": r.long_description,
"incident_photos": r.incident_photos,
"needs_escalation": r.needs_escalation,
"incident_score": r.incident_score,
"excel_row": r.excel_row,
"created_at": r.created_at.isoformat() if r.created_at else None,
}
excel_headers = _get_excel_headers_accident()
column_order = _get_excel_column_order_accident()
return {
"records": [to_dict(r) for r in rows],
"total": total,
"page": page,
"per_page": per_page,
"excel_headers": excel_headers,
"column_order": column_order,
}
@app.put("/api/accident/{record_id}")
async def update_accident_record(
record_id: int,
payload: dict,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update accident record. Sets last_edited_at so re-import won't overwrite.
Marks is_locked for explicit protection (UI optional).
"""
r = db.query(Accident).filter(Accident.id == record_id).first()
if not r:
raise HTTPException(status_code=404, detail="Record not found")
# Allowed fields to edit (whitelist for safety)
editable = {
"submitter_email", # was employee_name, renamed 2026-07-20
"injured_person_id", # was employee_id, renamed 2026-07-20
"injured_person", # was responsible_person, renamed 2026-07-19
"department", "description", "severity",
"time", "location", "patient_or_staff", "long_description",
"incident_photos", "action_taken", "needs_escalation",
"incident_score", "medical_report",
}
for k, v in payload.items():
if k in editable:
setattr(r, k, v)
r.last_edited_at = datetime.utcnow()
r.last_edited_by = current_user.id
db.commit()
logger.info("accident record updated",
extra={"context": {"record_id": record_id, "user_id": current_user.id}})
return {"message": "ok", "id": record_id}
@app.get("/api/accident/{record_id}")
async def get_accident_record(
record_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Get a single accident record by ID."""
r = db.query(Accident).filter(Accident.id == record_id, Accident.deleted_at.is_(None)).first()
if not r:
raise HTTPException(status_code=404, detail="Record not found")
excel_headers = _get_excel_headers_accident()
column_order = _get_excel_column_order_accident()
return {
"id": r.id,
"submitted_at": r.submitted_at.isoformat() if r.submitted_at else None,
"date": r.date.isoformat() if r.date else None,
"time": r.time,
"location": r.location,
"employee_name": r.submitter_email,
"employee_id": r.injured_person_id,
"department": r.department,
"description": r.description,
"severity": r.severity,
"medical_report": r.medical_report,
"action_taken": r.action_taken,
"responsible_person": r.injured_person,
"patient_or_staff": r.patient_or_staff,
"long_description": r.long_description,
"incident_photos": r.incident_photos,
"needs_escalation": r.needs_escalation,
"incident_score": r.incident_score,
"excel_row": r.excel_row,
"excel_headers": excel_headers,
"column_order": column_order,
"created_at": r.created_at.isoformat() if r.created_at else None,
}
@app.get("/api/accident/export/excel")
async def accident_export_excel(
date_from: Optional[str] = None,
date_to: Optional[str] = None,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Generate multi-sheet accident xlsx (Phase 2: formula-driven, 3 sheets).
Sheets:
- 總覽: KPIs, severity/dept distribution, case index
- 個案查詢: input case_no in C2; VLOOKUP shows detail (live, no macro needed)
+ multi-case list
+ 2-case side-by-side compare (input cells)
- 明細: full table, all Excel columns, conditional formatting on severity
"""
import xlsxwriter
import io
from datetime import datetime as _dt
from collections import Counter
q = db.query(Accident).filter(Accident.deleted_at.is_(None))
if date_from:
try:
q = q.filter(Accident.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(Accident.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
rows = q.order_by(Accident.date.desc()).all()
excel_headers = _get_excel_headers_accident()
column_order = _get_excel_column_order_accident()
# Map col index (1-based, for VLOOKUP) → SQL field key
# 明細 sheet will write records starting at row 2 (header row 1).
# Column order in 明細: id, submitted_at, employee_name, department, description,
# severity, time, location, patient_or_staff, long_description, incident_photos,
# action_taken, needs_escalation, incident_score, date
col_index_lookup = {col: i + 1 for i, col in enumerate(column_order)}
# ============ Workbook + formats ============
buf = io.BytesIO()
# force_full_calc_on_load + calc_on_load ensures Excel/Numbers recompute all
# formulas when opening the file (overrides cached 0 from xlsxwriter).
wb = xlsxwriter.Workbook(
buf,
{"in_memory": True,
"calc_on_load": True,
"calc_mode": "auto"})
# Belt-and-suspenders: set workbook calc properties via set_calc_mode
wb.set_calc_mode("auto", calc_id=1)
title_fmt = wb.add_format({"bold": True, "font_size": 16, "font_color": "#1E3A8A"})
subtitle_fmt = wb.add_format({"italic": True, "font_size": 10, "font_color": "#6B7280"})
section_fmt = wb.add_format({"bold": True, "font_size": 12, "font_color": "#1E3A8A",
"bottom": 2, "bottom_color": "#93C5FD"})
kpi_label_fmt = wb.add_format({"bold": True, "font_size": 9, "font_color": "#6B7280",
"align": "left"})
kpi_value_fmt = wb.add_format({"bold": True, "font_size": 22, "font_color": "#1E3A8A",
"align": "left"})
header_fmt = wb.add_format({"font_color": "#FFFFFF", "bold": True, "align": "center",
"bg_color": "#1E40AF", "border": 1, "border_color": "#1E3A8A"})
cell_fmt = wb.add_format({"align": "left", "valign": "top"})
cell_wrap_fmt = wb.add_format({"text_wrap": True, "align": "left", "valign": "top"})
detail_label_fmt = wb.add_format({"bold": True, "bg_color": "#F1F5F9",
"border": 1, "border_color": "#CBD5E1",
"valign": "top", "text_wrap": True, "align": "left",
"indent": 1})
detail_value_fmt = wb.add_format({"border": 1, "border_color": "#CBD5E1",
"valign": "top", "text_wrap": True, "align": "left",
"indent": 1})
detail_label_alt_fmt = wb.add_format({"bold": True, "bg_color": "#E2E8F0",
"border": 1, "border_color": "#CBD5E1",
"valign": "top", "text_wrap": True, "align": "left",
"indent": 1})
severity_fmts = {}
for code, font_c, bg_c in [
("1", "#065F46", "#D1FAE5"), ("2", "#1E40AF", "#DBEAFE"),
("3", "#92400E", "#FEF3C7"), ("4", "#9A3412", "#FED7AA"),
("5", "#7F1D1D", "#FEE2E2"),
]:
severity_fmts[code] = wb.add_format({"font_color": font_c, "bg_color": bg_c,
"bold": True, "align": "center", "border": 1,
"border_color": "#E5E7EB"})
severity_fmts["default"] = wb.add_format({"align": "center", "border": 1,
"border_color": "#E5E7EB"})
# ============ Sheet 1: 總覽 ============
overview = wb.add_worksheet("總覽")
overview.set_tab_color("#1E40AF")
overview.hide_gridlines(2)
overview.set_column("A:A", 2)
overview.set_column("B:B", 30)
overview.set_column("C:C", 18)
overview.set_column("D:D", 30)
overview.set_column("E:E", 18)
overview.set_column("F:F", 30)
overview.set_column("G:G", 18)
overview.write("B2", "意外記錄 總覽 Dashboard", title_fmt)
overview.write("B3", f"生成時間: {_dt.now().strftime('%Y-%m-%d %H:%M:%S')}", subtitle_fmt)
if date_from or date_to:
range_str = f"{date_from or '起始'} → {date_to or '現在'}"
else:
range_str = "全部記錄"
overview.write("B4", f"篩選範圍: {range_str}", subtitle_fmt)
overview.merge_range("B6:C6", "總記錄數", kpi_label_fmt)
overview.merge_range("B7:C7", len(rows), kpi_value_fmt)
overview.merge_range("D6:E6", "嚴重程度 5 (最高)", kpi_label_fmt)
sev5 = sum(1 for r in rows if str(r.severity or "") == "5")
overview.merge_range("D7:E7", sev5, kpi_value_fmt)
overview.merge_range("F6:G6", "最早事件日期", kpi_label_fmt)
earliest = min((r.date for r in rows if r.date), default=None)
overview.merge_range("F7:G7", earliest.isoformat() if earliest else "-", kpi_value_fmt)
overview.write("B10", "嚴重程度分佈", section_fmt)
overview.write("B11", "Severity", header_fmt)
overview.write("C11", "數量", header_fmt)
overview.write("D11", "百分比", header_fmt)
severity_codes = ["1", "2", "3", "4", "5"]
for i, code in enumerate(severity_codes):
overview.write(11 + i, 1, code, severity_fmts[code])
count = sum(1 for r in rows if str(r.severity or "") == code)
overview.write(11 + i, 2, count, cell_fmt)
pct = (count / len(rows) * 100) if rows else 0
overview.write(11 + i, 3, f"{pct:.1f}%", cell_fmt)
overview.write(11 + len(severity_codes), 1, "(空)", cell_fmt)
unk = sum(1 for r in rows if not str(r.severity or ""))
overview.write(11 + len(severity_codes), 2, unk, cell_fmt)
pct_unk = (unk / len(rows) * 100) if rows else 0
overview.write(11 + len(severity_codes), 3, f"{pct_unk:.1f}%", cell_fmt)
overview.write("F10", "部門事故數 (TOP 10)", section_fmt)
overview.write("F11", "部門", header_fmt)
overview.write("G11", "數量", header_fmt)
dept_counts = Counter((r.department or "(未分類)") for r in rows)
top_depts = dept_counts.most_common(10)
for i, (dept, count) in enumerate(top_depts):
overview.write(11 + i, 5, dept, cell_fmt)
overview.write(11 + i, 6, count, cell_fmt)
nav_start_row = 80
overview.write(f"B{nav_start_row}", "個案索引 (可 click → 跳去「個案查詢」對應 row)", section_fmt)
overview.write(f"B{nav_start_row + 1}", "#", header_fmt)
overview.write(f"C{nav_start_row + 1}", "日期", header_fmt)
overview.write(f"D{nav_start_row + 1}", "時間", header_fmt)
overview.write(f"E{nav_start_row + 1}", "部門", header_fmt)
overview.write(f"F{nav_start_row + 1}", "嚴重程度", header_fmt)
overview.write(f"G{nav_start_row + 1}", "→ 查個案", header_fmt)
link_fmt = wb.add_format({"font_color": "#2563EB", "underline": 1})
for idx, r in enumerate(rows):
row_n = nav_start_row + 2 + idx
overview.write(row_n, 1, r.id, cell_fmt)
overview.write(row_n, 2, r.date.isoformat() if r.date else "", cell_fmt)
overview.write(row_n, 3, r.time or "", cell_fmt)
overview.write(row_n, 4, r.department or "", cell_fmt)
overview.write(row_n, 5, str(r.severity or ""), severity_fmts.get(str(r.severity or ""), severity_fmts["default"]))
# Use HYPERLINK formula. When clicked, jumps to sheet 個案查詢 cell A1.
# User types case_no in C5 of 個案查詢 sheet.
# xlsxwriter requires single quotes around sheet names containing non-ASCII.
# Use proper xlsxwriter hyperlink element (write_url with internal:Sheet!Cell).
# HYPERLINK formula would cache value 0 - write_url sets display text directly.
overview.write_url(row_n, 6,
f"internal:個案查詢!A1",
link_fmt,
string=f"#{r.id} → 查詢",
tip=f"跳到「個案查詢」輸入個案 ID {r.id}")
overview.set_row(row_n, 18)
# ============ Charts block - 統計圖表 (Sheet 1 總覽 右側 columns I-N) ============
# 4 charts: Severity Pie, Severity Column, Department Bar TOP 10, Monthly Trend Line
# Position: I column (right of KPI + 個案索引 blocks), so no overlap with case index.
# --- Chart 1: Severity Pie (I6) ---
sev_pie = wb.add_chart({"type": "pie"})
sev_pie.add_series({
"name": "嚴重程度分佈",
# B12:B17 = severity codes + (空)
"categories": ["總覽", 11, 1, 16, 1],
"values": ["總覽", 11, 2, 16, 2],
"data_labels": {"percentage": True, "category": True, "position": "outside_end"},
})
sev_pie.set_title({"name": "嚴重程度分佈 (Pie)"})
sev_pie.set_style(10)
sev_pie.set_size({"width": 480, "height": 320})
sev_pie.set_legend({"position": "right"})
overview.insert_chart("I6", sev_pie, {"x_offset": 5, "y_offset": 5})
# --- Chart 2: Department Bar TOP 10 (I22) ---
# Dept table starts at F12, up to 10 entries + header row.
# F12:F21 = department names; G12:G21 = counts
dept_bar = wb.add_chart({"type": "bar"})
dept_bar.add_series({
"name": "部門事故數",
"categories": ["總覽", 11, 5, 20, 5],
"values": ["總覽", 11, 6, 20, 6],
"fill": {"color": "#3B82F6"},
"border": {"color": "#1E40AF"},
"data_labels": {"value": True},
})
dept_bar.set_title({"name": "部門事故數 (TOP 10)"})
dept_bar.set_x_axis({"name": "事故數"})
dept_bar.set_y_axis({"name": "部門", "reverse": True})
dept_bar.set_legend({"none": True})
dept_bar.set_size({"width": 480, "height": 320})
overview.insert_chart("I22", dept_bar, {"x_offset": 5, "y_offset": 5})
# --- Chart 3: Severity Column (I42) ---
sev_col_chart = wb.add_chart({"type": "column"})
sev_col_chart.add_series({
"name": "嚴重程度 數量",
"categories": ["總覽", 11, 1, 15, 1], # B12:B16 = severity 1-5 only (skip (空))
"values": ["總覽", 11, 2, 15, 2],
"fill": {"color": "#EF4444"},
"border": {"color": "#7F1D1D"},
"data_labels": {"value": True},
})
sev_col_chart.set_title({"name": "嚴重程度 數量 (Column)"})
sev_col_chart.set_x_axis({"name": "嚴重程度"})
sev_col_chart.set_y_axis({"name": "事故數"})
sev_col_chart.set_legend({"none": True})
sev_col_chart.set_size({"width": 480, "height": 320})
overview.insert_chart("I42", sev_col_chart, {"x_offset": 5, "y_offset": 5})
# --- Chart 4: Monthly Trend Line (I62) ---
# Build monthly data table inline below the case index (col B, starting at row 60).
# Header at row 60; data rows 61..72 for last 12 months.
trend_start_row = 60 # 0-indexed → Excel row 61
overview.write(trend_start_row, 1, "月份", header_fmt)
overview.write(trend_start_row, 2, "事故數", header_fmt)
# Compute last 12 months from data
from collections import OrderedDict
from datetime import date as _date
today = _dt.now().date()
months = []
for offset in range(11, -1, -1):
y = today.year
m = today.month - offset
while m <= 0:
m += 12
y -= 1
months.append(f"{y}-{m:02d}")
monthly_counts = OrderedDict((m, 0) for m in months)
for r in rows:
if r.date:
key = f"{r.date.year}-{r.date.month:02d}"
if key in monthly_counts:
monthly_counts[key] += 1
for i, (m, cnt) in enumerate(monthly_counts.items()):
overview.write(trend_start_row + 1 + i, 1, m, cell_fmt)
overview.write(trend_start_row + 1 + i, 2, cnt, cell_fmt)
n_months = len(months)
trend_end_row = trend_start_row + n_months # exclusive
line_chart = wb.add_chart({"type": "line"})
line_chart.add_series({
"name": "每月事故數",
"categories": ["總覽", trend_start_row + 1, 1, trend_end_row, 1],
"values": ["總覽", trend_start_row + 1, 2, trend_end_row, 2],
"line": {"color": "#10B981", "width": 2.25},
"marker": {"type": "circle", "size": 6,
"fill": {"color": "#10B981"},
"border": {"color": "#065F46"}},
"data_labels": {"value": True},
})
line_chart.set_title({"name": "每月事故趨勢 (Last 12 months)"})
line_chart.set_x_axis({"name": "月份"})
line_chart.set_y_axis({"name": "事故數"})
line_chart.set_legend({"none": True})
line_chart.set_size({"width": 480, "height": 320})
overview.insert_chart("I62", line_chart, {"x_offset": 5, "y_offset": 5})
# ============ Sheet 3 (built FIRST so detail-VLOOKUP can reference it): 明細 ============
# Build it now so that 個案查詢 formulas can resolve its range properly.
# However, xlsxwriter's defined_name allows forward reference; formula text is
# what matters at runtime - even if the sheet is added later, the formula string
# '明細!A:O' will be valid when Excel opens the file.
# We define 明細 here so that the formula range 明細!A2:O{max_row} resolves.
# But xlsxwriter requires sheets to be added in the order they're defined.
# Solution: build 個案查詢 first (it sets the VLOOKUP formula referencing 明細!A:O),
# then create 明細 LAST.
# Per xlsxwriter semantics, sheet order is the add_worksheet() call order.
# So: Sheet order on tab will be: 總覽 (already added), 個案查詢 (next), 明細 (last).
# The 明細 range in the formula will be accepted even if the sheet is created later.
# Actually in xlsxwriter, all sheets are written to the workbook buffer;
# the order in the file is the order of add_worksheet() calls.
# The formula reference to '明細' works as long as the sheet exists by the time
# the workbook closes. We'll add 個案查詢 before 明細.
# Define the detail fields and their source column indices in 明細
detail_fields = [
("id", "#"),
("submitted_at", "時間戳記"),
("date", "事件發生日期"),
("time", "事件發生時間"),
("department", "部門名稱"),
("employee_name", "電子郵件地址"),
("severity", "緊急程度"),
("location", "涉及部門"),
("patient_or_staff", "涉及的患者或相關人員"),
("long_description", "事件描述"),
("incident_photos", "事件相關照片"),
("description", "當日事件摘要"),
("action_taken", "處理情況 / 已採取的行動"),
("needs_escalation", "需要管理層介入"),
("incident_score", "分數"),
("injured_person", "受傷員工"),
("medical_report", "醫療報告"),
]
# ============ Sheet 2: 個案查詢 ============
detail = wb.add_worksheet("個案查詢")
detail.set_tab_color("#F59E0B")
detail.hide_gridlines(2)
detail.set_column("A:A", 2)
detail.set_column("B:B", 24) # field labels
detail.set_column("C:C", 50) # values
detail.write("B2", "個案查詢 Case Lookup", title_fmt)
detail.write("B3", "輸入個案 ID 即時顯示詳細資料 (Excel VLOOKUP 自動)", subtitle_fmt)
# Input row
input_label_fmt = wb.add_format({"bold": True, "bg_color": "#FEF3C7",
"border": 2, "border_color": "#F59E0B",
"align": "left", "indent": 1})
input_fmt = wb.add_format({"bold": True, "font_size": 14, "font_color": "#1E3A8A",
"bg_color": "#FFFBEB", "border": 2, "border_color": "#F59E0B",
"align": "center"})
# Text-type variant for cells that need string input (like case_no id)
text_input_fmt = wb.add_format({"bold": True, "font_size": 14, "font_color": "#1E3A8A",
"bg_color": "#FFFBEB", "border": 2, "border_color": "#F59E0B",
"align": "center", "num_format": "@"})
# B5 = label only (NOT merged with C5). Otherwise user-typed text goes to
# B5 (top-left of merge) instead of C5 where the formula references.
detail.write("B5", "輸入個案 ID:", input_label_fmt)
# C5 = case_no input cell (NOT merged)
detail.write_string("C5", "", text_input_fmt)
# Hint
detail.merge_range("B6:C6",
"提示:個案 ID 喺「總覽」個案索引或「明細」表嘅 id 欄。輸入後下面自動顯示。",
subtitle_fmt)
# Detail rows: each pulls from 明細! via VLOOKUP keyed on id (col A)
detail_row_start = 8
for i, (col_key, label) in enumerate(detail_fields):
row_n = detail_row_start + i
lf = detail_label_fmt if i % 2 == 0 else detail_label_alt_fmt
detail.write(row_n, 1, label, lf)
col_idx = col_index_lookup.get(col_key, 1)
# Use INDEX/MATCH for type-safe text lookup. C5 is text (id column is text).
formula = f'=IFERROR(INDEX(明細!$A:$O,MATCH($C$5,明細!$A:$A,0),{col_idx}),"(找不到個案)")'
detail.write_formula(row_n, 2, formula, detail_value_fmt)
detail.set_row(row_n, 22)
# Multi-case list (rows under detail block)
list_start_row = detail_row_start + len(detail_fields) + 2
detail.write(f"B{list_start_row}", "多個個案列表 (輸入 ID, 顯示詳情)", section_fmt)
list_header_row = list_start_row + 1
detail.write(list_header_row, 1, "個案 ID", header_fmt)
detail.write(list_header_row, 2, "日期", header_fmt)
detail.write(list_header_row, 3, "時間", header_fmt)
detail.write(list_header_row, 4, "部門", header_fmt)
detail.write(list_header_row, 5, "嚴重程度", header_fmt)
detail.write(list_header_row, 6, "員工/涉及人", header_fmt)
detail.write(list_header_row, 7, "事件摘要", header_fmt)
detail.set_column("D:H", 18)
n_list_rows = 10
for j in range(n_list_rows):
row_n = list_header_row + 1 + j
detail.write_string(row_n, 1, "", text_input_fmt) # text input cell for case_no
col_date = col_index_lookup.get("date", 14)
col_time = col_index_lookup.get("time", 7)
col_dept = col_index_lookup.get("department", 4)
col_sev = col_index_lookup.get("severity", 6)
col_emp = col_index_lookup.get("employee_name", 3)
col_desc = col_index_lookup.get("description", 5)
# Use INDIRECT or simply use cell reference for ID - but ID varies per row.
# Easier: each row's ID is in B; build formula with that.
# VLOOKUP needs lookup_value, table, col_index, FALSE.
# table 明細!$A:$O.
# For severity col we want to apply conditional formatting so we use formula without format.
# xlsxwriter 0-indexed; Excel cell row = row_n + 1
# detail.write_string(row_n, 1, "") writes to Excel B
cell_id_ref = f"$B${row_n + 1}"
# INDEX/MATCH for text id lookup (consistent with detail block)
detail.write_formula(row_n, 2,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({cell_id_ref},明細!$A:$A,0),{col_date}),"")',
cell_fmt)
detail.write_formula(row_n, 3,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({cell_id_ref},明細!$A:$A,0),{col_time}),"")',
cell_fmt)
detail.write_formula(row_n, 4,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({cell_id_ref},明細!$A:$A,0),{col_dept}),"")',
cell_fmt)
detail.write_formula(row_n, 5,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({cell_id_ref},明細!$A:$A,0),{col_sev}),"")',
severity_fmts["default"])
detail.write_formula(row_n, 6,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({cell_id_ref},明細!$A:$A,0),{col_emp}),"")',
cell_fmt)
detail.write_formula(row_n, 7,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({cell_id_ref},明細!$A:$A,0),{col_desc}),"")',
cell_wrap_fmt)
detail.set_row(row_n, 28)
# Conditional formatting on the severity column in the list (rows list_header_row+1 ..)
list_sev_first = list_header_row + 1
list_sev_last = list_header_row + n_list_rows
for code, fmt in severity_fmts.items():
if code == "default":
continue
# Match cells in column F (=index 5) that equal "1", "2", etc.
detail.conditional_format(list_sev_first, 5, list_sev_last, 5, {
"type": "cell",
"criteria": "equal to",
"value": f'"{code}"',
"format": severity_fmts[code],
})
# ============ Side-by-side compare (2 cases) ============
compare_start_row = list_header_row + n_list_rows + 3
detail.write(f"B{compare_start_row}", "兩個個案 並列比較 Side-by-Side", section_fmt)
detail.write(f"B{compare_start_row + 1}",
"輸入兩個個案 ID (左個案 A, 右個案 B),對比每個欄位。空白表示無資料。",
subtitle_fmt)
cmp_label_row = compare_start_row + 2
detail.write(cmp_label_row, 1, "個案 A ID:", input_label_fmt)
detail.write_string(cmp_label_row, 2, "", text_input_fmt)
detail.write(cmp_label_row, 4, "個案 B ID:", input_label_fmt)
detail.write_string(cmp_label_row, 5, "", text_input_fmt)
detail.set_column("F:F", 50)
cmp_header_row = cmp_label_row + 2
detail.write(cmp_header_row, 1, "欄位", header_fmt)
detail.write(cmp_header_row, 2, "個案 A 值", header_fmt)
detail.write(cmp_header_row, 3, "", header_fmt)
detail.write(cmp_header_row, 4, "", header_fmt)
detail.write(cmp_header_row, 5, "個案 B 值", header_fmt)
detail.set_column("D:D", 5)
detail.set_column("E:E", 5)
cmp_data_start = cmp_header_row + 1
for i, (col_key, label) in enumerate(detail_fields):
row_n = cmp_data_start + i
lf = detail_label_fmt if i % 2 == 0 else detail_label_alt_fmt
detail.write(row_n, 1, label, lf)
col_idx = col_index_lookup.get(col_key, 1)
# Case A: lookup at C. Use INDEX/MATCH for text id.
id_a = f"$C${cmp_label_row + 1}"
id_b = f"$F${cmp_label_row + 1}"
detail.write_formula(row_n, 2,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({id_a},明細!$A:$A,0),{col_idx}),"")',
detail_value_fmt)
detail.write_formula(row_n, 5,
f'=IFERROR(INDEX(明細!$A:$O,MATCH({id_b},明細!$A:$A,0),{col_idx}),"")',
detail_value_fmt)
detail.set_row(row_n, 22)
# Hint at top
detail.merge_range(f"B{cmp_data_start + len(detail_fields) + 2}:C{cmp_data_start + len(detail_fields) + 2}",
"💡 全部公式 (VLOOKUP),無需啟用巨集。Excel/Numbers 都正常運作。",
subtitle_fmt)
# ============ Sheet 3: 明細 ============
ws = wb.add_worksheet("明細")
ws.set_tab_color("#10B981")
ws.set_column("A:A", 8)
widths_by_col = {
"id": 8, "submitted_at": 18, "employee_name": 24, "department": 14,
"description": 38, "severity": 10, "time": 10, "location": 18,
"patient_or_staff": 28, "long_description": 60, "incident_photos": 30,
"action_taken": 38, "needs_escalation": 10, "incident_score": 8, "date": 12,
}
sev_col = None
for col_idx, col in enumerate(column_order):
header_text = (excel_headers.get(col) or col).strip()
ws.write(0, col_idx, header_text, header_fmt)
ws.set_column(col_idx, col_idx, widths_by_col.get(col, 16))
ws.set_row(0, 24)
if col == "severity":
sev_col = col_idx
ws.freeze_panes(1, 0)
for r_idx, r in enumerate(rows, start=1):
# Compose incident_datetime from date + time
incident_dt = ""
if r.date:
date_part = r.date.isoformat()
time_part = r.time or "00:00:00"
if len(time_part) == 5:
time_part += ":00"
incident_dt = f"{date_part} {time_part}"
rec = {
"id": r.id,
"submitted_at": r.submitted_at.isoformat() if r.submitted_at else "",
"date": r.date.isoformat() if r.date else "",
"time": r.time or "",
"incident_datetime": incident_dt,
"location": r.location or "",
"submitter_email": r.submitter_email or "",
"injured_person_id": r.injured_person_id or "",
"department": r.department or "",
"description": r.description or "",
"severity": str(r.severity or ""),
"medical_report": r.medical_report or "",
"action_taken": r.action_taken or "",
"injured_person": r.injured_person or "",
"patient_or_staff": r.patient_or_staff or "",
"long_description": r.long_description or "",
"incident_photos": r.incident_photos or "",
"needs_escalation": r.needs_escalation or "",
"incident_score": r.incident_score if r.incident_score is not None else "",
"excel_row": r.excel_row if r.excel_row is not None else "",
}
ws.set_row(r_idx, 30)
for c_idx, col in enumerate(column_order):
v = rec.get(col, "")
if col == "severity":
# Apply plain format; conditional formatting will color it later
ws.write(r_idx, c_idx, v, severity_fmts["default"])
elif col in ("description", "long_description", "action_taken"):
ws.write(r_idx, c_idx, v, cell_wrap_fmt)
elif col == "id":
# IMPORTANT: Write id as TEXT so VLOOKUP matches text input from cell C5
ws.write_string(r_idx, c_idx, str(v), cell_fmt)
else:
ws.write(r_idx, c_idx, v, cell_fmt)
# Apply conditional formatting on severity column (after writing data)
if sev_col is not None and rows:
first_row = 1
last_row = len(rows)
for code, fmt in severity_fmts.items():
if code == "default":
continue
ws.conditional_format(first_row, sev_col, last_row, sev_col, {
"type": "cell",
"criteria": "equal to",
"value": f'"{code}"',
"format": fmt,
})
# Also add a filter (Excel auto-filter) on the header row for easy column filtering.
ws.autofilter(0, 0, len(rows), len(column_order) - 1)
wb.close()
buf.seek(0)
date_suffix = ""
if date_from and date_to:
date_suffix = f"_{date_from}_{date_to}"
elif date_from:
date_suffix = f"_from_{date_from}"
elif date_to:
date_suffix = f"_to_{date_to}"
filename = f"accident{date_suffix}.xlsx"
logger.info("accident export to xlsx v3",
extra={"context": {"user_id": current_user.id, "rows": len(rows),
"date_from": date_from, "date_to": date_to}})
from fastapi.responses import StreamingResponse
return StreamingResponse(
buf,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": f"attachment; filename={filename}"},
)
# ============ 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),
date_from: Optional[str] = Query(None),
date_to: Optional[str] = Query(None),
db: Session = Depends(get_db)
):
"""Export Excel file.
For attendance: builds xlsx on-the-fly from attendance_records SQL table with
full computed columns (expected_in, expected_out, actual_in, actual_out,
status_code, status_text, late/early/ot minutes).
For accident: returns the stored Excel file (legacy).
"""
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from database import SECRET_KEY, ALGORITHM
from datetime import datetime as _dt
credentials_exception = HTTPException(
status_code=401,
detail="Invalid or expired token",
headers={"WWW-Authenticate": "Bearer"},
)
current_user = None
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 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")
# Attendance: build xlsx from SQL table
if section == "attendance":
import xlsxwriter
import io
q = db.query(AttendanceRecord)
if date_from:
try:
q = q.filter(AttendanceRecord.date >= _dt.strptime(date_from, "%Y-%m-%d").date())
except ValueError:
pass
if date_to:
try:
q = q.filter(AttendanceRecord.date <= _dt.strptime(date_to, "%Y-%m-%d").date())
except ValueError:
pass
rows = q.order_by(AttendanceRecord.date.desc(), AttendanceRecord.employee_name.asc()).all()
enrich_attendance_with_leave(rows, db)
buf = io.BytesIO()
wb = xlsxwriter.Workbook(
buf,
{"in_memory": True,
"calc_on_load": True,
"calc_mode": "auto"})
wb.set_calc_mode("auto", calc_id=1)
# ============ Sheet 1: 總覽 (Dashboard) ============
overview = wb.add_worksheet("總覽")
overview.set_tab_color("#1E40AF")
overview.hide_gridlines(2)
overview.set_column("A:A", 2)
overview.set_column("B:B", 28)
overview.set_column("C:C", 18)
overview.set_column("D:D", 30)
overview.set_column("E:E", 18)
overview.set_column("F:F", 26)
overview.set_column("G:G", 18)
title_fmt = wb.add_format({"bold": True, "font_size": 16, "font_color": "#1E3A8A"})
subtitle_fmt = wb.add_format({"italic": True, "font_size": 10, "font_color": "#6B7280"})
section_fmt = wb.add_format({"bold": True, "font_size": 12, "font_color": "#1E3A8A",
"bottom": 2, "bottom_color": "#93C5FD"})
kpi_label_fmt = wb.add_format({"bold": True, "font_size": 9, "font_color": "#6B7280",
"align": "left"})
kpi_value_fmt = wb.add_format({"bold": True, "font_size": 22, "font_color": "#1E3A8A",
"align": "left"})
header_fmt_ov = wb.add_format({"font_color": "#FFFFFF", "bold": True, "align": "center",
"bg_color": "#1E40AF", "border": 1, "border_color": "#1E3A8A"})
cell_fmt_ov = wb.add_format({"align": "left", "valign": "top"})
link_fmt_ov = wb.add_format({"font_color": "#2563EB", "underline": 1})
overview.write("B2", "出勤記錄 總覽 Dashboard", title_fmt)
overview.write("B3", f"生成時間: {_dt.now().strftime('%Y-%m-%d %H:%M:%S')}", subtitle_fmt)
if date_from or date_to:
range_str = f"{date_from or '起始'} → {date_to or '現在'}"
else:
range_str = "全部記錄"
overview.write("B4", f"篩選範圍: {range_str}", subtitle_fmt)
# Counters
n_normal = sum(1 for r in rows if (r.status_code or "").lower() == "normal")
n_late = sum(1 for r in rows if "late" in (r.status_code or "").lower() and "early" not in (r.status_code or "").lower())
n_early = sum(1 for r in rows if "early" in (r.status_code or "").lower() and "late" not in (r.status_code or "").lower())
n_ot = sum(1 for r in rows if "ot" in (r.status_code or "").lower())
n_abnormal = sum(1 for r in rows if (r.status_code or "").lower() == "abnormal")
total_late_min = sum(r.late_minutes or 0 for r in rows)
total_ot_min = sum(r.ot_minutes or 0 for r in rows)
overview.merge_range("B6:C6", "總記錄數", kpi_label_fmt)
overview.merge_range("B7:C7", len(rows), kpi_value_fmt)
overview.merge_range("D6:E6", "正常出勤", kpi_label_fmt)
overview.merge_range("D7:E7", n_normal, kpi_value_fmt)
overview.merge_range("F6:G6", "最早事件日期", kpi_label_fmt)
earliest = min((r.date for r in rows if r.date), default=None)
overview.merge_range("F7:G7", earliest.isoformat() if earliest else "-", kpi_value_fmt)
# Status distribution table - simplified 6-group (B10:D)
overview.write("B10", "出勤狀態分佈", section_fmt)
overview.write("B11", "Status", header_fmt_ov)
overview.write("C11", "數量", header_fmt_ov)
overview.write("D11", "百分比", header_fmt_ov)
# Simplified 6-group status
_grp_status = {
"正常": 0, "遲到": 0, "早退": 0, "加班": 0, "異常/缺勤": 0, "假期/請假": 0, "(空)": 0,
}
for r in rows:
c = (r.status_code or "").lower()
if not c.strip():
_grp_status["(空)"] += 1
elif c == "normal":
_grp_status["正常"] += 1
elif "late" in c and "early" in c:
_grp_status["遲到"] += 1
elif "late" in c:
_grp_status["遲到"] += 1
elif "early" in c:
_grp_status["早退"] += 1
elif "ot" in c:
_grp_status["加班"] += 1
elif c in ("abnormal", "missing", "", " "):
_grp_status["異常/缺勤"] += 1
else:
_grp_status["假期/請假"] += 1
_total = len(rows)
for i, (label, count) in enumerate(_grp_status.items()):
overview.write(11 + i, 1, label, cell_fmt_ov)
overview.write(11 + i, 2, count, cell_fmt_ov)
pct = (count / _total * 100) if _total else 0
overview.write(11 + i, 3, f"{pct:.1f}%", cell_fmt_ov)
_n_status_groups = len(_grp_status)
# Also update KPI counters to match simplified grouping
n_late_grp = _grp_status["遲到"]
n_early_grp = _grp_status["早退"]
n_ot_grp = _grp_status["加班"]
n_abn_grp = _grp_status["異常/缺勤"]
# Department TOP 10 table (F10:G)
overview.write("F10", "部門記錄數 (TOP 10)", section_fmt)
overview.write("F11", "部門", header_fmt_ov)
overview.write("G11", "數量", header_fmt_ov)
from collections import Counter
dept_counts = Counter((r.department or "(未分類)") for r in rows)
top_depts = dept_counts.most_common(10)
for i, (dept, count) in enumerate(top_depts):
overview.write(11 + i, 5, dept, cell_fmt_ov)
overview.write(11 + i, 6, count, cell_fmt_ov)
# ============ Charts block (Sheet 1 總覽 右側 columns I-P) ============
# Chart 1: Status Pie (I6)
status_pie = wb.add_chart({"type": "pie"})
status_pie.add_series({
"name": "出勤狀態分佈",
"categories": ["總覽", 11, 1, 11 + _n_status_groups, 1],
"values": ["總覽", 11, 2, 11 + _n_status_groups, 2],
"data_labels": {"percentage": True, "category": False, "position": "outside_end"},
})
status_pie.set_title({"name": "出勤狀態分佈 (Pie)"})
status_pie.set_style(10)
status_pie.set_size({"width": 480, "height": 320})
status_pie.set_legend({"position": "right"})
overview.insert_chart("I6", status_pie, {"x_offset": 5, "y_offset": 5})
# Chart 2: Department Bar TOP 10 (I22)
dept_bar = wb.add_chart({"type": "bar"})
dept_bar.add_series({
"name": "部門記錄數",
"categories": ["總覽", 11, 5, 20, 5],
"values": ["總覽", 11, 6, 20, 6],
"fill": {"color": "#3B82F6"},
"border": {"color": "#1E40AF"},
"data_labels": {"value": True},
})
dept_bar.set_title({"name": "部門記錄數 (TOP 10)"})
dept_bar.set_x_axis({"name": "記錄數"})
dept_bar.set_y_axis({"name": "部門", "reverse": True})
dept_bar.set_legend({"none": True})
dept_bar.set_size({"width": 480, "height": 320})
overview.insert_chart("I22", dept_bar, {"x_offset": 5, "y_offset": 5})
# Chart 3: Status Column (I42)
status_col_chart = wb.add_chart({"type": "column"})
status_col_chart.add_series({
"name": "出勤狀態 數量",
"categories": ["總覽", 11, 1, 11 + _n_status_groups - 1, 1],
"values": ["總覽", 11, 2, 11 + _n_status_groups - 1, 2],
"fill": {"color": "#10B981"},
"border": {"color": "#065F46"},
"data_labels": {"value": True},
})
status_col_chart.set_title({"name": "出勤狀態 數量 (Column)"})
status_col_chart.set_x_axis({"name": "狀態"})
status_col_chart.set_y_axis({"name": "記錄數"})
status_col_chart.set_legend({"none": True})
status_col_chart.set_size({"width": 480, "height": 320})
overview.insert_chart("I42", status_col_chart, {"x_offset": 5, "y_offset": 5})
# Chart 4: Monthly Trend Line (I62)
trend_start_row = 60
overview.write(trend_start_row, 1, "月份", header_fmt_ov)
overview.write(trend_start_row, 2, "記錄數", header_fmt_ov)
from collections import OrderedDict
today = _dt.now().date()
months = []
for offset in range(11, -1, -1):
y = today.year
m = today.month - offset
while m <= 0:
m += 12
y -= 1
months.append(f"{y}-{m:02d}")
monthly_counts = OrderedDict((m, 0) for m in months)
for r in rows:
if r.date:
key = f"{r.date.year}-{r.date.month:02d}"
if key in monthly_counts:
monthly_counts[key] += 1
for i, (m, cnt) in enumerate(monthly_counts.items()):
overview.write(trend_start_row + 1 + i, 1, m, cell_fmt_ov)
overview.write(trend_start_row + 1 + i, 2, cnt, cell_fmt_ov)
n_months = len(months)
trend_end_row = trend_start_row + n_months
line_chart = wb.add_chart({"type": "line"})
line_chart.add_series({
"name": "每月記錄數",
"categories": ["總覽", trend_start_row + 1, 1, trend_end_row, 1],
"values": ["總覽", trend_start_row + 1, 2, trend_end_row, 2],
"line": {"color": "#10B981", "width": 2.25},
"marker": {"type": "circle", "size": 6,
"fill": {"color": "#10B981"},
"border": {"color": "#065F46"}},
"data_labels": {"value": True},
})
line_chart.set_title({"name": "每月記錄趨勢 (Last 12 months)"})
line_chart.set_x_axis({"name": "月份"})
line_chart.set_y_axis({"name": "記錄數"})
line_chart.set_legend({"none": True})
line_chart.set_size({"width": 480, "height": 320})
overview.insert_chart("I62", line_chart, {"x_offset": 5, "y_offset": 5})
# 員工索引 (row 80+)
nav_start_row = 80
overview.write(f"B{nav_start_row}", "員工出勤索引 (可 click → 跳去「明細」對應 row)", section_fmt)
overview.write(f"B{nav_start_row + 1}", "姓名", header_fmt_ov)
overview.write(f"C{nav_start_row + 1}", "部門", header_fmt_ov)
overview.write(f"D{nav_start_row + 1}", "日期", header_fmt_ov)
overview.write(f"E{nav_start_row + 1}", "狀態", header_fmt_ov)
overview.write(f"F{nav_start_row + 1}", "遲到 min", header_fmt_ov)
overview.write(f"G{nav_start_row + 1}", "→ 查明細", header_fmt_ov)
for idx, r in enumerate(rows):
row_n = nav_start_row + 2 + idx
overview.write(row_n, 1, r.employee_name or "", cell_fmt_ov)
overview.write(row_n, 2, r.department or "", cell_fmt_ov)
overview.write(row_n, 3, r.date.isoformat() if r.date else "", cell_fmt_ov)
overview.write(row_n, 4, r.status_text or "", cell_fmt_ov)
overview.write(row_n, 5, r.late_minutes or 0, cell_fmt_ov)
detail_row = idx + 2
overview.write_url(row_n, 6,
f"internal:明細!A{detail_row}",
link_fmt_ov,
string=f"#{idx + 1} → 查詢",
tip=f"跳到「明細」row {detail_row}")
overview.set_row(row_n, 18)
# ============ Sheet 2: 明細 (renamed from "Attendance") ============
ws = wb.add_worksheet("明細")
ws.set_tab_color("#10B981")
# Per-status formats (font color + bg color)
status_fmts = {
"normal": wb.add_format({"font_color": "#065F46", "bg_color": "#D1FAE5", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"late": wb.add_format({"font_color": "#92400E", "bg_color": "#FEF3C7", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"late_early": wb.add_format({"font_color": "#92400E", "bg_color": "#FED7AA", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"late_ot": wb.add_format({"font_color": "#7C2D12", "bg_color": "#FED7AA", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"early": wb.add_format({"font_color": "#1E40AF", "bg_color": "#DBEAFE", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"early_ot": wb.add_format({"font_color": "#1E40AF", "bg_color": "#E0E7FF", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"ot": wb.add_format({"font_color": "#5B21B6", "bg_color": "#EDE9FE", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"abnormal": wb.add_format({"font_color": "#7F1D1D", "bg_color": "#FEE2E2", "bold": True, "align": "center", "border": 1, "border_color": "#E5E7EB"}),
"missing": wb.add_format({"font_color": "#374151", "bg_color": "#E5E7EB", "bold": True, "align": "center", "border": 1, "border_color": "#D1D5DB"}),
"default": wb.add_format({"align": "center"}),
}
# Header format
header_fmt = wb.add_format({
"font_color": "#1E3A8A",
"bold": True,
"align": "center",
"bg_color": "#DBEAFE",
"border": 1,
"border_color": "#93C5FD",
})
# Column widths
widths = [
("Staff Name", 18), ("Company", 12), ("Department", 14),
("Date", 12), ("Weekday", 10), ("Shift", 10),
("Expected In", 12), ("Expected Out", 12),
("Actual In", 12), ("Actual Out", 12),
("Status Code", 12), ("Status Text", 22),
("Late Min", 10), ("Early Min", 10), ("OT Min", 10),
("Manual Edit", 12),
]
# Write headers
for col, (label, w) in enumerate(widths):
ws.write(0, col, label, header_fmt)
ws.set_column(col, col, w)
ws.freeze_panes(1, 0) # Freeze header row
# Write data
for r_idx, r in enumerate(rows, start=1):
sc = (r.status_code or "").lower()
fmt = status_fmts.get(sc, status_fmts["default"])
ws.write(r_idx, 0, r.employee_name or "")
ws.write(r_idx, 1, r.company or "")
ws.write(r_idx, 2, r.department or "")
ws.write(r_idx, 3, r.date.isoformat() if r.date else "")
ws.write(r_idx, 4, r.weekday or "")
ws.write(r_idx, 5, r.shift_code or "")
ws.write(r_idx, 6, r.expected_in or "-")
ws.write(r_idx, 7, r.expected_out or "-")
ws.write(r_idx, 8, r.actual_in or "-")
ws.write(r_idx, 9, r.actual_out or "-")
ws.write(r_idx, 10, sc, fmt)
ws.write(r_idx, 11, r.status_text or "")
ws.write(r_idx, 12, r.late_minutes or 0)
ws.write(r_idx, 13, r.early_minutes or 0)
ws.write(r_idx, 14, r.ot_minutes or 0)
ws.write(r_idx, 15, "Yes" if r.is_manually_edited else "")
ws.set_row(r_idx, 20)
wb.close()
buf.seek(0)
date_suffix = ""
if date_from and date_to:
date_suffix = f"_{date_from}_{date_to}"
elif date_from:
date_suffix = f"_from_{date_from}"
elif date_to:
date_suffix = f"_to_{date_to}"
filename = f"attendance{date_suffix}.xlsx"
logger.info("attendance export to xlsx",
extra={"context": {"user_id": current_user.id, "rows": len(rows),
"date_from": date_from, "date_to": date_to}})
from fastapi.responses import StreamingResponse
return StreamingResponse(
buf,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": f"attachment; filename={filename}"},
)
# Accident: legacy file
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"},
)
# ============ Roster / Shift Management ============
from fastapi.staticfiles import StaticFiles
import os as _os
# Serve built frontend (Vite output)
if _os.path.isdir("/app/static"):
app.mount("/assets", StaticFiles(directory="/app/static/assets"), name="assets")
@app.get("/")
async def root():
return RedirectResponse(url="/index.html")
@app.get("/index.html")
async def index_html():
"""Serve the built React app index.html"""
path = "/app/static/index.html"
if not _os.path.exists(path):
raise HTTPException(status_code=404, detail="Frontend not built")
return FileResponse(path, media_type="text/html")
@app.get("/favicon.ico")
async def favicon():
p = "/app/static/favicon.ico"
if _os.path.exists(p):
return FileResponse(p)
raise HTTPException(status_code=404)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
# ============ Client Logs (frontend error reports) ============
class ClientLogEntry(BaseModel):
level: str = Field(..., pattern="^(debug|info|warn|error)$")
msg: str = Field(..., max_length=2000)
stack: Optional[str] = Field(None, max_length=8000)
page_url: Optional[str] = Field(None, max_length=2000)
app_version: Optional[str] = Field(None, max_length=64)
request_id: Optional[str] = Field(None, max_length=128)
browser: Optional[str] = Field(None, max_length=500)
ts: Optional[str] = Field(None, max_length=64)
@app.post("/api/client-logs")
async def client_logs(entry: ClientLogEntry, request: Request):
"""Receive error reports from frontend. No auth required."""
level_map = {
"debug": logging.DEBUG,
"info": logging.INFO,
"warn": logging.WARNING,
"error": logging.ERROR,
}
log_level = level_map.get(entry.level, logging.INFO)
# Use server's request_id as authoritative if client sent one that looks fishy
server_rid = getattr(request.state, "request_id", None)
final_rid = entry.request_id or server_rid or "-"
extra = {
"route": entry.page_url or "-",
"page_url": entry.page_url,
"app_version": entry.app_version,
"browser": entry.browser,
"client_ip": request.client.host if request.client else None,
"context": {
"source": "client",
"request_id_from_client": entry.request_id,
"server_request_id": server_rid,
"ts_from_client": entry.ts,
},
}
if entry.stack:
# Log as if exception
logger.log(log_level, "client error: %s", entry.msg, extra={
**extra,
"context": {
**extra["context"],
"error": {"name": "ClientError", "message": entry.msg, "stack": entry.stack},
},
})
else:
logger.log(log_level, "client: %s", entry.msg, extra=extra)
return {"status": "ok", "request_id": server_rid}